feat: 霓虹都市→白日梦境 + 渐变网格→云石纹理 — 两个白色浅色主题登录页重写

登录页风格优化与扩展/sql-platform/tests/conftest.py

@@ -0,0 +1,212 @@
+"""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}"}