Flask 的 SQLAlchemy 没有回滚适当的事务？

Question

我有一系列简单的集成测试，我想在每次测试后发出回滚。我面临的问题是，虽然我看到控制台中发出了回滚语句（我的配置中有

SQLALCHEMY_ECHO = True

），但数据仍然存在于数据库中。显然，我希望每次测试后插入的数据都会从数据库中消失。

这是一个示例，其中的代码足以触发我的问题 -

import pytest
from app import create_app
from app import db as _db
from app.main.models import Customer


@pytest.yield_fixture(scope='session')
def app():
    app = create_app('testing')
    app_ctx = app.app_context()
    app_ctx.push()

    yield app

    app_ctx.pop()


@pytest.yield_fixture(scope='session')
def db(app):
    _db.app = app
    _db.create_all()

    yield _db

    print('drop_all()')


@pytest.yield_fixture(scope='function')
def session(db):
    conn = db.engine.connect()
    trans = conn.begin()
    session = db.create_scoped_session(options={ 'bind': conn })
    db.session = session

    yield session

    trans.rollback()
    conn.close()
    session.remove()


def test_foo(session, app):        
    cust = Customer(name='foo')
    session.add(cust)
    cust2 = Customer(name='bar')
    session.add(cust2)
    session.commit()

    client = app.test_client()
    response = client.get('/api/customers')

    from json import loads
    json_response = loads(response.data.decode('utf-8'))
    custs = json_response['customers']

    assert len(custs) == 2
    assert custs[0]['name'] == 'foo'

SQLAlchemy 日志 -

2015-09-24 13:56:27,601 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-09-24 13:56:27,667 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-09-24 13:56:27,669 INFO sqlalchemy.engine.base.Engine INSERT INTO customers (name) OUTPUT inserted.id VALUES (?)
2015-09-24 13:56:27,669 INFO sqlalchemy.engine.base.Engine ('foo')
2015-09-24 13:56:27,715 INFO sqlalchemy.engine.base.Engine INSERT INTO customers (name) OUTPUT inserted.id VALUES (?) 
2015-09-24 13:56:27,715 INFO sqlalchemy.engine.base.Engine ('bar')
2015-09-24 13:56:27,749 INFO sqlalchemy.engine.base.Engine COMMIT
2015-09-24 13:56:27,779 INFO sqlalchemy.engine.base.Engine BEGIN (implicit) 
2015-09-24 13:56:27,779 INFO sqlalchemy.engine.base.Engine SELECT customers.id AS customers_id, customers.name AS customers_name FROM customers 
2015-09-24 13:56:27,780 INFO sqlalchemy.engine.base.Engine ()
2015-09-24 13:56:27,827 INFO sqlalchemy.engine.base.Engine ROLLBACK
2015-09-24 13:56:27,827 INFO sqlalchemy.engine.base.Engine ROLLBACK

如您所见，有 3 个隐式事务正在启动，但只有 2 个回滚。为什么我在测试会话期间开始的事务没有被回滚？

我已经阅读了 SQLAlchemy 的有关测试期间回滚的文档，但没有任何内容。

Answer 1

固定装置中的肮脏黑客行为。对我有帮助

tmp_engine = db.engines[None]
connection = db.engine.connect()
transaction = connection.begin()
db.engines[None] = connection
yield db.session
db.session.remove()
transaction.rollback()
connection.close()
db.engines[None] = tmp_engine