258 lines
6.6 KiB
Python
258 lines
6.6 KiB
Python
"""
|
|
Integration tests with real Kwork API.
|
|
|
|
These tests require valid credentials and make real API calls.
|
|
Skip these tests in CI/CD or when running unit tests only.
|
|
|
|
Usage:
|
|
pytest tests/integration/ -m integration
|
|
|
|
Or with credentials:
|
|
KWORK_USERNAME=user KWORK_PASSWORD=pass pytest tests/integration/ -m integration
|
|
"""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from kwork_api import KworkAuthError, KworkClient
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client() -> KworkClient | None:
|
|
"""
|
|
Create authenticated client for integration tests.
|
|
|
|
Requires KWORK_USERNAME and KWORK_PASSWORD environment variables.
|
|
Skip tests if not provided.
|
|
"""
|
|
username = os.getenv("KWORK_USERNAME")
|
|
password = os.getenv("KWORK_PASSWORD")
|
|
|
|
if not username or not password:
|
|
pytest.skip("KWORK_USERNAME and KWORK_PASSWORD not set")
|
|
|
|
# Create client
|
|
import asyncio
|
|
|
|
async def create_client():
|
|
return await KworkClient.login(username, password)
|
|
|
|
return asyncio.run(create_client())
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestAuthentication:
|
|
"""Test authentication with real API."""
|
|
|
|
def test_login_with_credentials(self):
|
|
"""Test login with real credentials."""
|
|
username = os.getenv("KWORK_USERNAME")
|
|
password = os.getenv("KWORK_PASSWORD")
|
|
|
|
if not username or not password:
|
|
pytest.skip("Credentials not set")
|
|
|
|
import asyncio
|
|
|
|
async def login():
|
|
client = await KworkClient.login(username, password)
|
|
assert client._token is not None
|
|
assert "userId" in client._cookies
|
|
await client.close()
|
|
return True
|
|
|
|
result = asyncio.run(login())
|
|
assert result
|
|
|
|
def test_invalid_credentials(self):
|
|
"""Test login with invalid credentials."""
|
|
import asyncio
|
|
|
|
async def try_login():
|
|
try:
|
|
await KworkClient.login("invalid_user_12345", "wrong_password")
|
|
return False
|
|
except KworkAuthError:
|
|
return True
|
|
|
|
result = asyncio.run(try_login())
|
|
assert result # Should raise auth error
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestCatalogAPI:
|
|
"""Test catalog endpoints with real API."""
|
|
|
|
def test_get_catalog_list(self, client: KworkClient):
|
|
"""Test getting catalog list."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
async def fetch():
|
|
result = await client.catalog.get_list(page=1)
|
|
return result
|
|
|
|
result = asyncio.run(fetch())
|
|
|
|
assert result.kworks is not None
|
|
assert len(result.kworks) > 0
|
|
assert result.pagination is not None
|
|
|
|
def test_get_kwork_details(self, client: KworkClient):
|
|
"""Test getting kwork details."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
async def fetch():
|
|
# First get a kwork ID from catalog
|
|
catalog = await client.catalog.get_list(page=1)
|
|
if not catalog.kworks:
|
|
return None
|
|
|
|
kwork_id = catalog.kworks[0].id
|
|
details = await client.catalog.get_details(kwork_id)
|
|
return details
|
|
|
|
result = asyncio.run(fetch())
|
|
|
|
if result:
|
|
assert result.id is not None
|
|
assert result.title is not None
|
|
assert result.price is not None
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestProjectsAPI:
|
|
"""Test projects endpoints with real API."""
|
|
|
|
def test_get_projects_list(self, client: KworkClient):
|
|
"""Test getting projects list."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
async def fetch():
|
|
return await client.projects.get_list(page=1)
|
|
|
|
result = asyncio.run(fetch())
|
|
|
|
assert result.projects is not None
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestReferenceAPI:
|
|
"""Test reference data endpoints."""
|
|
|
|
def test_get_cities(self, client: KworkClient):
|
|
"""Test getting cities."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
async def fetch():
|
|
return await client.reference.get_cities()
|
|
|
|
result = asyncio.run(fetch())
|
|
|
|
assert isinstance(result, list)
|
|
# Kwork has many cities, should have at least some
|
|
assert len(result) > 0
|
|
|
|
def test_get_countries(self, client: KworkClient):
|
|
"""Test getting countries."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
result = asyncio.run(client.reference.get_countries())
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) > 0
|
|
|
|
def test_get_timezones(self, client: KworkClient):
|
|
"""Test getting timezones."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
result = asyncio.run(client.reference.get_timezones())
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) > 0
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestUserAPI:
|
|
"""Test user endpoints."""
|
|
|
|
def test_get_user_info(self, client: KworkClient):
|
|
"""Test getting current user info."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
result = asyncio.run(client.user.get_info())
|
|
|
|
assert isinstance(result, dict)
|
|
# Should have user data
|
|
assert result # Not empty
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestErrorHandling:
|
|
"""Test error handling with real API."""
|
|
|
|
def test_invalid_kwork_id(self, client: KworkClient):
|
|
"""Test getting non-existent kwork."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
async def fetch():
|
|
try:
|
|
await client.catalog.get_details(999999999)
|
|
return False
|
|
except Exception:
|
|
return True
|
|
|
|
asyncio.run(fetch())
|
|
# May or may not raise error depending on API behavior
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestRateLimiting:
|
|
"""Test rate limiting behavior."""
|
|
|
|
def test_multiple_requests(self, client: KworkClient):
|
|
"""Test making multiple requests."""
|
|
if not client:
|
|
pytest.skip("No client")
|
|
|
|
import asyncio
|
|
|
|
async def fetch_multiple():
|
|
results = []
|
|
for page in range(1, 4):
|
|
catalog = await client.catalog.get_list(page=page)
|
|
results.append(catalog)
|
|
# Small delay to avoid rate limiting
|
|
await asyncio.sleep(0.5)
|
|
return results
|
|
|
|
results = asyncio.run(fetch_multiple())
|
|
|
|
assert len(results) == 3
|
|
for result in results:
|
|
assert result.kworks is not None
|