如何在单元测试中测试 SQLAlchemy 版本控制 - Python

How to test SQLAlchemy versioning in unit tests - 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()

对于遇到这个问题的任何人,我目前的假设是无法完成。 SQLAlchemy 非常强大,鉴于功能如此之好以至于我们无法测试这一行,我们应该相信它会按预期工作

实际上是可以的,您需要创建两个单独的会话。请参阅 the unit test of SQLAlchemy itself 以获取灵感。这是我们使用 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,这就是我们自己增加版本的原因。有关详细信息,请参阅 the docs