python - 如何在查询时在模式之间切换

Question

通过 MRE，我有一个名为 query 的简单模块将记录添加到 Bet table :

from container_test import Bet, DataAccessLayer

def main():
    dal = DataAccessLayer()
    dal.create_session()
    query_bets(dal)

def add_new_id(dal):
    new_id = {"id_": 3}
    dal.session.add(Bet(**new_id))
    dal.session.commit()

if __name__ == "__main__":
    main()

container_test包含以下内容:

from sqlalchemy import Column, Integer, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Bet(Base):

    __tablename__ = "bet"
    __table_args__ = ({"schema": "belgarath", "extend_existing": True})

    id_ = Column(Integer, primary_key=True)

class DataAccessLayer():

    def __init__(self):
        conn_string = "mysql+mysqlconnector://root:root@localhost/"
        self.engine = create_engine(conn_string)

    def create_session(self):
        Base.metadata.create_all(self.engine)
        Session = sessionmaker()
        Session.configure(bind=self.engine)
        self.session = Session()

我希望能够测试 query_bets函数，为此我创建了一个简单的 pytest另一个文件中的函数:

def test_add_new_id(
    create_dal, # creates an instance of a dal via conftest
    setup_teardown, # creates a blank belgarath_test schema and deletes afterwards via conftest
):
    expected = 3
    sandpit.add_new_id(create_dal)
    actual = query_bet(create_dal) # function that will return a scalar result
    assert actual == expected

但是，我希望能够更改标准 belgarath模式到测试模式； belgarath_test所以我不会冒险主数据库的完整性。我想我需要某种可以在 test_add_new_id 中激活的“开关”测试功能，但我不知道如何创建一个。
我一直在尝试遵循这篇文章中提出的解决方案: How do I change the schema for both a table and a foreign key?但是我很挣扎。我不明白我需要在 test_add_new_id 中做什么测试函数来改变模式。
作为引用，我根据解决方案创建了以下模块: container_test :

from sqlalchemy import Column, Integer, create_engine
from sqlalchemy.orm import sessionmaker

from base_class import SCHEMA_TEST, Base, declared_attr


class Bet(Base):

    @declared_attr
    def __table_args__(cls):
        return {'schema': SCHEMA_TEST}

    __tablename__ = "bet"

    id_ = Column(Integer, primary_key=True)


class DataAccessLayer():

    def __init__(self):
        conn_string = "mysql+mysqlconnector://root:root@localhost/"
        self.engine = create_engine(conn_string)

    def create_session(self):
        Base.metadata.create_all(self.engine)
        Session = sessionmaker()
        Session.configure(bind=self.engine)
        self.session = Session()

base_class (尝试使用 __init__.py 时出现错误):

from sqlalchemy.ext.declarative import declarative_base, declared_attr

SCHEMA_MAIN = 'belgarath'  # figure out how you want to retrieve this
SCHEMA_TEST = 'belgarath_test'


class _Base():

    @declared_attr
    def __table_args__(cls):
        return {'schema': SCHEMA_MAIN}


Base = declarative_base(cls=_Base)
Base.metadata.schema = SCHEMA_MAIN

这成功改变了 belgarath至 belgarath_test但我只希望在运行 test_add_new_id 时发生这种情况测试功能。

Answer 1

与其在同一个数据库的不同模式中运行测试，不如在不同的数据库中运行测试？
这也将改善测试隔离，因为测试将不再具有生产数据库的密码，因此有缺陷的测试将无法破坏产品数据。