我想编写一些 py.test 代码来测试基于 本教程 创建的 2 个简单的 sqlalchemy ORM 类。问题是,如何将 py.test 中的数据库设置为测试数据库并在测试完成后回滚所有更改?是否可以在不实际连接到数据库的情况下模拟数据库并运行测试?
这是我的课程的代码:
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")
我通常这样做:
我不使用模型声明实例化引擎和会话,而是只声明一个没有绑定的 Base:
Base = declarative_base()
我只在需要时创建一个会话
engine = create_engine('<the db url>')
db_session = sessionmaker(bind=engine)
您可以通过在
intern_session
方法中不使用 add_book
而是使用 session
参数来执行相同的操作。
def add_book(self, session, title):
b = Book(Title=title, AuthorId=self.AuthorId)
session.add(b)
session.commit()
它使您的代码更具可测试性,因为您现在可以在调用该方法时传递您选择的会话。 您不再受困于绑定到硬编码数据库 URL 的会话。
我使用其
--dburl
钩子向 pytest 添加了自定义
pytest_addoption
选项。
只需将其添加到您的顶级
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>
然后我可以从
dburl
夹具中检索
request
选项
来自自定义夹具:
@pytest.fixture()
def db_url(request):
return request.config.getoption("--dburl")
# ...
测试内部:
def test_something(request):
db_url = request.config.getoption("--dburl")
# ...
此时您可以:
db_url
在每次测试中都这样做是相当混乱的,因此您可以有效地使用 pytest 夹具来简化过程。
以下是我使用的一些灯具:
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
夹具,您可以为每次测试获得新鲜干净的 db_session
。
当测试结束时,db_session
将回滚,保持数据库干净。
fasdf d d d
d
dddddd## asdfasdf ##