如何使用 pytest 对 sqlalchemy orm 类进行单元测试 [英] How use pytest to unit test sqlalchemy orm classes

问题描述

我想编写一些 py.test 代码来测试基于 创建的 2 个简单的 sqlalchemy ORM 类本教程.问题是，如何将 py.test 中的数据库设置为测试数据库并在测试完成后回滚所有更改?是否可以在不实际连接到数据库的情况下模拟数据库并运行测试?

I want to write some py.test code to test 2 simple sqlalchemy ORM classes that were created based on this Tutorial. The problem is, how do I set a the database in py.test to a test database and rollback all changes when the tests are done? Is it possible to mock the database and run tests without actually connect to 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")                           

推荐答案

我通常是这样做的:

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

1. I do not instantiate engine and session with the model declarations, instead I only declare a Base with no bind:

Base = declarative_base()

我只在需要时创建一个会话

and I only create a session when needed with

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

您可以不使用 add_book 方法中的 intern_session 而是使用 session 参数来实现相同的效果.

You can do the same by not using the intern_session in your add_book method but rather use a session parameter.

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

它使您的代码更具可测试性，因为您现在可以在调用该方法时传递您选择的会话.您再也不会被绑定到硬编码数据库 url 的会话困住了.

It makes your code more testable since you can now pass the session of your choice when you call the method. And you are no more stuck with a session bound to a hardcoded database url.

使用其 pytest_addoption 钩子.

Add a custom --dburl option to pytest using its pytest_addoption hook.

只需将此添加到您的顶级conftest.py:

Simply add this to your top-level 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>

然后您可以从 请求fixture

Then you can retrieve the dburl option from the request fixture

• 来自自定义装置:

• From a custom fixture:

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

在测试中:

Inside a test:

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

---

此时您可以:

---

At this point you are able to:

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

在每次测试中都这样做很麻烦，因此您可以有用地使用 pytest 固定装置来简化过程.

It is quite a mess to do this in every test, so you can make a usefull usage of pytest fixtures to ease the process.

以下是我使用的一些装置:

Below are some fixtures I use:

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 被回滚，保持数据库干净.

Using the db_session fixture you can get a fresh and clean db_session for each single test. When the test ends, the db_session is rollbacked, keeping the database clean.

这篇关于如何使用 pytest 对 sqlalchemy orm 类进行单元测试的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！