如何使用TempCluster编写asyncpg的集成测试？

TempCluster是asyncpg内部用于测试的临时PostgreSQL集群工具，无需依赖外部数据库实例，适合编写能覆盖真实数据库交互逻辑的集成测试，帮你实现100%代码覆盖率。

前置条件 #

• 系统环境中已安装PostgreSQL二进制文件（如pg_ctl、initdb）
• 安装依赖包：

pip install asyncpg pytest pytest-asyncio

测试代码示例 #

1. 编写Pytest Fixtures

import asyncpg
import pytest
from asyncpg.testing import TempCluster

@pytest.fixture(scope="session")
async def temp_cluster():
    # 启动临时PostgreSQL集群
    cluster = TempCluster()
    await cluster.start()
    yield cluster
    # 测试会话结束后停止集群
    await cluster.stop()

@pytest.fixture
async def pg_connection(temp_cluster):
    # 连接到临时集群的默认数据库
    conn = await asyncpg.connect(
        user=temp_cluster.user,
        database="postgres",
        host=temp_cluster.host,
        port=temp_cluster.port
    )
    yield conn
    # 测试用例结束后关闭连接
    await conn.close()

2. 编写测试用例

async def test_create_table_and_fetch(pg_connection):
    # 创建测试表
    await pg_connection.execute('''
        CREATE TABLE test_users (
            id SERIAL PRIMARY KEY,
            name TEXT NOT NULL
        )
    ''')
    
    # 插入单条数据
    await pg_connection.execute('INSERT INTO test_users (name) VALUES ($1)', 'Bob')
    
    # 查询验证
    user = await pg_connection.fetchrow('SELECT * FROM test_users WHERE id = $1', 1)
    assert user['name'] == 'Bob'

async def test_executemany_operation(pg_connection):
    await pg_connection.execute('''
        CREATE TABLE test_scores (
            id SERIAL PRIMARY KEY,
            score INTEGER NOT NULL
        )
    ''')
    
    # 批量插入数据
    batch_data = [(85,), (90,), (95,)]
    await pg_connection.executemany('INSERT INTO test_scores (score) VALUES ($1)', batch_data)
    
    # 验证插入数量
    total = await pg_connection.fetchval('SELECT COUNT(*) FROM test_scores')
    assert total == 3

关键说明 #

• TempCluster的作用域设为session，可以避免每个测试用例都重复启动/销毁集群，大幅提升测试效率。
• 所有数据库操作都是在真实的临时PostgreSQL实例中执行，能完整覆盖asyncpg的调用逻辑，比Mock方式更贴近生产场景。
• Fixture会自动处理集群的启动/停止和连接的创建/关闭，无需手动清理资源。

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