在Flask-SQLAlchemy中隔离py.test数据库会话

Question

我正在尝试使用Flask-SQLAlchemy构建Flask应用程序; 我使用pytest来测试数据库.其中一个问题似乎是在不同测试之间创建隔离的数据库会话.

我做了一个最小的,完整的例子来突出问题,请注意test_user_schema1()并且test_user_schema2()是相同的.

文件名: test_db.py

from models import User

def test_user_schema1(session):
    person_name = 'Fran Clan'
    uu = User(name=person_name)
    session.add(uu)
    session.commit()

    assert uu.id==1
    assert uu.name==person_name

def test_user_schema2(session):
    person_name = 'Stan Clan'
    uu = User(name=person_name)
    session.add(uu)
    session.commit()

    assert uu.id==1
    assert uu.name==person_name

如果db在我的测试之间真正隔离,那么两个测试都应该通过.但是,最后一次测试总是失败,因为我还没有找到使数据库会话正确回滚的方法.

conftest.py根据我在Alex Michael的博客文章中看到的内容使用以下内容,但是这个夹具代码中断了,因为它显然没有隔离夹具之间的db会话.

@pytest.yield_fixture(scope='function')
def session(app, db):
    connection = db.engine.connect()
    transaction = connection.begin()

    #options = dict(bind=connection, binds={})
    options = dict(bind=connection)
    session = db.create_scoped_session(options=options)

    yield session

    # Finalize test here
    transaction.rollback()
    connection.close()
    session.remove()

出于这个问题的目的,我构建了一个要点,其中包含了重现它所需的一切; 你可以克隆它git clone https://gist.github.com/34fa8d274fc4be240933.git.

我使用以下包裹......

Flask==0.10.1
Flask-Bootstrap==3.3.0.1
Flask-Migrate==1.3.0
Flask-Moment==0.4.0
Flask-RESTful==0.3.1
Flask-Script==2.0.5
Flask-SQLAlchemy==2.0
Flask-WTF==0.11
itsdangerous==0.24
pytest==2.6.4
Werkzeug==0.10.1

两个问题:

1. 为什么现状破裂？同样的py.test夹具似乎适用于其他人.
2. 如何解决此问题才能正常工作？

Answer 1

Alex Michael的博客文章中介绍的方法不起作用,因为它不完整.根据关于加入会话的sqlalchemy文档,Alex的解决方案仅在没有回滚调用时才有效.另一个区别是,Session与Alex的博客上的作用域会话相比,sqla文档中使用了vanilla 对象.

对于flask-sqlalchemy,在请求拆除时会自动删除作用域会话.发出呼叫session.remove,在发动机罩下发出回滚.要支持测试范围内的回滚,请使用SAVEPOINT:

import sqlalchemy as sa


@pytest.yield_fixture(scope='function')
def db_session(db):
    """
    Creates a new database session for a test. Note you must use this fixture
    if your test connects to db.

    Here we not only support commit calls but also rollback calls in tests.
    """
    connection = db.engine.connect()
    transaction = connection.begin()

    options = dict(bind=connection, binds={})
    session = db.create_scoped_session(options=options)

    session.begin_nested()

    # session is actually a scoped_session
    # for the `after_transaction_end` event, we need a session instance to
    # listen for, hence the `session()` call
    @sa.event.listens_for(session(), 'after_transaction_end')
    def resetart_savepoint(sess, trans):
        if trans.nested and not trans._parent.nested:
            session.expire_all()
            session.begin_nested()

    db.session = session

    yield session

    session.remove()
    transaction.rollback()
    connection.close()

您的数据库必须支持SAVEPOINT.

Answer 2

1.

根据Session Basics - SQLAlchemy文档:

commit()用于提交当前事务.它总是事先发出flush()以将任何剩余状态刷新到数据库; 这与"autoflush"设置无关.....

因此transaction.rollback()会话夹具功能不起作用,因为事务已经提交.

2.

更改灯具的范围,function而不是session每次都清除db.

@pytest.yield_fixture(scope='function')
def app(request):
    ...

@pytest.yield_fixture(scope='function')
def db(app, request):
    ...

顺便说一句,如果你使用内存中的sqlite数据库,你不需要删除db文件,它会更快:

DB_URI = 'sqlite://'  # SQLite :memory: database

...

@pytest.yield_fixture(scope='function')
def db(app, request):
    _db.app = app
    _db.create_all()
    yield _db
    _db.drop_all()