【问题标题】:How do I properly set up a single SQLAlchemy session for each unit test?如何为每个单元测试正确设置一个 SQLAlchemy 会话?
【发布时间】:2019-01-28 02:58:27
【问题描述】:

在使用 WebTest 测试我的 Pyramid 应用程序时,我无法在测试中创建/使用单独的 Session,而不会收到有关已存在范围会话的警告。

这里是 Pyramid 应用程序的 main() 函数,也是配置数据库的地方。

# __init__.py of Pyramid application

from pyramid_sqlalchemy import init_sqlalchemy
from sqlalchemy import create_engine


def main(global_config, **settings):
    ...
    db_url = 'some-url'
    engine = create_engine(db_url)
    init_sqlalchemy(engine)  # Warning thrown here.

这是测试代码。

# test.py (Functional tests)

import transaction
from unittest import TestCase
from pyramid.paster import get_appsettings
from pyramid_sqlalchemy import init_sqlalchemy, Session
from sqlalchemy import create_engine
from webtest import TestApp

from app import main
from app.models.users import User


class BaseTestCase(TestCase):
    def base_set_up(self):
        # Create app using WebTest
        settings = get_appsettings('test.ini', name='main')
        app = main({}, **settings)
        self.test_app = TestApp(app)

        # Create session for tests.
        db_url = 'same-url-as-above'
        engine = create_engine(db_url)
        init_sqlalchemy(engine)
        # Note: I've tried both using pyramid_sqlalchemy approach here and 
        # creating a "plain, old" SQLAlchemy session here using sessionmaker.

    def base_tear_down(self):
        Session.remove()


class MyTests(BaseTestCase):
    def setUp(self):
        self.base_set_up()

        with transaction.manager:
            self.user = User('user@email.com', 'John', 'Smith')
            Session.add(self.user)
            Session.flush()

            Session.expunge_all()
        ...

    def tearDown(self):
        self.base_tear_down()

    def test_1(self):
        # This is a typical workflow on my tests.
        response = self.test_app.patch_json('/users/{0}'.format(self.user.id), {'email': 'new.email@email.com')
        self.assertEqual(response.status_code, 200)

        user = Session.query(User).filter_by(id=self.user.id).first()
        self.assertEqual(user.email, 'new.email@email.com')
    ...
    def test_8(self):
        ...

运行测试给了我 8 个通过,7 个警告,其中除了第一个测试之外的每个测试都给出以下警告:

来自 Pyramid 应用程序:__init__.py -> main -> init_sqlalchemy(engine): sqlalchemy.exc.SAWarning:至少一个作用域会话已经存在。 configure() 不能影响已经创建的会话。

如果这有任何用处,我相信我在这里看到了同样的问题,除了我使用的是 pyramid_sqlalchemy 而不是创建自己的 DBSession。

https://github.com/Pylons/webtest/issues/5

【问题讨论】:

  • 你能给我们你的测试代码之一发生警告
  • 更新了帖子以包含我发现的更多信息。
  • 我无法使用您链接的代码重现警告,您显然连续两次调用 Session.configure() 但这些调用并未出现在您向我们展示的内容中,并且有在 init_sqlalchemy 中没有这样的调用,我也不明白为什么你调用你的 main 来设置一个引擎和一个会话,然后在你的基本设置中设置第二个引擎到同一个数据库和一个会话,因为你只是覆盖了第一个,也许尝试删除主中的会话配置,或者只是不要在测试中调用它。
  • @bboumend 感谢您为此付出的努力。我相信要重现此问题,您需要在应用程序和测试中使用会话。我最初想设置单独的会话来测试对数据库的实际更改,因为会话中的状态可能与某些(意外)情况下实际提交的状态不同。但是,我采纳了您的建议,并正在重用同一会话,并根据需要使用 Session.expire_all() 清除状态。

标签: python-3.x sqlalchemy pyramid webtest


【解决方案1】:

回答我自己的问题:我不确定这是否是最好的方法,但对我有用。

我没有尝试在我的测试中创建单独的会话,而是使用在应用程序中配置的 pyramid_sqlalchemy 会话工厂。据我所知,在测试和应用程序代码中对 Session 的调用返回相同的注册 scoped_session。

我为测试创建单独会话的初衷是确认记录正在写入数据库,而不仅仅是在活动的 SQLAlchemy 会话中更新。使用这种新方法,我通过在测试事务和应用程序事务之间转换的测试点发出 Session.expire_all() 来设法避免这些“缓存”问题。

# test.py (Functional tests)

import transaction
from unittest import TestCase
from pyramid.paster import get_appsettings
from pyramid_sqlalchemy import Session
from webtest import TestApp

from app import main
from app.models.users import User


class BaseTestCase(TestCase):
    def base_set_up(self):
        # Create app using WebTest
        settings = get_appsettings('test.ini', name='main')
        app = main({}, **settings)
        self.test_app = TestApp(app)

        # Don't set up an additional session in the tests. Instead import
        # and use pyramid_sqlalchemy.Session, which is set up in the application.

    def base_tear_down(self):
        Session.remove()


class MyTests(BaseTestCase):
    def setUp(self):
        self.base_set_up()

        with transaction.manager:
            self.user = User('user@email.com', 'John', 'Smith')
            Session.add(self.user)
            Session.flush()

            Session.expunge_all()
            Session.expire_all()  # "Reset" your session.

    def tearDown(self):
        self.base_tear_down()

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 2011-02-02
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2011-02-06
    • 2022-11-04
    • 2019-12-20
    相关资源
    最近更新 更多