Pytest批量运行测试用例时sqlalchemy数据库连接异常终止问题

根本原因 #

1. pytest默认fixture作用域为function级别，每个测试函数执行完成后，都会执行db_connection中yield后的代码：删除测试数据库再重建。其中DROP DATABASE ... WITH (FORCE)命令会强制断开所有连接到该数据库的活跃连接，包括你的API服务内部持有的数据库连接池中的存量连接。
2. 单条测试执行时，测试结束后进程直接退出，不存在后续用例使用旧连接的场景，因此不会报错。批量执行时，第一个测试结束后删库重建，而API的连接池没有被重置，第二个测试调用API时复用了已经被强制断开的旧连接，就会触发AdminShutdown错误。

解决办法 #

方案1：重置API连接池（改动最小，推荐） #

在db_connection的teardown阶段，删库重建完成后增加API全局数据库连接池的重置逻辑，释放所有旧连接。假设你的API服务的全局engine存放在app.db.engine，示例代码如下：

@pytest.fixture
def db_connection(test_config, admin_db_connection):
    config = test_config()
    engine = sqlalchemy.create_engine(config.DB_URL)
    connection = engine.connect()
    yield connection
    connection.close()

    # clear database
    from psycopg2.extensions import AsIs  # handle SQL quoting

    with admin_db_connection.cursor() as curs:
        curs.execute("drop database %s with (force);", (AsIs(config.DB_NAME),))
        curs.execute(
            "create database %s template vtag_template;", (AsIs(config.DB_NAME),)
        )
    # 新增：重置API的连接池，销毁所有旧连接
    from your_app import app
    app.db.engine.dispose()

后续测试请求触发数据库操作时，API会自动创建新的有效连接。

方案2：调整fixture作用域 #

如果你的测试场景允许整个测试文件共用同一个数据库实例，可以把db_connection的作用域调整为module级别，整个测试文件只执行一次初始化和清理操作，避免中途删库断开连接：

@pytest.fixture(scope="module")
def db_connection(test_config, admin_db_connection):
    # 原有逻辑不变

方案3：改用事务回滚清理测试数据（更轻量稳定） #

放弃删库重建的重量级清理方式，改用事务回滚来撤销测试产生的所有数据修改，性能更高也不会触发连接断开问题：

@pytest.fixture
def db_connection(test_config):
    config = test_config()
    engine = sqlalchemy.create_engine(config.DB_URL)
    connection = engine.connect()
    # 开启顶层事务
    transaction = connection.begin()
    yield connection
    # 回滚事务，所有测试产生的修改都会被撤销
    transaction.rollback()
    connection.close()

注意事项 #

• 检查所有依赖数据库的fixture的事务处理逻辑，避免残留未提交/回滚的事务占用连接。
• 如果必须保留删库重建的逻辑，确保所有持有该测试数据库连接的对象（包括fixture内部连接、API连接池）都在删库前被正确关闭释放。

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