时间戳

首页 > 解决方案 > 夹具中的数据库更改似乎不会持续测试

python - 夹具中的数据库更改似乎不会持续测试

问题描述

我正在使用 sqlite db 编写一些 Pytest 代码，以测试一些逻辑。我设置了一个根级夹具来实例化一个数据库引擎：

class SqliteEngine:
    def __init__(self):
        self._conn_engine = create_engine("sqlite://")
        self._conn_engine.execute("pragma foreign_keys=ON")

    def get_engine(self):
        return self._conn_engine

    def get_session(self):
        Session = sessionmaker(bind=self._conn_engine, autoflush=True)
        return Session()


@pytest.fixture(scope="session")
def sqlite_engine():
    sqlite_engine = SqliteEngine()
    return sqlite_engine

然后在我的测试课上，我有

class TestRbac:
    @pytest.fixture(scope="class")
    def setup_rbac_tables(self, sqlite_engine):
        conn_engine = sqlite_engine.get_engine()
        conn_engine.execute("attach ':memory:' as rbac")
        Application.__table__.create(conn_engine)
        Client.__table__.create(conn_engine)
        Role.__table__.create(conn_engine)

        session = sqlite_engine.get_session()

        application = Application(id=1, name="test-application")
        session.add(application)
        session.flush()

        client = Client(id=0, name="Test", email_pattern="")
        session.add(client)
        session.flush()

最后在那堂课的测试中，我尝试了

    def test_query_config_data_default(self, sqlite_engine, setup_rbac_tables, rbac):
        conn_engine = sqlite_engine.get_engine()
        session = sqlite_engine.get_session()

        client = Client(id=1, name=factory.Faker("name").generate(), email_pattern="")
        session.add(client)
        session.flush()

        clients = sqlite_engine.get_session().query(Client).all()
        for client in clients:
            print(client.id, client.name)

但是，只有一个客户端打印（如果我尝试 for Application，则不打印），我不知道为什么。这是夹具范围的问题吗？还是发动机？或者 sqlite 如何在 pytest 中工作？

标签： pythonsqlitesqlalchemypytest

解决方案

我不是这方面的专家，但我认为您需要以共享会话的方式定义夹具，除非您计划在每个夹具中提交。在setup_rbac_tables会话中用函数作用域销毁。当再次调用 get_session 时，会创建一个新会话。

在我的 pytest sqlalchemy 测试中，我做了这样的事情，其中​​ db 夹具是在夹具之间和测试中重用的 db 会话：

@pytest.fixture
def customer_user(db):
    from ..model.user import User
    from ..model.auth import Group
    group = db.query(Group).filter(
        Group.name == 'customer').first()
    if not group:
        group = Group(name='customer', label='customer')
    user = User(email=test_email_fmt.format(uuid4().hex), group=group)
    db.add(user)
    return user

推荐阅读