Используйте несколько AsyncClients с Pytest для интеграционного тестирования

Я хочу использовать два разных httpx-клиента, один из которых взаимодействует с реальной базой данных, а другой — с тестовой базой данных. Но по какой-то причине все транзакции происходят в реальной базе данных.

# conftest.py
@pytest.fixture(scope = "session")
def anyio_backend():
    return "asyncio"


@pytest.fixture(scope = "function")
async def real_client():
    async with LifespanManager(app):
        async with AsyncClient(
            transport=ASGITransport(app), base_url = "http://localhost"
        ) as c:
            yield c


async def init_db(db_url, create_db: bool = False, schemas: bool = False) -> None:
    """Initial database connection"""
    await Tortoise.init(
        db_url=db_url, modules = {"models": ["app.database.models"]}, _create_db=create_db
    )
    if create_db:
        print(f"Database created! {db_url = }")
    if schemas:
        await Tortoise.generate_schemas()
        print("Success to generate schemas")


async def init():
    DB_URL = "sqlite://:memory:"
    await init_db(DB_URL, True, True)

    dashboard_permission_model = CreatePermission(
        name = "Dashboard", description = "Dashboard Page"
    )
    camera_permission_model = CreatePermission(name = "Camera", description = "Camera Page")
    await Permission.create(**dashboard_permission_model.dict())
    await Permission.create(**camera_permission_model.dict())

    admin_role_model = CreateRole(name = "Admin", description = "Admin dashboard")

    user = await User.create(
        first_name = "Admin",
        last_name = "",
        email=ADMIN_EMAIL,
        password=get_password_hash(password = "some@password"),
    )

    admin_role = await Role.create(**admin_role_model.dict())
    await UserRole.create(role=admin_role, user=user)

    # Add all permission to admin
    all_permissions = await Permission.all()

    for permission in all_permissions:
        await UserPermission.create(permission=permission, user=user)


@pytest.fixture(scope = "function")
async def test_client():
    async with AsyncClient(app=ASGITransport(app), base_url = "http://test") as client:
        yield client


@pytest.fixture(scope = "session", autouse=True)
async def initialize_tests():
    await init()
    yield
    await Tortoise._drop_databases()

Ниже тестируется роль маршрутизатора с использованием тестового клиента (тестовой базы данных):

# test_role.py
@pytest.mark.anyio
async def test_creat_role(
    test_client: AsyncClient, auth_headers: dict[str, str], role_data
):
    response = await test_client.post(
        url = "/role/add", headers=auth_headers, json=role_data
    )

    logging.debug(response)

    assert response.status_code == 200
    assert response.json()["name"] == "Test role"
    assert response.json()["description"] == "Some test role"

Ниже тест графического маршрутизатора с использованием реального клиента (фактической базы данных):

# test_graph.py
@pytest.mark.anyio
@pytest.mark.parametrize(
    "camera_ids, start_time, end_time, location_ids, time_frame",
    [
        ([-8, -10], "2024-04-01", "2024-04-01", [], "day"),  # Invalid camera ids
    ],
)
async def test_dwell_time_and_trends_invalid_camera_ids(
    real_client: AsyncClient,
    auth_headers: dict[str, str],
    current_user: int,
    camera_ids: list[int],
    start_time: str,
    end_time: str,
    location_ids: list[int],
    time_frame: str,
):
    store_ids = await get_store_ids(current_user)

    response = await real_client.get(
        url = "/dwell-time-and-trends",
        headers=auth_headers,
        params = {
            "store_ids": store_ids,
            "camera_ids": camera_ids,
            "start_time": start_time,
            "end_time": end_time,
            "location_ids": location_ids,
            "time_frame": time_frame,
        },
    )

    logging.debug(response.content)

    assert response.status_code == 500
    assert response.json()["detail"] == "list index out of range"

Это работает, когда я игнорирую тестовые примеры, в которых используется приспособление real_client, в котором создается локальная база данных и все транзакции происходят в ней. Но когда я запускаю все тесты, все транзакции происходят в реальной базе данных.

У меня есть некоторая теория, возможно, это связано с серверной частью Anyio или конфликтуют клиенты в одном сеансе. Я заметил, что тот клиент, который вызывается первым, является постоянным и используется на протяжении всего сеанса.

Можете ли вы также поделиться своими тестовыми кодами? Спасибо

volkan 29.04.2024 14:20

@volkan, я обновил описание примерами тестов как для реального, так и для тестового клиента. Кстати, для вашей информации я использую черепаху-орм.

Misto Shiawase 29.04.2024 15:00
Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
0
2
79
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Вы можете попробовать изменить область действия real_client с сеанса на модуль и связанные с ним приспособления для вашего случая, current_user и т. д. А объединение приспособления Initialize_test() с приспособлением test_client гарантирует, что транзакции, связанные с тестовой базой данных, происходят только при использовании test_client.

Убедитесь, что real_client работает на уровне модуля, и только необходимые модули будут использовать real_client, а остальная часть теста сможет использовать test_client для всего сеанса.

Ваш файл conftest.py

@pytest.fixture(scope = "session")
def anyio_backend():
    return "asyncio"


@pytest.fixture(scope = "module")
async def real_client():
    async with LifespanManager(app):
        async with AsyncClient(
            transport=ASGITransport(app), base_url = "http://localhost"
        ) as c:
            yield c


@pytest.fixture(scope = "module")
async def user_email():
    return await User.filter(id=1).first()


@pytest.fixture(scope = "module")
async def current_user(user_email):
    return await User.filter(email=user_email).first()


@pytest.fixture(scope = "module")
def auth_headers(user_email):
    refresh_token = create_refresh_token(user_email)

    return {"Authorization": f"Bearer {refresh_token}"}


async def init_db(db_url, create_db: bool = False, schemas: bool = False) -> None:
    """Initial database connection"""
    await Tortoise.init(
        db_url=db_url, modules = {"models": ["app.database.models"]}, _create_db=create_db
    )
    if create_db:
        print(f"Database created! {db_url = }")
    if schemas:
        await Tortoise.generate_schemas()
        print("Success to generate schemas")


async def init():
    DB_URL = "sqlite://:memory:"
    await init_db(DB_URL, True, True)

    dashboard_permission_model = CreatePermission(
        name = "Dashboard", description = "Dashboard Page"
    )
    camera_permission_model = CreatePermission(name = "Camera", description = "Camera Page")
    await Permission.create(**dashboard_permission_model.dict())
    await Permission.create(**camera_permission_model.dict())

    admin_role_model = CreateRole(name = "Admin", description = "Admin dashboard")

    user = await User.create(
        first_name = "Admin",
        last_name = "",
        email=ADMIN_EMAIL,
        password=get_password_hash(password = "some@password"),
    )

    admin_role = await Role.create(**admin_role_model.dict())
    await UserRole.create(role=admin_role, user=user)

    # Add all permission to admin
    all_permissions = await Permission.all()

    for permission in all_permissions:
        await UserPermission.create(permission=permission, user=user)


@pytest.fixture(scope = "session")
async def test_client():
    await init()
    async with AsyncClient(transport=ASGITransport(app), base_url = "http://test") as client:
        yield client
    await Tortoise._drop_databases()

Надеюсь, что это работает для вас!

Другие вопросы по теме