Pytest 和 PostgreSQL：每次测试的新数据库（第二部分）

百变鹏仔 3天前 #Python

文章标签第二部分

在上一篇文章中，我们创建了 pytest 夹具，它将在测试方法之前/之后创建/删除 postgres 数据库。在这一部分中，我想在 pytest 工厂固定装置的帮助下改进固定装置，使其更加灵活和可配置。

静态夹具的限制

例如，如果您有多个数据库要在测试中模拟

def test_create_user(test_db1, test_db2):    ...

您必须创建几乎两个相同的灯具：

test_db_url = "postgresql://localhost"test_db1_name = "test_foo"test_db2_name = "test_bar"@pytest.fixturedef test_db1():    with psycopg.connect(test_db_url, autocommit=true) as conn:        cur = conn.cursor()        cur.execute(f'drop database if exists "{test_db1_name}" with (force)')        cur.execute(f'create database "{test_db1_name}"')        with psycopg.connect(test_db_url, dbname=test_db1_name) as conn:            yield conn        cur.execute(f'drop database if exists "{test_db1_name}" with (force)')@pytest.fixturedef test_db2():    with psycopg.connect(test_db_url, autocommit=true) as conn:        cur = conn.cursor()        cur.execute(f'drop database if exists "{test_db2_name}" with (force)')        cur.execute(f'create database "{test_db2_name}"')        with psycopg.connect(test_db_url, dbname=test_db2_name) as conn:            yield conn        cur.execute(f'drop database if exists "{test_db2_name}" with (force)')

pytest 夹具工厂

“静态”装置在这里有点限制。当需要几乎相同且仅有细微差别时，您需要复制代码。希望 pytest 有工厂作为固定装置的概念。

工厂固定装置是一个返回另一个固定装置的固定装置。因为，像每个工厂一样，它是一个函数，它可以接受参数来自定义返回的固定装置。按照惯例，您可以在它们前面加上 make_* 前缀，例如 make_test_db。

专用夹具

我们的装置工厂 make_test_db 的唯一参数将是要创建/删除的测试数据库名称。

那么，让我们基于 make_test_db 工厂装置创建两个“专用”装置。

用法如下：

@pytest.fixturedef test_db_foo(make_test_db):    yield from make_test_db("test_foo")@pytest.fixturedef test_db_bar(make_test_db):    yield from make_test_db("test_bar")

旁注：产量来自

你注意到产量了吗？ yield 和 yield 之间的一个关键区别在于它们如何处理生成器内的数据流和控制。

在python中，yield和yield from都在生成器函数中使用来生成一系列值，但是

也就是说，我们不想从专门的夹具“屈服”，而是从夹具工厂“屈服”。因此这里需要yield from。

用于创建/删除数据库的夹具工厂

除了将代码包装到内部函数之外，对我们原始夹具创建/删除数据库所需的更改实际上几乎不需要任何更改。

@pytest.fixturedef make_test_db():    def _(test_db_name: str):        with psycopg.connect(test_db_url, autocommit=true) as conn:            cur = conn.cursor()            cur.execute(f'drop database if exists "{test_db_name}" with (force)') # type: ignore            cur.execute(f'create database "{test_db_name}"') # type: ignore            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)') # type: ignore    yield _

奖励：将迁移固定装置重写为工厂固定装置

在上一部分中，我还有一个固定装置，将 yoyo 迁移应用于刚刚创建的空数据库。它也不是很灵活。让我们做同样的事情并将实际代码包装到内部函数中。

在这种情况下，因为代码不需要在从测试方法返回后进行清理（其中没有yield），所以

@pytest.fixturedef make_yoyo():    """applies yoyo migrations to test db."""    def _(test_db_name: str, migrations_dir: str):        url = (            urlparse(test_db_url)            .            _replace(scheme="postgresql+psycopg")            .            _replace(path=test_db_name)            .geturl()        )        backend = get_backend(url)        migrations = read_migrations(migrations_dir)        if len(migrations) == 0:            raise valueerror(f"no yoyo migrations found in '{migrations_dir}'")        with backend.lock():            backend.apply_migrations(backend.to_apply(migrations))    return _@pytest.fixturedef yoyo_foo(make_yoyo):    migrations_dir = str(path(__file__, "../../foo/migrations").resolve())    make_yoyo("test_foo", migrations_dir)@pytest.fixturedef yoyo_bar(make_yoyo):    migrations_dir = str(path(__file__, "../../bar/migrations").resolve())    make_yoyo("test_bar", migrations_dir)

需要两个数据库并对它们应用迁移的测试方法：

from psycopg import Connectiondef test_get_new_users_since_last_run(        test_db_foo: Connection,        test_db_bar: Connection,        yoyo_foo,        yoyo_bar):    test_db_foo.execute("...")    ...

结论

构建自己的夹具工厂，为 pytest 方法创建和删除数据库实际上是练习 python 生成器和运算符的产量/产量的一个很好的练习。

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

文章推荐

Pytest 和 PostgreSQL：每次测试的新数据库（第二部分）

静态夹具的限制

pytest 夹具工厂

专用夹具

旁注：产量来自

用于创建/删除数据库的夹具工厂

奖励：将迁移固定装置重写为工厂固定装置

结论

Python - 使用切片获取最后K个列表项的总和

Python程序计算标准差

使用内置函数的Python程序来反转数组的元素

Python程序检测链表中的循环

在Python中，两个字典键的乘积