Тестирование сервера aiohttp с асинхронными устройствами - PullRequest
Новая
       77

Тестирование сервера aiohttp с асинхронными устройствами

2 голосов
/ 02 марта 2020

Я пытаюсь использовать pytest-aiohttp для тестирования моего REST API на основе aiohttp. Существует конечная точка / authenticate , которая возвращает токен аутентификации, и по существу всем остальным конечным точкам требуется токен в заголовке авторизации. У меня есть следующие тесты, которые отлично работают: 

from pytest import fixture

from agora_backend.server import get_app

NAME, PASSWORD = "clin123", "password123"


@fixture
def client(loop, aiohttp_client):
    return loop.run_until_complete(aiohttp_client(get_app))


async def test_patient_success(client):
    auth_response = await client.post("/authenticate",
        json={"user_name": NAME, "user_pass": PASSWORD})

    auth_token =  await auth_response.text()

    response = await client.get("/clinician/patients",
        headers={"Authorization": f"Bearer {auth_token}"})

    assert response.status == 200


async def test_session_id_success(client):
    auth_response = await client.post("/authenticate",
        json={"user_name": NAME, "user_pass": PASSWORD})

    auth_token =  await auth_response.text()

    response = await client.get("/clinician/session/1",
        headers={"Authorization": f"Bearer {auth_token}"})

    assert response.status == 200

Однако, это не очень DRY. Я хотел бы использовать один и тот же токен авторизации для всех тестов, поэтому я решил, что приспособление будет хорошим выбором. Поэтому я попробовал следующий код: 

from pytest import fixture

from agora_backend.server import get_app

NAME, PASSWORD = "clin123", "password123"


@fixture
def client(loop, aiohttp_client):
    return loop.run_until_complete(aiohttp_client(get_app))


@fixture
async def auth_token(client):
    auth_response = await client.post("/authenticate",
        json={"user_name": NAME, "user_pass": PASSWORD})

    return await auth_response.text()


async def test_patient_success(client, auth_token):
    response = await client.get("/clinician/patients",
        headers={"Authorization": f"Bearer {auth_token}"})

    assert response.status == 200


async def test_session_id_success(client, auth_token):
    response = await client.get("/clinician/session/1",
        headers={"Authorization": f"Bearer {auth_token}"})

    assert response.status == 200

Однако это приводит к исключению RuntimeError: Timeout context manager should be used inside a task. Я пытался посмотреть этот, казалось бы, связанный ответ , но я не уверен, что он полностью применим к моей ситуации. И если это так, я не совсем понимаю, как применить его к моему коду. Любая помощь будет оценена!

...