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

在上一篇文章中，我们创建了 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.fixture
def 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.fixture
def 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.fixture
def test_db_foo(make_test_db):
    yield from make_test_db("test_foo")

@pytest.fixture
def test_db_bar(make_test_db):
    yield from make_test_db("test_bar")

旁注：产量来自

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

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

yield 用于暂停生成器函数的执行并向调用者返回单个值。

而yield from用于将值的生成委托给另一个生成器。它本质上“展平”了嵌套生成器，将其生成的值直接传递给外部生成器的调用者。

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

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

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

@pytest.fixture
def 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），所以

工厂装置返回（不是yield）内部函数

专门的夹具调用（不是从工厂夹具中产生）

@pytest.fixture
def 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.fixture
def yoyo_foo(make_yoyo):
    migrations_dir = str(path(__file__, "../../foo/migrations").resolve())
    make_yoyo("test_foo", migrations_dir)

@pytest.fixture
def yoyo_bar(make_yoyo):
    migrations_dir = str(path(__file__, "../../bar/migrations").resolve())
    make_yoyo("test_bar", migrations_dir)

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

from psycopg import Connection

def test_get_new_users_since_last_run(
        test_db_foo: Connection,
        test_db_bar: Connection,
        yoyo_foo,
        yoyo_bar):
    test_db_foo.execute("...")
    ...