如何使用pytest对sqlalchemy orm类进行单元测试

Question

我想写一些py.测试测试基于this Tutorial创建的2个简单sqlalchemy ORM类的代码。问题是，如何在py.测试到测试数据库并在测试完成时回滚所有更改？是否可以模拟数据库并运行测试而不实际连接到de database？在

以下是我的课程代码：

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")                           

Answer 1

我通常是这样做的：

1. 我没有用模型声明实例化引擎和会话，而是只声明一个没有绑定的基：

   Base = declarative_base()
   

   我只在需要的时候用

   ^{pr2}$

   您可以通过不在add_book方法中使用intern_session而使用session参数来实现相同的目的。在

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

   它使代码更易于测试，因为现在可以在调用方法时通过所选的会话。 而且，您不再被绑定到硬编码数据库url的会话所困扰。

2. 使用其^{} hook向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>

3. 然后可以从^{}夹具中检索dburl选项

   • 从自定义夹具：

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

   • 在测试中：

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

---

此时，您可以：

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

在每个测试中都要这样做是相当混乱的，所以您可以充分利用pytest fixture来简化这个过程。在

以下是我使用的一些固定装置：

@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=engine))


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

    yield session_

    session_.rollback()
    session_.close()

使用db_session夹具，您可以为每个单独的测试获得一个新鲜而干净的db_session。 当测试结束时，db_session被回滚，保持数据库的干净。在