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

Question

我想编写一些 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")                           

Answer 1

我通常这样做：

1. 我不使用模型声明实例化引擎和会话，而是只声明一个没有绑定的 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 的会话。

2. 我使用其

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

3. 然后我可以从

   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

将回滚，保持数据库干净。

Answer 2

fasdf d d d

d

dddddd## asdfasdf ##