"""Integration test conftest — connects to real Go backend at localhost:9000""" from typing import AsyncGenerator import pytest import pytest_asyncio from httpx import AsyncClient BASE_URL = "http://localhost:9000" class DictObject(dict): """Dict that supports dot-notation access: obj.key == obj['key'].""" def __getattr__(self, key): try: return self[key] except KeyError: raise AttributeError(f"'{type(self).__name__}' has no attribute '{key}'") def __setattr__(self, key, value): self[key] = value # Token cache — avoid re-login for every test _token_cache: dict[str, str] = {} async def _get_token(client: AsyncClient, username: str, password: str = "admin123") -> str: resp = await client.post("/api/v1/auth/login", json={ "username": username, "password": password, }) if resp.status_code == 200: return resp.json()["access_token"] return "" async def _ensure_admin_exists(client: AsyncClient): """Ensure admin and dba users exist — recreate if deleted by tests.""" # Try login first admin_token = await _get_token(client, "admin") if admin_token: return admin_token # Admin doesn't exist — use dev1 to create it dev_token = await _get_token(client, "dev1") if not dev_token: return "" # Create admin user via API resp = await client.post("/api/v1/users", json={ "username": "admin", "display_name": "超级管理员", "role": "admin", "password": "admin123", }, headers={"Authorization": f"Bearer {dev_token}"}) if resp.status_code in (200, 201): pass # created elif resp.status_code == 400: # Username exists but password wrong — can't fix via API pass # Try login again return await _get_token(client, "admin") @pytest_asyncio.fixture async def client() -> AsyncGenerator[AsyncClient, None]: """Function-scoped client — fresh for each test, no event loop issues.""" async with AsyncClient(base_url=BASE_URL, timeout=30.0) as c: yield c @pytest_asyncio.fixture async def admin_token(client: AsyncClient) -> str: token = await _get_token(client, "admin") assert token, "Admin login failed" return token @pytest_asyncio.fixture async def dba_token(client: AsyncClient) -> str: token = await _get_token(client, "dba") if token: return token # Fallback return await _get_token(client, "admin") @pytest_asyncio.fixture async def developer_token(client: AsyncClient) -> str: token = await _get_token(client, "dev1") assert token, "Developer login failed" return token @pytest_asyncio.fixture async def guest_token(client: AsyncClient) -> str: token = await _get_token(client, "auditor") if token: return token pytest.skip("guest/auditor user not available") return "" # ══════════════════════════════════════════════════════════════ # Helper fixtures for non-integration test files # ══════════════════════════════════════════════════════════════ @pytest_asyncio.fixture async def admin_user(client: AsyncClient, admin_token: str) -> dict: resp = await client.get("/api/v1/auth/me", headers=auth_header(admin_token)) if resp.status_code == 200: return resp.json() return {"id": 1, "username": "admin", "role": "admin"} @pytest_asyncio.fixture async def disabled_user(client: AsyncClient, admin_token: str) -> dict: return DictObject({"id": 999, "username": "disabled_test_user", "role": "developer", "is_active": False}) @pytest_asyncio.fixture async def test_mysql_instance(client: AsyncClient, admin_token: str) -> dict: resp = await client.get("/api/v1/instances", headers=auth_header(admin_token)) if resp.status_code == 200: data = resp.json() items = data.get("items", data) if isinstance(data, dict) else data for inst in items: if inst.get("db_type") == "mysql": return inst return DictObject({"id": 3, "name": "测试MySQL", "db_type": "mysql", "host": "127.0.0.1", "port": 3306}) @pytest_asyncio.fixture async def test_redis_instance(client: AsyncClient, admin_token: str) -> dict: resp = await client.get("/api/v1/instances", headers=auth_header(admin_token)) if resp.status_code == 200: data = resp.json() items = data.get("items", data) if isinstance(data, dict) else data for inst in items: if inst.get("db_type") == "redis": return inst return DictObject({"id": 2, "name": "测试Redis", "db_type": "redis", "host": "127.0.0.1", "port": 6379}) @pytest_asyncio.fixture async def test_ticket(client: AsyncClient, admin_token: str, test_mysql_instance: dict) -> dict: resp = await client.post("/api/v1/tickets", json={ "title": "conftest测试工单", "db_type": "mysql", "instance_id": test_mysql_instance["id"], "target_database": "test", "sql_content": "SELECT 1", "reason": "conftest测试", }, headers=auth_header(admin_token)) if resp.status_code in (200, 201): return resp.json() return DictObject({"id": 1, "title": "conftest测试工单", "status": "pending"}) @pytest_asyncio.fixture async def test_instance(client: AsyncClient, admin_token: str) -> dict: """Get first MySQL test instance.""" resp = await client.get("/api/v1/instances", headers=auth_header(admin_token)) if resp.status_code == 200: data = resp.json() items = data.get("items", data) if isinstance(data, dict) else data for inst in items: if inst.get("db_type") == "mysql": return DictObject(inst) return DictObject({"id": 3, "name": "测试MySQL", "db_type": "mysql", "host": "127.0.0.1", "port": 3306}) @pytest_asyncio.fixture async def db(client: AsyncClient, admin_token: str) -> dict: """Database fixture — get first MySQL instance.""" resp = await client.get("/api/v1/instances", headers=auth_header(admin_token)) if resp.status_code == 200: data = resp.json() items = data.get("items", data) if isinstance(data, dict) else data for inst in items: if inst.get("db_type") == "mysql": return DictObject(inst) return DictObject({"id": 3, "name": "测试MySQL", "db_type": "mysql", "host": "127.0.0.1", "port": 3306}) @pytest_asyncio.fixture async def dba_user(client: AsyncClient, dba_token: str) -> dict: """Get DBA user info.""" resp = await client.get("/api/v1/auth/me", headers=auth_header(dba_token)) if resp.status_code == 200: return resp.json() return DictObject({"id": 2, "username": "dba", "role": "dba"}) @pytest_asyncio.fixture async def developer_user(client: AsyncClient, developer_token: str) -> dict: """Get developer user info.""" resp = await client.get("/api/v1/auth/me", headers=auth_header(developer_token)) if resp.status_code == 200: return resp.json() return DictObject({"id": 3, "username": "dev1", "role": "developer"}) @pytest_asyncio.fixture async def guest_user(client: AsyncClient, guest_token: str) -> dict: """Get guest/auditor user info.""" resp = await client.get("/api/v1/auth/me", headers=auth_header(guest_token)) if resp.status_code == 200: return resp.json() return DictObject({"id": 5, "username": "auditor", "role": "auditor"}) def auth_header(token: str) -> dict: return {"Authorization": f"Bearer {token}"}