113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
"""
|
||
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 для тестового аккаунта."""
|
||
return {
|
||
"username": os.getenv("KWORK_USERNAME"),
|
||
"password": os.getenv("KWORK_PASSWORD"),
|
||
}
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def require_credentials(kwork_credentials):
|
||
"""Пропускает тест если нет credentials."""
|
||
if not kwork_credentials["username"] or not kwork_credentials["password"]:
|
||
pytest.skip(
|
||
"E2E credentials not set. "
|
||
"Copy tests/e2e/.env.example to tests/e2e/.env and fill in 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."""
|
||
slowmo = request.config.getoption("--slowmo", default=0)
|
||
if slowmo > 0:
|
||
import time
|
||
|
||
time.sleep(slowmo)
|
||
|
||
|
||
def pytest_configure(config):
|
||
"""Регистрация маркера e2e."""
|
||
config.addinivalue_line("markers", "e2e: mark test as end-to-end (requires credentials)")
|
||
|
||
|
||
def pytest_addoption(parser):
|
||
"""Добавляет опцию --slowmo."""
|
||
parser.addoption(
|
||
"--slowmo", type=float, default=0, help="Delay between tests in seconds (for rate limiting)"
|
||
)
|