Flask + SQLAlchemy + pytest - 不回滚我的会话

Question

关于堆栈溢出有几个类似的问题，如果我通过询问另一个问题来破坏礼节，我提前道歉，但我似乎无法想出正确的咒语来完成这项工作。

我正在尝试使用 Flask + Flask-SQLAlchemy，然后使用 pytest 来管理会话，这样当函数范围的 pytest 固定装置被拆除时，当前的事务就会回滚。

其他一些问题似乎提倡在函数范围内使用 db “drop all and create all” pytest 固定装置，但我正在尝试使用加入的会话并使用回滚，因为我有很多测试.这将大大加快速度。

http://alexmic.net/flask-sqlalchemy-pytest/ 是我找到原始想法的地方，Isolating py.test DB sessions in Flask-SQLAlchemy 是建议使用函数级数据库重新创建的问题之一。

我也见过 https://github.com/mitsuhiko/flask-sqlalchemy/pull/249 ，但它似乎已与 flask-sqlalchemy 2.1（我正在使用）一起发布。

我当前的（非常小，希望可以立即理解）回购在这里： https://github.com/hoopes/flask-pytest-example

有两个打印语句 - 第一个（在 example/__init__.py 中）应该有一个 Account 对象，第二个（在 test/conftest.py 中）是我希望在事务完成后清除数据库的地方回滚。

如果您在测试目录中pip install -r requirements.txt 并运行py.test -s，您应该会看到两条打印语句。

我快走到尽头了 - 一定有什么我错过了，但就我的一生而言，我似乎无法找到它。

帮帮我，所以，你是我唯一的希望！

Answer 1

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from unittest import TestCase

# global application scope.  create Session class, engine
Session = sessionmaker()

engine = create_engine('postgresql://...')

class SomeTest(TestCase):
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()

        # begin a non-ORM transaction
        self.trans = self.connection.begin()

        # bind an individual Session to the connection
        self.session = Session(bind=self.connection)

    def test_something(self):
        # use the session in tests.

        self.session.add(Foo())
        self.session.commit()

    def tearDown(self):
        self.session.close()

        # rollback - everything that happened with the
        # Session above (including calls to commit())
        # is rolled back.
        self.trans.rollback()

        # return connection to the Engine
        self.connection.close()

sqlalchemy doc has solution for the case

Answer 2

您可能想试试pytest-flask-sqlalchemy-transactions。这是一个插件，它公开了一个db_session 固定装置，可以完成您正在寻找的东西：允许您运行数据库更新，当测试退出时这些更新将被回滚。该插件基于 Alex Michael 的博客文章，对涵盖更广泛用户案例的嵌套事务提供了一些额外支持。还有一些配置选项可用于在您的应用程序中模拟可连接，因此您也可以从代码库中运行任意方法。

对于test_accounts.py，你可以这样做：

from example import db, Account


class TestAccounts(object):

    def test_update_view(self, db_session):

        test_acct = Account(username='abc')

        db_session.add(test_acct)
        db_session.commit()

        resp = self.client.post('/update',
                                data={'a':1},
                                content_type='application/json')

        assert resp.status_code == 200

插件需要通过_db 固定装置访问您的数据库，但由于您已经在conftest.py 中定义了db 固定装置，您可以轻松设置数据库访问：

@pytest.fixture(scope='session')
def _db(db):
    return db

您可以在the docs 中找到有关如何设置和安装的详细信息。希望这会有所帮助！

Answer 3

我也遇到了回滚问题，可以找到我的代码here

阅读了一些文档后，似乎应该在会话中调用 begin() 函数。

所以在你的情况下，我会将会话夹具更新为：

@pytest.yield_fixture(scope='function', autouse=True)
def session(db, request):
    """Creates a new database session for a test."""

    db.session.begin()

    yield db.session

    db.session.rollback()
    db.session.remove()

我没有测试这个代码，但是当我在我的代码上尝试它时，我得到了以下错误：

INTERNALERROR> Traceback (most recent call last):
INTERNALERROR>   File "./venv/lib/python2.7/site-packages/_pytest/main.py", line 90, in wrap_session
INTERNALERROR>     session.exitstatus = doit(config, session) or 0
...
INTERNALERROR>   File "./venv/lib/python2.7/site-packages/_pytest/python.py", line 59, in filter_traceback
INTERNALERROR>     return entry.path != cutdir1 and not entry.path.relto(cutdir2)
INTERNALERROR> AttributeError: 'str' object has no attribute 'relto'