单元测试依赖于数据库的函数

Question

我正在运行测试一些功能。我有一个使用数据库查询的函数。所以，我浏览了博客和文档，说我们必须在内存或测试数据库中创建一个才能使用这些功能。下面是我的函数，

def already_exists(story_data,c):
    # TODO(salmanhaseeb): Implement de-dupe functionality by checking if it already
    # exists in the DB.
    c.execute("""SELECT COUNT(*) from posts where post_id = ?""", (story_data.post_id,))
    (number_of_rows,)=c.fetchone()
    if number_of_rows > 0:
        return True
    return False

此函数命中生产数据库。我的问题是，在测试时，我创建一个内存数据库并在那里填充我的值，我将查询该数据库（测试数据库）。但我想测试我的 already_exists() 函数，在从测试中调用我的 already_exists 函数后，我的生产数据库将被命中。如何在测试此功能时使我的测试数据库命中？

Answer 1

问题在于确保您的代码始终使用相同的数据库连接。然后你可以将它设置为适合当前环境的任何值。

与其将数据库连接从一个方法传递到另一个方法，不如将其设为单例。

def already_exists(story_data):
    # Here `connection` is a singleton which returns the database connection.
    connection.execute("""SELECT COUNT(*) from posts where post_id = ?""", (story_data.post_id,))
    (number_of_rows,) = connection.fetchone()
    if number_of_rows > 0:
        return True
    return False

或者在每个 class 上创建 connection 一个方法，然后将 already_exists 变成一个方法。无论如何，它应该是一种方法。

def already_exists(self):
    # Here the connection is associated with the object.
    self.connection.execute("""SELECT COUNT(*) from posts where post_id = ?""", (self.post_id,))
    (number_of_rows,) = self.connection.fetchone()
    if number_of_rows > 0:
        return True
    return False

但实际上您不应该自己滚动此代码。相反，您应该使用 ORM such as SQLAlchemy which takes care of basic queries and connection management like this for you. It has a single connection, the "session".

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from sqlalchemy_declarative import Address, Base, Person

engine = create_engine('sqlite:///sqlalchemy_example.db')
Base.metadata.bind = engine

DBSession = sessionmaker(bind=engine)
session = DBSession()

然后你用它来进行查询。例如，it has an exists method.

session.query(Post.id).filter(q.exists()).scalar()

使用 ORM 将大大简化您的代码。这里是 a short tutorial for the basics, and a longer and more complete tutorial.

Answer 2

解决这个问题有两种方法：

进行集成测试而不是单元测试，只需使用真实数据库的副本。
为方法提供伪造的而不是实际的连接对象。

你应该做哪一个取决于你想要达到的目标。

如果您想测试查询本身是否有效，那么您应该使用集成测试。句号。确保查询按预期进行的唯一方法是运行它与数据库副本中已有的测试数据一起使用。运行它针对 不同的 数据库技术（例如，当您的生产数据库在 PostgreSQL 中时运行ning 针对 SQLite）将不能确保它在生产中工作。需要数据库的副本意味着您将需要一些自动化的部署过程，可以针对单独的数据库轻松调用。无论如何，你应该有这样一个自动化过程，因为它有助于确保你的跨环境部署是一致的，允许你在发布之前测试它们，以及 "documents" 升级数据库的过程。对此的标准解决方案是使用您的编程语言编写的迁移工具，例如 albemic or tools to execute raw SQL like yoyo or Flyway。您需要在运行测试之前调用部署并用测试数据填充它，然后运行测试并断言您希望 return 编辑的输出。

如果您想测试查询周围的代码而不是查询本身，那么您可以为连接对象使用伪造的代码。最常见的解决方案是 a mock。模拟提供了可以配置为接受函数调用和输入以及 return 一些输出来代替真实对象的替代品。这将允许您测试该方法的逻辑是否正常工作，假设查询 returns 是您期望的结果。对于您的方法，此类测试可能如下所示：

from unittest.mock import Mock ... def test_already_exists_returns_true_for_positive_count(): mockConn = Mock( execute=Mock(), fetchone=Mock(return_value=(5,)), ) story = Story(post_id=10) # Making some assumptions about what your object might look like. result = already_exists(story, mockConn) assert result # Possibly assert calls on the mock. Value of these asserts is debatable. mockConn.execute.assert_called("""SELECT COUNT(*) from posts where post_id = ?""", (story.post_id,)) mockConn.fetchone.assert_called()

单元测试依赖于数据库的函数

Unit testing a function that depends on database

python

unit-testing

python-unittest

database-testing