在 pytest（每个人最喜欢的 python 测试框架）中，fixture 是一段可重用的代码，它在测试进入之前安排一些事情，并在测试退出后进行清理。例如，临时文件或文件夹、设置环境、启动 web 服务器等。在这篇文章中，我们将了解如何创建 pytest 夹具，该夹具创建一个可以清理的测试数据库（空或已知状态） ，允许每个测试在完全干净的数据库上运行.
目标我们将使用 psycopg 3 创建一个 pytest 夹具来准备和清理测试数据库。因为空数据库对测试几乎没有帮助，所以我们将选择应用 yoyo 迁移（在撰写本文时网站已关闭，请转到 archive.org 快照）来填充它。
因此，对本博文中创建的名为 test_db 的 pytest 夹具的要求是：
- 删除测试数据库如果在测试之前存在
- 测试前创建一个空数据库 可选地 在测试之前应用迁移或创建测试数据
- 提供到测试数据库 的连接到测试
- 测试后删除测试数据库 （即使失败）
- 任何通过列出测试方法参数来请求它的测试方法：
def test_create_admin_table(test_db):
...
将收到连接到测试数据库的常规 psycopg connection 实例。测试可以做任何需要做的事情，就像普通的 psycopg 常见用法一样，例如：
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 ，它也有同样的承诺。在编写自己的装置之前我已经尝试过它，但我无法让它为我工作。也许是因为他们的文档让我很困惑。
另一个，pytest-dbt-postgres，我根本没有尝试过。
├── src │ └── tuvok │ ├── __init__.py │ └── sales │ └── new_user.py ├── tests │ ├── conftest.py │ └── sales │ └── test_new_user.py ├── requirements.txt └── yoyo.ini
如果你使用像 fantasyal yoyo 这样的迁移库，迁移脚本很可能在migrations/:
├── migrations
├── 20240816_01_yn3ca-sales-user-user-add-last-run-table.py
├── ...
配置
连接 url
- - （无数据库）
- 测试数据库名称 - 将为每个测试重新创建
- （可选）迁移文件夹 - 应用于每个测试的迁移脚本
- pytest 有一个天然的地方 conftest.py 用于跨多个文件共享固定装置。灯具配置也会在那里：
# without db name! test_db_url = "postgresql://localhost" test_db_name = "test_tuvok" test_db_migrations_dir = str(path(__file__, "../../migrations").resolve())
您可以从环境变量或任何适合您情况的值来设置这些值。
有了
postgresql和psycopg库的知识
，在conftest.py中编写fixture：
@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)')
创建迁移固定装置
yoyo 迁移
。将应用迁移编写为另一个名为 yoyo 的固定装置：
@pytest.fixture
def yoyo():
# yoyo expect `driver://user:pass@host:port/database_name?param=value`.
# 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))
如果你想
将迁移应用到每个测试数据库
@pytest.fixture
def test_db(yoyo):
...
要
仅将迁移应用于某些测试
def test_create_admin_table(test_db, yoyo):
...
结论
我希望这篇文章对您自己的数据库测试套件有所帮助。请随时在评论中留下您的问题，祝您编码愉快！
