FastAPI异步Pytest测试报错：数据库占用及AsyncClient复用问题

1. 解决数据库被占用的OperationalError #

问题原因 #

测试结束后SQLAlchemy连接池仍有活跃连接未释放，导致删除测试数据库时提示被其他用户占用。

解决方案 #

在删除测试数据库前，强制终止所有连接到该库的外部会话，同时用with语句管理连接确保自动释放资源。修改测试数据库操作代码：

if is_testing: 
    db_url = self._engine.url
    if db_url.host != "localhost":
        raise Exception("db host must be 'localhost' in test environment")
    except_schema_db_url = f"{db_url.drivername}://{db_url.username}@{db_url.host}"   
    schema_name = db_url.database         # test                                                    
    temp_engine = create_engine(except_schema_db_url, echo=echo, pool_recycle=pool_recycle, pool_pre_ping=True)
    
    with temp_engine.connect() as conn:
        conn = conn.execution_options(autocommit=True)  # 启用自动提交避免事务阻塞
        
        # 强制终止所有连接到测试库的外部会话（排除当前连接）
        conn.execute(f"""
            SELECT pg_terminate_backend(pid) 
            FROM pg_stat_activity 
            WHERE datname = '{schema_name}' 
              AND pid <> pg_backend_pid();
        """)
        
        try:
            conn.execute(f"DROP DATABASE {schema_name}")
            print(f"test db dropped! about to create {schema_name}")
        except ProgrammingError:
            print(f"could not drop the database, probably does not exist.")
        except OperationalError:
            print("could not drop database because it's being accessed by other users(psql prompt open?)")
        
        conn.execute(f"CREATE DATABASE {schema_name}")
        try:
            conn.execute(f"create user test with encrypted password 'test'")
        except:
            print("User already exist")
    
    temp_engine.dispose()

2. 解决AsyncClient不可调用的TypeError #

问题原因 #

测试代码错误地将AsyncClient实例当作函数调用（如client()），而非使用其HTTP方法（如get()、post()）。

解决方案 #

检查所有测试用例，确保调用AsyncClient的正确方法：

错误示例：

response = client()  # 错误：AsyncClient实例不可调用

正确示例：

response = await client.get("/api/endpoint")  # 正确：使用对应HTTP方法

3. 解决ClientState重复打开/关闭的错误 #

问题原因 #

会话级（scope="session"）的AsyncClient被多个测试复用，部分测试可能关闭客户端导致后续测试无法重新打开；且原fixture未使用异步装饰器适配AsyncClient特性。

解决方案 #

将client fixture作用域改为function，确保每个测试使用独立实例；同时用pytest_asyncio.fixture处理异步逻辑：

修改conftest.py的client fixture：

import pytest_asyncio
from httpx import AsyncClient

@pytest_asyncio.fixture(scope="function")
async def client(app):
    # 用async with管理客户端生命周期，测试后自动关闭
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

若使用异步SQLAlchemy，同步修改session fixture为异步：

@pytest_asyncio.fixture(scope="function", autouse=True)
async def session():
    async with db.session() as sess:
        yield sess
        await clear_all_table_data(session=sess, metadata=Base.metadata, except_tables=[])
        await sess.rollback()

async def clear_all_table_data(session: AsyncSession, metadata, except_tables: List[str] = None):
    await session.execute("SET session_replication_role = 'replica';")          
    for table in metadata.sorted_tables:
        if table.name not in except_tables:
            await session.execute(table.delete())
    await session.execute("SET session_replication_role = 'origin';")         
    await session.commit()

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