在 pytest（每个人最喜欢的 Python 测试框架）中，fixture 是一段可重用的代码，它在测试进入之前安排一些事情，并在测试退出后进行清理。例如，临时文件或文件夹、设置环境、启动 Web 服务器等。在这篇文章中，我们将了解如何创建 pytest 夹具，该夹具创建一个可以清理的测试数据库（空或已知状态） ，允许每个测试在完全干净的数据库上运行.
目标
我们将使用 PSycopg 3 创建一个 pytest 夹具来准备和清理测试数据库。因为空数据库对测试几乎没有帮助，所以我们将选择应用 yoyo 迁移（在撰写本文时网站已关闭，请转到 archive.org 快照）来填充它。
因此，对本博文中创建的名为 test_db 的 pytest 夹具的要求是：
- 删除测试数据库如果在测试之前存在
- 测试前创建一个空数据库 可选地
在测试之前应用迁移或创建测试数据
- 提供到测试数据库
的连接到测试
- 测试后删除测试数据库
（即使失败）
- 任何通过列出测试方法参数来请求它的测试方法：
|
1
2
|
def test_create_admin_Table(test_db):
...
|
将收到连接到测试数据库的常规 psycopg connection 实例。测试可以做任何需要做的事情，就像普通的 psycopg 常见用法一样，例如：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
def test_create_admin_table(test_db):
# open a cursor to perfORM database operations
cur = test_db.cursor()
# pass data to fill a query placeholders and let psycopg perform
# the correct conversion (no sql injections!)
cur.execute(
"insert into test (num, data) values (%s, %s)",
(100, "abc'def"))
# query the database and obtain data as python objects.
cur.execute("select * from test")
cur.fetchone()
# will return (1, 100, "abc'def")
# you can use `cur.fetchmany()`, `cur.fetchall()` to return a list
# of several records, or even iterate on the cursor
for record in cur:
print(record)
|
动机和替代方案
我尝试过 pytest-postgresql ，它也有同样的承诺。在编写自己的装置之前我已经尝试过它，但我无法让它为我工作。也许是因为他们的文档让我很困惑。
项目文件布局
在经典的python项目中，源代码位于src/中，测试位于tests/中：
|
1
2
3
4
5
6
7
8
9
10
11
|
├── src
│ └── tuvok
│ ├── __init__.py
│ └── sales
│ └── new_user.py
├── tests
│ ├── conftest.py
│ └── sales
│ └── test_new_user.py
├── requirements.txt
└── yoyo.ini
|
如果你使用像 fantasyal yoyo 这样的迁移库，迁移脚本很可能在migrations/:
|
1
2
3
|
├── migrations
├── 20240816_01_yn3ca-sales-user-user-add-last-run-table.py
├── ...
|
配置
我们的测试数据库夹具需要很少的配置：
连接 url
– （无数据库）
- 测试数据库名称
– 将为每个测试重新创建
- （可选）迁移文件夹
– 应用于每个测试的迁移脚本
- pytest 有一个天然的地方 conftest.py 用于跨多个文件共享固定装置。灯具配置也会在那里：
|
1
2
3
4
|
# without db name!
test_db_url = "postgresql://localhost"
test_db_name = "test_tuvok"
test_db_migrations_dir = str(path(__file__, "../../migrations").resolve())
|
您可以从环境变量或任何适合您情况的值来设置这些值。
创建 test_db 夹具
有了
postgresql和psycopg库的知识
，在conftest.py中编写fixture：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
<strong>@pytest.fixture
def test_db():
# autocommit=true start no transaction because create/drop database
# cannot be executed in a transaction block.
with psycopg.connect(test_db_url, autocommit=true) as conn:
cur = conn.cursor()
# create test db, drop before
cur.execute(f'drop database if exists "{test_db_name}" with (force)')
cur.execute(f'create database "{test_db_name}"')
# return (a new) connection to just created test db
# unfortunately, you cannot directly change the database for an existing psycopg connection. once a connection is established to a specific database, it's tied to that database.
with psycopg.connect(test_db_url, dbname=test_db_name) as conn:
yield conn
cur.execute(f'drop database if exists "{test_db_name}" with (force)')
</strong>
|
创建迁移固定装置
在我们的例子中，我们使用
yoyo 迁移
。将应用迁移编写为另一个名为 yoyo 的固定装置：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
<strong>@pytest.fixture
def yoyo():
# yoyo expect `driver:
# in passed url we need to
url = (
urlparse(test_db_url)
.
# 1) change driver (schema part) with `postgresql+psycopg` to use
# psycopg 3 (not 2 which is `postgresql+psycopg2`)
_replace(scheme="postgresql+psycopg")
.
# 2) change database to test db (in which migrations will apply)
_replace(path=test_db_name)
.geturl()
)
backend = get_backend(url)
migrations = read_migrations(test_db_migrations_dir)
if len(migrations) == 0:
raise valueerror(f"no yoyo migrations found in '{test_db_migrations_dir}'")
with backend.lock():
backend.apply_migrations(backend.to_apply(migrations))
</strong>
|
如果你想
将迁移应用到每个测试数据库
，需要 yoyo 夹具用于 test_db 夹具：
|
1
2
3
4
|
<strong>@pytest.fixture
def test_db(yoyo):
...
</strong>
|
要
仅将迁移应用于某些测试
，需要单独使用yoyo：
|
1
2
3
|
<strong>def test_create_admin_table(test_db, yoyo):
...
</strong>
|
结论
构建自己的装置来为您的测试提供一个干净的数据库对我来说是一次有益的经历，让我能够更深入地研究 pytest 和 postgres。
我希望这篇文章对您自己的数据库测试套件有所帮助。请随时在评论中留下您的问题，祝您编码愉快！
