如何让docker-py启动的PostgreSQL容器优先于pytest-postgresql fixture运行？

问题背景 #

原本手动执行docker run ...启动PostgreSQL容器后，使用pytest-postgresql插件测试应用一切正常。原有conftest.py配置如下：

from pytest_postgresql import factories

postgresql_noproc = factories.postgresql_noproc(
    host="127.0.0.1",
    port=65432,
    password="supersecretpassword",
)
postgres_client = factories.postgresql("postgresql_noproc")

测试代码示例：

from psycopg import sql
from psycopg.rows import dict_row

def test_tables_count(postgres_client):
    with postgres_client.cursor(row_factory=dict_row) as cursor:
        cursor.execute(
            sql.SQL(
                "SELECT COUNT(*) FROM information_schema.tables "
                "WHERE table_type = 'BASE TABLE'"
                " AND table_schema NOT IN ('pg_catalog', 'information_schema')"
            )
        )
        result = cursor.fetchone()
    assert result["count"] == 0

希望通过docker-py自动启动容器，编写了psql_docker fixture：

import time
import docker
import pytest

@pytest.fixture
def psql_docker():
    client = docker.from_env()
    container = client.containers.run(
        image="postgres:12",
        auto_remove=True,
        environment=dict(
            POSTGRES_PASSWORD="supersecretpassword",
        ),
        name="test_postgres",
        ports={"5432/tcp": ("127.0.0.1", 65432)},
        detach=True,
        remove=True,
    )
    
    # 等待容器启动（实际使用更复杂的检查逻辑）
    time.sleep(5)
    
    yield

    container.stop()

但由于factories.postgresql_noproc和factories.postgresql是模块级实体，初始化早于psql_docker fixture，导致容器未启动就尝试连接。两次尝试均失败：

尝试0：将postgres_client改为自定义fixture #

@pytest.fixture
def postgres_client(psql_docker):
    return factories.postgresql("postgresql_noproc")

报错：

postgres_client = <function postgresql.<locals>.postgresql_factory at 0x7f810cb0d670>

    def test_tables_count(postgres_client):
>       with postgres_client.cursor(row_factory=dict_row) as cursor:
E       AttributeError: 'function' object has no attribute 'cursor'

tests/unit_postgres_so/test_psql_connection.py:6: AttributeError

原因：返回的是工厂函数对象，而非数据库连接实例。

尝试1：在自定义fixture中直接调用工厂函数 #

@pytest.fixture
def postgres_client(psql_docker):
    return factories.postgresql("postgresql_noproc")()

报错：Fixture "postgresql_factory" called directly，违反pytest fixture的调用规则。

可行解决方案 #

核心思路是让postgresql_noproc fixture依赖psql_docker，确保容器先启动再初始化数据库连接。修改后的conftest.py如下：

import time
import docker
import pytest
from pytest_postgresql import factories
import psycopg
from psycopg import OperationalError

@pytest.fixture
def psql_docker():
    client = docker.from_env()
    # 清理同名容器避免冲突
    try:
        existing_container = client.containers.get("test_postgres")
        existing_container.stop()
        existing_container.remove()
    except docker.errors.NotFound:
        pass

    container = client.containers.run(
        image="postgres:12",
        auto_remove=True,
        environment={"POSTGRES_PASSWORD": "supersecretpassword"},
        name="test_postgres",
        ports={"5432/tcp": ("127.0.0.1", 65432)},
        detach=True,
        remove=True,
    )
    
    # 替换sleep为可靠的连接检查，确保PostgreSQL服务就绪
    max_retries = 10
    retry_interval = 1
    for _ in range(max_retries):
        try:
            with psycopg.connect(
                host="127.0.0.1",
                port=65432,
                user="postgres",
                password="supersecretpassword",
                dbname="postgres"
            ):
                break
        except OperationalError:
            time.sleep(retry_interval)
    else:
        raise RuntimeError("PostgreSQL容器启动超时，无法建立连接")
    
    yield container

    container.stop()

# 自定义postgresql_noproc fixture，依赖psql_docker确保容器先启动
@pytest.fixture
def postgresql_noproc(psql_docker):
    return factories.postgresql_noproc(
        host="127.0.0.1",
        port=65432,
        password="supersecretpassword",
    )()

# 基于自定义noproc fixture创建数据库客户端
postgres_client = factories.postgresql("postgresql_noproc")

方案说明 #

1. 调整fixture依赖顺序：将postgresql_noproc定义为依赖psql_docker的fixture，pytest会优先执行psql_docker启动容器，再初始化数据库连接配置。
2. 可靠的启动检查：替换固定sleep为循环检查PostgreSQL连接，避免因容器启动速度慢导致的连接失败。
3. 容器冲突处理：启动前清理同名容器，避免端口占用或容器残留问题。

测试代码无需修改，直接运行即可自动完成容器启动、测试执行、容器清理的全流程。

内容的提问来源于stack exchange，提问作者sgtStrom