106 lines
2.6 KiB
Python
106 lines
2.6 KiB
Python
"""Integration tests configuration and fixtures."""
|
|
|
|
import os
|
|
import pytest
|
|
import httpx
|
|
from fastapi.testclient import TestClient
|
|
from app.main import app
|
|
from app.dependencies import get_db_client
|
|
from app.interfaces.db_api_client import DBApiClient
|
|
|
|
|
|
|
|
TEST_DB_API_URL = os.getenv("TEST_DB_API_URL", "http://localhost:8081/api/v1")
|
|
TEST_LOGIN = os.getenv("TEST_LOGIN", "99999999")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def db_api_url():
|
|
"""DB API URL for integration tests."""
|
|
return TEST_DB_API_URL
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def test_login():
|
|
"""Test user login."""
|
|
return TEST_LOGIN
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def db_client(db_api_url):
|
|
"""Real DB API client for integration tests."""
|
|
client = DBApiClient(api_prefix=db_api_url, timeout=30.0)
|
|
yield client
|
|
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def client(db_api_url):
|
|
"""FastAPI test client with real DB API integration."""
|
|
|
|
def get_real_db_client():
|
|
return DBApiClient(api_prefix=db_api_url, timeout=30.0)
|
|
|
|
app.dependency_overrides[get_db_client] = get_real_db_client
|
|
|
|
with TestClient(app) as test_client:
|
|
yield test_client
|
|
|
|
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def auth_token(client, test_login):
|
|
"""Get authentication token for test user."""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
params={"login": test_login}
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
pytest.skip(f"Cannot authenticate test user: {response.status_code}")
|
|
|
|
return response.json()["access_token"]
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def auth_headers(auth_token):
|
|
"""Authorization headers with JWT token."""
|
|
return {"Authorization": f"Bearer {auth_token}"}
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def test_user_id(client, test_login):
|
|
"""Get test user ID."""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
params={"login": test_login}
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
pytest.skip(f"Cannot get test user ID: {response.status_code}")
|
|
|
|
return response.json()["user"]["user_id"]
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def clean_test_sessions(db_client, test_user_id):
|
|
"""Clean up test sessions after test."""
|
|
yield
|
|
|
|
|
|
try:
|
|
sessions = db_client.get_sessions(test_user_id, limit=200)
|
|
for session in sessions.sessions:
|
|
db_client.delete_session(test_user_id, session.session_id)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def pytest_configure(config):
|
|
"""Configure pytest for integration tests."""
|
|
config.addinivalue_line(
|
|
"markers", "integration: mark test as integration test (requires real DB API)"
|
|
)
|