From 44ac6dfee3409070f518e9a3e31abe9f5be769bd Mon Sep 17 00:00:00 2001 From: root Date: Mon, 30 Mar 2026 00:13:42 +0000 Subject: [PATCH] test: revert to login per test (pytest-asyncio compatibility) Note: pytest-asyncio creates new event loop per test, so we cannot reuse KworkClient across tests. Each test logs in independently. This is acceptable because: 1. Login is fast (<1s) 2. Tests are independent (no shared state) 3. Auth tests verify login works correctly 4. Catalog tests verify API endpoints work --- tests/e2e/conftest.py | 56 ------------ tests/e2e/test_catalog.py | 186 ++++++++++++++++++++++++-------------- 2 files changed, 118 insertions(+), 124 deletions(-) diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 0a80b33..1af71d7 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -4,34 +4,16 @@ E2E тесты для Kwork API. Требуют реальных credentials и запускаются только локально. """ -import asyncio import os from pathlib import Path import pytest from dotenv import load_dotenv -from kwork_api import KworkClient - # Загружаем .env load_dotenv(Path(__file__).parent / ".env") -class E2EClient: - """Wrapper for KworkClient that manages event loop lifecycle.""" - - def __init__(self, client: KworkClient): - self.client = client - self.loop = asyncio.new_event_loop() - - def __getattr__(self, name): - return getattr(self.client, name) - - def close(self): - self.loop.run_until_complete(self.client.close()) - self.loop.close() - - @pytest.fixture(scope="session") def kwork_credentials(): """Credentials для тестового аккаунта.""" @@ -52,44 +34,6 @@ def require_credentials(kwork_credentials): return kwork_credentials -@pytest.fixture(scope="module") -def e2e_client(require_credentials): - """ - E2E клиент - логинится ОДИН РАЗ для всех тестов в модуле. - - Используется во всех тестах кроме test_auth.py (там тестируем сам логин). - """ - loop = asyncio.new_event_loop() - - async def create(): - return await KworkClient.login( - username=require_credentials["username"], - password=require_credentials["password"], - ) - - client = loop.run_until_complete(create()) - wrapper = E2EClient(client) - yield wrapper - wrapper.close() - loop.close() - - -@pytest.fixture(scope="module") -def catalog_kwork_id(e2e_client): - """ - Получить ID первого кворка из каталога. - - Выполняется ОДИН РАЗ в начале модуля и переиспользуется. - """ - async def get(): - catalog = await e2e_client.catalog.get_list(page=1) - if len(catalog.kworks) > 0: - return catalog.kworks[0].id - return None - - return e2e_client.loop.run_until_complete(get()) - - @pytest.fixture(scope="function") def slowmo(request): """Задержка между тестами для rate limiting.""" diff --git a/tests/e2e/test_catalog.py b/tests/e2e/test_catalog.py index e0af9a5..57d7168 100644 --- a/tests/e2e/test_catalog.py +++ b/tests/e2e/test_catalog.py @@ -1,82 +1,115 @@ """ E2E тесты для каталога и проектов. -Используют module-scoped e2e_client fixture - логин ОДИН РАЗ для всех тестов модуля. +Каждый тест логинится самостоятельно (pytest-asyncio compatibility). Все тесты read-only - ничего не изменяют на сервере. Endpoints основаны на HAR анализе (mitmproxy + har-analyzer skill). """ import pytest +from kwork_api import KworkClient + @pytest.mark.e2e -async def test_get_catalog_list(e2e_client): +async def test_get_catalog_list(require_credentials): """E2E: Получить список кворков из каталога. HAR: POST https://api.kwork.ru/catalogMainv2 """ - # Первая страница каталога - catalog = await e2e_client.client.client.catalog.get_list(page=1) - - assert catalog is not None - # API может вернуть пустой список (это нормально) - if len(catalog.kworks) > 0: - # Проверка структуры первого кворка - first_kwork = catalog.kworks[0] - assert first_kwork.id is not None - assert first_kwork.title is not None - assert first_kwork.price is not None + client = await KworkClient.login( + username=require_credentials["username"], + password=require_credentials["password"], + ) + + try: + catalog = await client.catalog.get_list(page=1) + + assert catalog is not None + if len(catalog.kworks) > 0: + first_kwork = catalog.kworks[0] + assert first_kwork.id is not None + assert first_kwork.title is not None + assert first_kwork.price is not None + finally: + await client.close() @pytest.mark.e2e -async def test_get_kwork_details(e2e_client, catalog_kwork_id): +async def test_get_kwork_details(require_credentials): """E2E: Получить детали кворка. HAR: POST https://api.kwork.ru/getKworkDetails """ - # Пропускаем если каталог пустой - if catalog_kwork_id is None: - pytest.skip("Catalog is empty") - - # Получаем детали - details = await e2e_client.client.catalog.get_details(catalog_kwork_id) - - assert details is not None - assert details.id == catalog_kwork_id - assert details.title is not None - assert details.price is not None + client = await KworkClient.login( + username=require_credentials["username"], + password=require_credentials["password"], + ) + + try: + # Сначала получаем каталог чтобы найти реальный ID + catalog = await client.catalog.get_list(page=1) + + if len(catalog.kworks) == 0: + pytest.skip("Catalog is empty") + + kwork_id = catalog.kworks[0].id + + # Получаем детали + details = await client.catalog.get_details(kwork_id) + + assert details is not None + assert details.id == kwork_id + assert details.title is not None + assert details.price is not None + finally: + await client.close() @pytest.mark.e2e -async def test_get_projects_list(e2e_client): +async def test_get_projects_list(require_credentials): """E2E: Получить список проектов с биржи. HAR: POST https://api.kwork.ru/projects """ - projects = await e2e_client.client.projects.get_list(page=1) - - assert projects is not None - # Проекты могут быть пустыми - if len(projects.projects) > 0: - first_project = projects.projects[0] - assert first_project.id is not None - assert first_project.title is not None + client = await KworkClient.login( + username=require_credentials["username"], + password=require_credentials["password"], + ) + + try: + projects = await client.projects.get_list(page=1) + + assert projects is not None + if len(projects.projects) > 0: + first_project = projects.projects[0] + assert first_project.id is not None + assert first_project.title is not None + finally: + await client.close() @pytest.mark.e2e -async def test_get_user_info(e2e_client): +async def test_get_user_info(require_credentials): """E2E: Получить информацию о текущем пользователе. HAR: POST https://api.kwork.ru/user """ - user = await e2e_client.client.user.get_info() - assert user is not None - # API возвращает dict с данными пользователя - assert isinstance(user, dict) + client = await KworkClient.login( + username=require_credentials["username"], + password=require_credentials["password"], + ) + + try: + user = await client.user.get_info() + assert user is not None + assert isinstance(user, dict) + finally: + await client.close() @pytest.mark.e2e -async def test_get_reference_data(e2e_client): +async def test_get_reference_data(require_credentials): """E2E: Получить справочные данные (города, страны, фичи). HAR endpoints: @@ -85,47 +118,64 @@ async def test_get_reference_data(e2e_client): - POST https://api.kwork.ru/getAvailableFeatures - POST https://api.kwork.ru/getBadgesInfo """ - # Города (может вернуть пустой список) - cities = await e2e_client.client.reference.get_cities() - assert isinstance(cities, list) - - # Страны (может вернуть пустой список) - countries = await e2e_client.client.reference.get_countries() - assert isinstance(countries, list) - - # Фичи - features = await e2e_client.client.reference.get_features() - assert isinstance(features, list) - - # Бейджи - badges = await e2e_client.client.reference.get_badges_info() - assert isinstance(badges, list) + client = await KworkClient.login( + username=require_credentials["username"], + password=require_credentials["password"], + ) + + try: + cities = await client.reference.get_cities() + assert isinstance(cities, list) + + countries = await client.reference.get_countries() + assert isinstance(countries, list) + + features = await client.reference.get_features() + assert isinstance(features, list) + + badges = await client.reference.get_badges_info() + assert isinstance(badges, list) + finally: + await client.close() @pytest.mark.e2e -async def test_get_notifications(e2e_client): +async def test_get_notifications(require_credentials): """E2E: Получить уведомления. HAR: POST https://api.kwork.ru/notifications """ - notifications = await e2e_client.client.notifications.get_list() - assert notifications is not None - # Уведомления могут быть пустыми - assert hasattr(notifications, 'notifications') + client = await KworkClient.login( + username=require_credentials["username"], + password=require_credentials["password"], + ) + + try: + notifications = await client.notifications.get_list() + assert notifications is not None + assert hasattr(notifications, 'notifications') + finally: + await client.close() @pytest.mark.e2e -async def test_get_user_orders(e2e_client): +async def test_get_user_orders(require_credentials): """E2E: Получить заказы пользователя. HAR endpoints: - POST https://api.kwork.ru/payerOrders - POST https://api.kwork.ru/workerOrders """ - # Заказы как заказчик - payer_orders = await e2e_client.client.projects.get_payer_orders() - assert isinstance(payer_orders, list) - - # Заказы как исполнитель - worker_orders = await e2e_client.client.projects.get_worker_orders() - assert isinstance(worker_orders, list) + client = await KworkClient.login( + username=require_credentials["username"], + password=require_credentials["password"], + ) + + try: + payer_orders = await client.projects.get_payer_orders() + assert isinstance(payer_orders, list) + + worker_orders = await client.projects.get_worker_orders() + assert isinstance(worker_orders, list) + finally: + await client.close()