Как передать параметры для тестирования асинхронных функций?

У меня есть готовое API для тестирования, но когда я передаю через fixture cвое API, то возникает ошибка с асинхронным генератором.

сonftest.py

import pickle

import pytest
import aioredis

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from src.api import MoodleApi, ASUApi, PortalDPUApi
from src.conf.const import (
    DB_HOST, DB_USER, DB_PASSWORD,
    TEST_TELEGRAM_ID
    
)
from src.database.models import Base

TEST_DATABASE_URL = f'mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}/test'

engine = create_engine(url=TEST_DATABASE_URL)


@pytest.fixture(scope='session')
def db_engine():
    Base.metadata.create_all(engine)

    yield engine.connect()

    Base.metadata.drop_all(engine)


@pytest.fixture
def session(db_engine):
    session_local = sessionmaker(
        bind=db_engine, expire_on_commit=False
    )()

    db_engine.begin()

    yield session_local

    db_engine.rollback()

    session_local.close()




@pytest.fixture(scope='session')
async def api():
    async with Api(int(TEST_TELEGRAM_ID)) as client:
        yield client

test_api.py

import pytest

from src.conf.const import TEST_USER_LOGIN, TEST_USER_PASSWORD


class TestAPI:
    @pytest.mark.asyncio
    async def test_auth(self, api):
        token = await api.login(TEST_USER_LOGIN, TEST_USER_PASSWORD)

        assert token is not None
        assert isinstance(token, str)

    @pytest.mark.asyncio
    async def test_profile(self, api):

        profile = await api.get_profile()

        assert isinstance(profile, dict)

        username = profile.get('username')
        email = profile.get('email')

        assert username is not None
        assert email is not None

    @pytest.mark.asyncio
    async def test_journal(self, api):

        journal = await api.get_journal()

        assert isinstance(journal, list)

        lesson = journal[0]

        assert isinstance(lesson, dict)

        lesson_name = lesson.get('name')
        lesson_id = lesson.get('id')

        assert lesson_name is not None and lesson_id is not None

    @pytest.mark.asyncio
    async def test_week_schedule_for_student(self, api):

        week_schedule = await api.get_week_schedule_for_student()

        assert isinstance(week_schedule, list)

api.py

class Api:

    def __init__(self, telegram_id: Optional[int] = None):
        self._telegram_id = telegram_id
        self._client = ClientSession()

    async def _make_request(
            self, method: str, path: str, params: Optional[Dict] = None,
            json: Optional[Dict] = None, authorized: bool = False
    ) -> Union[List, Dict]:
        """Робимо запит до API ASU.

        Параметри:
            method: метод запиту;
            path: шлях запиту;
            params: параметри запиту;
            json: дані запиту;
            authorized: вказує чи повинна бути авторизація за допомогою токену;
        """

        if params is None:
            params = {}

        if json is None:
            json = {}

        headers = {}

        if authorized:
            token = await self._get_auth_token()
            if not token:
                raise AuthenticationError('Немає токена авторизації')
            headers = {
                 "Authorization": f'Bearer {token}'
            }

        try:
            response = await self._client.request(
                method=method, url=f'{ASU_URL_API}/{path}',
                json=json, params=params,
                headers=headers
            )
        except (ClientConnectionError, ClientConnectorError) as error:
            raise ApiConnectionError(f'Проблема зі з\'єднанняm:  {error}')

        if response.status != 200:
            raise ServerError(f'Сервер повертає статус - {response.status}, деталі: {response.text}')

        return await response.json()

 ...

 async def close_session(self):
        await self._client.close()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close_session()

Проблема

FAILED test_api.py::TestAPI::test_auth - AttributeError: 'async_generator' object has no attribute 'login'
FAILED test_api.py::TestAPI::test_profile - AttributeError: 'async_generator' object has no attribute 'get_profile'
FAILED test_api.py::TestAPI::test_journal - AttributeError: 'async_generator' object has no attribute 'get_journal'
FAILED test_api.py::TestASUAPI::test_week_schedule_for_student - AttributeError: 'async_generator' object has no attribute 'get_week_schedule_for_student'

Может кто-то знает, как правильно это нужно делать?


Ответы (1 шт):

Автор решения: Isaac Azimov

Это async generator:

async def ...:
    yield

Чтобы им пользоваться нужно его итерировать или вызвать next().

Я не совсем понял почему в этой функции Api в контексте менеджере. Но если можно без with, то вот так должно работать:

@pytest.fixture(scope='session')
async def api():
    return Api(int(TEST_TELEGRAM_ID))

А контекстный менеджер можно открывать внутри тестовый функций.

→ Ссылка