Как протестировать версии SQLAlchemy в модульных тестах — Python
Примечание. Здесь используется flask_sqlalchemy.
Я работаю над добавлением версий для нескольких служб в одной и той же БД. Чтобы убедиться, что это работает, я добавляю модульные тесты, которые подтверждают, что я получаю сообщение об ошибке (в этом случае моя ошибка должна быть StaleDataError). Для других сервисов на других языках я дважды извлек один и тот же объект из БД, обновил один экземпляр, сохранил его, обновил другой экземпляр, а затем попытался сохранить и его.
Однако, поскольку SQLAlchemy добавляет уровень фальшивого кэша между БД и службой, когда я обновляю первый объект, он автоматически обновляет другой объект, который я храню в памяти. У кого-нибудь есть способ обойти это? Я создал второй сеанс (это решение работало на других языках), но SQLAlchemy знает, что нельзя хранить один и тот же объект в двух сеансах.
Я смог вручную протестировать его, поместив time.sleep() на полпути теста и вручную изменив данные в БД, но я хотел бы проверить это, используя только код модуля.
Пример кода:
def test_optimistic_locking(self):
c = Customer(formal_name='John', id=1)
db.session.add(c)
db.session.flush()
cust = Customer.query.filter_by(id=1).first()
db.session.expire(cust)
same_cust = Customer.query.filter_by(id=1).first()
db.session.expire(same_cust)
same_cust.formal_name = 'Tim'
db.session.add(same_cust)
db.session.flush()
db.session.expire(same_cust)
cust.formal_name = 'Jon'
db.session.add(cust)
with self.assertRaises(StaleDataError): db.session.flush()
db.session.rollback()
Ответы 2
Для всех, кто сталкивается с этим вопросом, моя текущая гипотеза заключается в том, что это невозможно сделать. SQLAlchemy невероятно мощный, и, учитывая, что функциональность настолько хороша, что мы не можем протестировать эту строку, мы должны верить, что она работает так, как ожидалось.
На самом деле это возможно, вам нужно создать две отдельные сессии. См. модульный тест самого SQLAlchemy для вдохновения. Вот фрагмент кода одного из наших модульных тестов, написанных с помощью pytest:
def test_article__versioning(connection, db_session: Session):
article = ProductSheetFactory(title = "Old Title", version=1)
db_session.refresh(article)
assert article.version == 1
db_session2 = Session(bind=connection)
article2 = db_session2.query(ProductSheet).get(article.id)
assert article2.version == 1
article.title = "New Title"
article.version += 1
db_session.commit()
assert article.version == 2
with pytest.raises(sqlalchemy.orm.exc.StaleDataError):
article2.title = "Yet another title"
assert article2.version == 1
article2.version += 1
db_session2.commit()
Надеюсь, это поможет. Обратите внимание, что мы используем "version_id_generator": False в модели, поэтому мы сами увеличиваем версию. Подробности смотрите в документации.