opsxsql/tests/conftest.py

"""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}"}