6

我正在尝试学习烧瓶技术堆栈,对于我的应用程序,我正在使用 Flask-SQLAlchemy。一切都很完美,但我正在努力编写集成测试。我不想使用 SQLite,因为在生产中我使用的是 PostgreSQL,并且放置大量模拟实际上会更多地测试我自己的实现而不是逻辑本身。

因此,经过一些研究,我决定实施将数据写入测试数据库的测试,并在每次测试后回滚更改(为了性能起见)。实际上,我正在尝试实现类似于这种方法的东西: http: //sontek.net/blog/detail/writing-tests-for-pyramid-and-sqlalchemy

我的问题是创建正确的事务并能够回滚它。这是我的基类的代码:

from flask.ext.sqlalchemy import SQLAlchemy
db = SQLAlchemy()

class MyAppIntegrationTestCase(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
    app.config['TESTING'] = True
    app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql+psycopg2:///db_test'
    init_app()
    db.app = app
    db.create_all(app=app)

    @classmethod
    def tearDownClass(cls):
        db.drop_all(app=app)

    def setUp(self):
        db.session.rollback()
        self.trans = db.session.begin(subtransactions=True)

    def tearDown(self):
        self.trans.rollback()

当我尝试执行测试时,出现以下错误:

Traceback (most recent call last):
 File "myapp/src/core/tests/__init__.py", line 53, in tearDown
   self.trans.rollback()
 File "myapp/venv/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 370, in rollback
   self._assert_active(prepared_ok=True, rollback_ok=True)
 File "myapp/venv/lib/python2.7/site-packages/sqlalchemy/orm/session.py", line 203, in _assert_active
   raise sa_exc.ResourceClosedError(closed_msg)
ResourceClosedError: This transaction is closed

我敢打赌这是 scoped_session 的问题,当我运行测试时,它会为所有测试重用一个全局会话,但我对 SQLAlchemy 的了解还不够深入。

任何帮助将不胜感激!谢谢!

4

3 回答 3

0

您是 tearDownClass 和 setUpClass 导致了问题。

setUpClass 在所有测试之前调用一次,而 tearDownClass 在类中的所有测试之后调用。

所以如果你有3个测试。

调用 setUpClass

设置被称为

调用了 tearDown (您回滚,但您没有开始会话,这会引发错误)

setUp 被调用(另一个会出错的回滚)

ETC...

将 db.session.begin 添加到您的 tearDown 中,您会没事的。

from flask.ext.sqlalchemy import SQLAlchemy
db = SQLAlchemy()

class MyAppIntegrationTestCase(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        app.config['TESTING'] = True
        app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql+psycopg2:///db_test'
        init_app()
        db.app = app
        db.create_all(app=app)

    @classmethod
    def tearDownClass(cls):
        db.drop_all(app=app)

    def setUp(self):
        db.session.rollback()
        self.trans = db.session.begin(subtransactions=True)

    def tearDown(self):
        self.trans.rollback()
        db.session.begin()
于 2015-07-23T03:34:42.240 回答
0

您的问题的可能解决方案:

如果您的数据库的数据量不是很大并且您希望保持数据不变,您可以在设置中进行备份(通过直接编写 sql 语句)

"CREATE TABLE {0}_backup SELECT * FROM {0}".format(table_name)

并在拆解中恢复

"DROP TABLE {0}".format(table_name)
"RENAME TABLE {0}_backup TO {0}".format(table_name)
于 2016-08-26T03:44:22.267 回答
0

我写了一篇关于如何设置的博客文章……简而言之,您必须创建一个嵌套事务,以便应用程序内的任何 session.commit() 调用都不会破坏您的隔离。然后将侦听器应用于内部事务以在有人尝试提交或回滚它时重新启动它。 设置 Flask-Sqlalchemy 事务测试用例

于 2016-08-26T02:31:46.717 回答