在 Pytest（每个人最喜欢的 Python 测试框架）中，fixture 是一段可重用的代码，它在测试进入之前安排 something，并在测试退出后进行清理。例如，临时文件或文件夹、设置环境、启动 Web 服务器等。在这篇文章中，我们将了解如何创建 Pytest 夹具，该夹具创建一个测试数据库（空或已知状态），该数据库获取清理，允许每个测试在完全干净的数据库上运行。

目标

我们将使用 Psycopg 3 创建一个 Pytest 夹具来准备和清理测试数据库。因为空数据库对测试几乎没有帮助，所以我们将选择应用 Yoyo 迁移（在撰写本文时网站已关闭，请转到 archive.org 快照）来填充它。

因此，在此博文中创建的名为 test_db 的 Pytest 夹具的要求是​​：

• 删除测试数据库如果在测试之前存在
• 在测试之前创建一个空数据库
可选地
• 在测试之前应用迁移或创建测试数据
• 提供到测试数据库的连接到测试
• 测试后删除测试数据库（即使失败）

通过列出测试方法参数来请求它的任何测试方法：

def test_create_admin_table(test_db): ...

def test_create_admin_table(test_db):
    ...

将收到连接到测试数据库的常规 Psycopg 连接实例。测试可以做任何它需要的事情，就像普通的 Psycopg 常见用法一样，例如：

def test_create_admin_table(test_db): # 打开游标执行数据库操作 cur = test_db.cursor() # 传递数据来填充查询占位符并让 Psycopg 执行 # 正确的转换（没有 SQL 注入！） 当前.执行( “插入测试（数字，数据）值（%s，%s）”， （100，“abc'def”）） # 查询数据库并获取数据作为Python对象。 cur.execute("从测试中选择*") cur.fetchone() # 将返回 (1, 100, "abc'def") # 你可以使用 `cur.fetchmany()`, `cur.fetchall()` 返回一个列表 # 多个记录，甚至在光标上迭代 以 cur 记录： 打印（记录）

def test_create_admin_table(test_db):
    ...

动机和替代方案

看起来有一些 Pytest 插件承诺为依赖数据库的测试提供 PostgreSQL 固定装置。它们可能很适合你。
我尝试过 pytest-postgresql 它承诺相同。在编写自己的装置之前我已经尝试过它，但我无法让它为我工作。也许是因为他们的文档让我很困惑。

另一个，pytest-dbt-postgres，我根本没有尝试过。

项目文件布局

在经典的Python项目中，源代码位于src/中，测试位于tests/中：

├── src │ └── 杜沃克 │ ├── __init__.py │ └── 销售 │ └── new_user.py ├── 测试 │ ├──conftest.py │ └── 销售 │ └── test_new_user.py ├── 需求.txt └── yoyo.ini

def test_create_admin_table(test_db):
    ...

如果你使用像梦幻般的Yoyo这样的迁移库，迁移脚本可能在migrations/:

├── 迁移 ├── 20240816_01_Yn3Ca-sales-user-user-add-last-run-table.py ├── ...

def test_create_admin_table(test_db):
    ...

配置

我们的测试数据库夹具需要很少的配置：

• 连接 URL - （无数据库）
• 测试数据库名称 - 将为每个测试重新创建
（可选）
• 迁移文件夹 - 应用于每个测试的迁移脚本

Pytest 有一个天然的地方 conftest.py 用于跨多个文件共享固定装置。灯具配置也会在那里：

# 没有数据库名称！ TEST_DB_URL = “postgresql://localhost” TEST_DB_NAME =“test_tuvok” TEST_DB_MIGRATIONS_DIR = str(Path(__file__, "../../migrations").resolve())

def test_create_admin_table(test_db):
    ...

您可以从环境变量或任何适合您情况的值中设置这些值。

创建 test_db 夹具

了解

PostgreSQL和Psycopg库，在conftest.py中编写fixture：

@pytest.fixture def test_db(): # autocommit=True 不启动事务，因为 CREATE/DROP DATABASE # 不能在事务块中执行。 使用 psycopg.connect(TEST_DB_URL, autocommit=True) 作为 conn： cur = conn.cursor() # 创建测试数据库，之前删除 cur.execute(f'如果存在“{TEST_DB_NAME}”则删除数据库(FORCE)') cur.execute(f'创建数据库“{TEST_DB_NAME}”') # 返回到刚刚创建的测试数据库的（新）连接 # 不幸的是，您无法直接更改现有 Psycopg 连接的数据库。一旦建立了与特定数据库的连接，它就与该数据库绑定在一起。 使用 psycopg.connect(TEST_DB_URL, dbname=TEST_DB_NAME) 作为 conn： 产量康涅狄格州 cur.execute(f'如果存在“{TEST_DB_NAME}”则删除数据库(FORCE)')

def test_create_admin_table(test_db):
    ...

创建迁移固定装置

在我们的例子中，我们使用

Yoyo 迁移。将应用迁移编写为另一个名为 yoyo 的固定装置：

@pytest.fixture def yoyo(): # Yoyo 期望 `driver://user:pass@host:port/database_name?param=value`。 # 在传递的 URL 中我们需要 网址 = ( urlparse(TEST_DB_URL) 。 # 1) 使用 `postgresql psycopg` 更改驱动程序（架构部分）以使用 # psycopg 3（不是 2，即 `postgresql psycopg2`） _replace(scheme="postgresql psycopg") 。 # 2) 将数据库更改为测试数据库（其中将应用迁移） _replace（路径=TEST_DB_NAME） .geturl() ） 后端 = get_backend(url) 迁移 = read_migrations(TEST_DB_MIGRATIONS_DIR) 如果 len(迁移) == 0: raise ValueError(f"在 '{TEST_DB_MIGRATIONS_DIR}' 中找不到 Yoyo 迁移'") 与 backend.lock(): backend.apply_migrations（backend.to_apply（迁移））

def test_create_admin_table(test_db):
    ...

如果你想

将迁移应用到每个测试数据库，需要 yoyo 夹具用于 test_db 夹具：

@pytest.fixture def test_db(yoyo): ...

def test_create_admin_table(test_db):
    ...

要

仅将迁移应用于某些测试，需要单独使用 yoyo：

def test_create_admin_table(test_db, yoyo): ...

def test_create_admin_table(test_db):
    ...

结论

构建自己的装置来为您的测试提供一个干净的数据库对我来说是一次有益的经历，让我能够更深入地研究 Pytest 和 Postgres。

我希望本文对您自己的数据库测试套件有所帮助。请随时在评论中留下您的问题并祝您编码愉快！