python-3.x 如何使用pytest对sqlalchemy orm类进行单元测试

我想写一些py.test代码来测试两个简单的基于this Tutorial创建的sqlalchemy ORM类。问题是，我如何将py.test中的数据库设置为测试数据库，并在测试完成后回滚所有更改？是否可以模拟数据库并运行测试，而无需实际连接到de数据库？
下面是我的类的代码：

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

eng = create_engine('mssql+pymssql://user:pass@host/my_database')

Base = declarative_base(eng)
Session = sessionmaker(eng)
intern_session = Session()

class Author(Base):
    __tablename__ = "Authors"

    AuthorId = Column(Integer, primary_key=True)
    Name = Column(String)  
    Books = relationship("Book")

    def add_book(self, title):
        b = Book(Title=title, AuthorId=self.AuthorId)
        intern_session.add(b)
        intern_session.commit()

class Book(Base):
    __tablename__ = "Books"

    BookId = Column(Integer, primary_key=True)
    Title = Column(String)      
    AuthorId = Column(Integer, ForeignKey("Authors.AuthorId"))    

    Author = relationship("Author")

我通常是这样做的：
1.我没有使用模型声明示例化引擎和会话，相反，我只声明了一个没有绑定的Base：

Base = declarative_base()

并且我只在需要时才创建会话

engine = create_engine('<the db url>')
db_session = sessionmaker(bind=engine)

您可以通过在add_book方法中不使用intern_session，而是使用session参数来完成相同的操作。

def add_book(self, session, title):
    b = Book(Title=title, AuthorId=self.AuthorId)
    session.add(b)
    session.commit()

它使你的代码更易于测试，因为你现在可以在调用方法时传递你选择的会话，而且你不再被绑定到硬编码数据库url的会话所束缚。
1.我使用pytest_addoption钩子向pytest添加了一个定制的--dburl选项。
只需将其添加到您的顶级conftest.py：

def pytest_addoption(parser):
    parser.addoption('--dburl',
                     action='store',
                     default='<if needed, whatever your want>',
                     help='url of the database to use for tests')

现在您可以运行pytest --dburl <url of the test database>

然后，我可以从request fixture中检索dburl选项
从自定义夹具：

@pytest.fixture()
def db_url(request):
    return request.config.getoption("--dburl")
    # ...

在测试中：

def test_something(request):
    db_url = request.config.getoption("--dburl")
    # ...

此时，您可以：

在任何测试或夹具中获取测试db_url
用它来创造一个引擎
创建绑定到引擎的会话
将会话传递给测试方法

在每个测试中这样做是相当混乱的，所以您可以有效地使用pytest fixture来简化这个过程。
下面是我使用的一些固定装置：

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

@pytest.fixture(scope='session')
def db_engine(request):
    """yields a SQLAlchemy engine which is suppressed after the test session"""
    db_url = request.config.getoption("--dburl")
    engine_ = create_engine(db_url, echo=True)

    yield engine_

    engine_.dispose()

@pytest.fixture(scope='session')
def db_session_factory(db_engine):
    """returns a SQLAlchemy scoped session factory"""
    return scoped_session(sessionmaker(bind=db_engine))

@pytest.fixture(scope='function')
def db_session(db_session_factory):
    """yields a SQLAlchemy connection which is rollbacked after the test"""
    session_ = db_session_factory()

    yield session_

    session_.rollback()
    session_.close()

使用db_session fixture，您可以为每个测试获得一个新的干净的db_session，当测试结束时，回滚db_session，保持数据库干净。

python-3.x 如何使用pytest对sqlalchemy orm类进行单元测试

1条答案

相关问题

热门标签

最新问答