带有 sqlalchemy db 连接的 Pytest 在后续测试用例中终止答案

【问题标题】：Pytest with sqlalchemy db conection terminates in subsequent test case带有 sqlalchemy db 连接的 Pytest 在后续测试用例中终止
【发布时间】：2021-10-27 17:04:34
【问题描述】：

我使用 sqlalchemy 引擎连接作为 pytest 夹具来测试需要数据库交互的函数。当我通过运行pytest tests/filename.py::test_get_token_data 或pytest tests/filename.py::test_create 单独运行测试用例时，测试通过；但是，在运行整个测试用例 pytest tests/filename.py 时，出现以下错误：

E       sqlalchemy.exc.OperationalError: (psycopg2.errors.AdminShutdown) terminating connection due to administrator command
E       server closed the connection unexpectedly
E               This probably means the server terminated abnormally
E               before or while processing the request.
(Background on this error at: http://sqlalche.me/e/13/e3q8)

据我了解，每次测试后，数据库都应该被清除（我已经确认），但是我不明白这个错误。我的代码如下。

conftest.py

@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),)
        )

文件名.py

import sqlalchemy as sa


@pytest.fixture
def db_injection(db_connection):
    with db_connection.begin():
        some_value = db_connection.execute(
            sa.sql.text(
                """insert into general.some_table as t (some_id, name, description, is_active) values (:some_id, :name, :description, :is_active) returning t.some_id;

                """
            ),
            tenant_id='1234',
            description="TEST",
            is_active=True,
            name="TEST",
        )
        tenant_id_ = some_value.first()[0]
         
                
@pytest.fixture
def some_function(db_connection):

    with db_connection.begin():
        some_table = db_connection.execute(
            sa.sql.text(
                """
                SELECT e.name
                FROM general.some_table e
                WHERE e.id = 1234
            """
            )
        )
    return some_table.first()


def test_get_token_data(client, headers_with_token, db_injection, some_function):

    token = some_function.name
    #API is using the same db connection 
    response = client.get(f"/api/v2/{token}", headers=headers_with_token)

    assert response.status_code == 200


def test_create(client, img, b_64, headers_with_token, db_injection):

    items = Items(
        user_id=1234,
        template_id=67,
        b_64=b_64,
    )

    data = {
        "json": json.dumps(asdict(items)),
        "file": ("some_name", img, "multipart/form-data"),
    }

    response = client.post(
        "/api/v2/create",
        data=data,
        follow_redirects=True,
        content_type="multipart/form-data",
        headers=headers_with_token
    )

    assert response.status_code == 200

【问题讨论】：

标签： python postgresql sqlalchemy pytest

【解决方案1】：

问题是由于应用程序使用了未绑定的Session。由于我们的大多数应用程序都使用 sqlalchemy 的 Engine API，因此我们决定通过 with engine.begin() 重构使用原始 sql。另一种替代解决方案可能是添加 engine into the session

【讨论】：