如何将 py.test 固定装置与 Flask-SQLAlchemy 和 PostgreSQL 结合使用？

Question

我正在努力编写 py.test 固定装置来管理我的应用程序的数据库，以最大限度地提高速度，支持 pytest-xdist 并行化测试，并将测试相互隔离。

我正在对 PostgreSQL 9.4 数据库使用 Flask-SQLAlchemy 2.1。

这是我要完成的工作的大致轮廓：

1. $ py.test -n 3 启动三个测试会话以运行测试。

2. 在每个会话中，py.test 夹具运行一次以设置事务、创建数据库表，然后在会话结束时回滚事务。创建数据库表需要在仅对该特定测试会话可见的 PostgreSQL 事务中进行，否则pytest-xdist 创建的并行化测试会话会导致相互冲突。

3. 为每个测试运行的第二个 py.test 夹具连接到现有事务以查看创建的表，创建嵌套保存点，运行测试，然后回滚到嵌套保存点。

4. 理想情况下，这些 pytest 夹具支持调用 db.session.rollback() 的测试。在SQLAlchemy doc 的底部有一个潜在的方法来完成此操作。

5. 理想情况下，pytest 设备应该产生db 对象，而不仅仅是会话，这样 人们可以编写测试，而不必记住使用 不同于他们在整个应用中使用的标准db.session。

这是我目前所拥有的：

import pytest

# create_app() is my Flask application factory
# db is just 'db = SQLAlchemy()' + 'db.init_app(app)' within the create_app() function
from app import create_app, db as _db 


@pytest.yield_fixture(scope='session', autouse=True)
def app():
    '''Session-wide test application'''
    a = create_app('testing')
    with a.app_context():
        yield a

@pytest.yield_fixture(scope='session')
def db_tables(app):
    '''Session-wide test database'''
    connection = _db.engine.connect()
    trans = connection.begin() # begin a non-ORM transaction

    # Theoretically this creates the tables within the transaction
    _db.create_all()
    yield _db
    trans.rollback()
    connection.close()

@pytest.yield_fixture(scope='function')
def db(db_tables):
    '''db session that is joined to existing transaction'''

    # I am quite sure this is broken, but it's the general idea 

    # bind an individual Session to the existing transaction
    db_tables.session = db_tables.Session(bind=db_tables.connection)

    # start the session in a SAVEPOINT...
    db_tables.session.begin_nested()

    # yield the db object, not just the session so that tests
    # can be written transparently using the db object
    # without requiring someone to understand the intricacies of these
    # py.test fixtures or having to remember when to use a session that's
    # different than db.session
    yield db_tables

    # rollback to the savepoint before the test ran
    db_tables.session.rollback()
    db_tables.session.remove() # not sure this is needed

这是我在谷歌搜索时发现的最有用的参考资料：

http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites

http://koo.fi/blog/2015/10/22/flask-sqlalchemy-and-postgresql-unit-testing-with-transaction-savepoints/

https://github.com/mitsuhiko/flask-sqlalchemy/pull/249

Answer 1

我在这里晚了几年，但您可能对 pytest-flask-sqlalchemy 感兴趣，这是我为帮助解决这个确切问题而编写的插件。

该插件提供了两个夹具，db_session 和 db_engine，您可以像常规的 Session 和 Engine 对象一样使用它们来运行将在测试结束时回滚的更新。它还公开了一些配置指令（mocked-engines 和 mocked-sessions），它们将模拟您的应用程序中的可连接对象并将它们替换为这些固定装置，以便您可以运行方法并确保在测试时清除任何状态更改退出。

该插件应该适用于各种数据库，但它已针对 Postgres 9.6 进行了最严格的测试，并且在 https://dedupe.io 的测试套件中正在生产中。您可以在 the documentation 中找到一些可以帮助您入门的示例，但如果您愿意提供一些代码，我也很乐意演示如何使用该插件。

Answer 2

我在尝试组合 yield 固定装置时遇到了类似的问题。不幸的是，根据doc，您不能组合多个yield 级别。

但是您也许可以使用request.finalizer 找到解决方法：

@pytest.fixture(scope='session', autouse=True)
def app():
    '''Session-wide test application'''
    a = create_app('testing')
    with a.app_context():
        return a

@pytest.fixture(scope='session')
def db_tables(request, app):
    '''Session-wide test database'''
    connection = _db.engine.connect()
    trans = connection.begin() # begin a non-ORM transaction

    # Theoretically this creates the tables within the transaction
    _db.create_all()
    def close_db_session():
        trans.rollback()
        connection.close()
    request.addfinalizer(close_db_session)
    return _db