Как передать параметры для тестирования асинхронных функций?
У меня есть готовое 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))
А контекстный менеджер можно открывать внутри тестовый функций.