"""Audit service 单元测试 (U4 — RBAC + deprovisioning 审计, KTD-8)。

验证：
- RBAC 角色变更审计记录（actor/target/role_before/role_after）
- Deprovisioning 审计记录（actor/target/reason/source）
- 审计记录持久化到 auth_audit_log 表
- 按目标用户查询审计历史
"""

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
from uuid import uuid4

import aiosqlite
import pytest

from agentkit.server.auth.audit import (
    AuditService,
    EVENT_DEPROVISION,
    EVENT_ROLE_CHANGE,
    SOURCE_BREAKGLASS,
    SOURCE_MANUAL,
    get_audit_service,
    set_audit_service,
)
from agentkit.server.auth.models import audit_log_row_to_dict, init_auth_db


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
async def auth_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    db_path = tmp_path / "auth.db"
    await init_auth_db(db_path)
    monkeypatch.setenv("AGENTKIT_AUTH_DB", str(db_path))
    return db_path


@pytest.fixture
def audit_service(auth_db: Path) -> AuditService:
    svc = AuditService(db_path=auth_db)
    set_audit_service(svc)
    return svc


@pytest.fixture
async def seed_users(auth_db: Path) -> dict[str, str]:
    """插入两个用户（admin + target）用于审计测试。"""
    now = datetime.now(timezone.utc).isoformat()
    admin_id = str(uuid4())
    target_id = str(uuid4())
    async with aiosqlite.connect(str(auth_db)) as db:
        await db.execute(
            "INSERT INTO users (id, username, email, password_hash, role, is_active, "
            "is_terminal_authorized, is_server_terminal_authorized, created_at, updated_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (admin_id, "admin_alice", "alice@example.com", "hash", "admin", 1, 0, 0, now, now),
        )
        await db.execute(
            "INSERT INTO users (id, username, email, password_hash, role, is_active, "
            "is_terminal_authorized, is_server_terminal_authorized, created_at, updated_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (target_id, "target_bob", "bob@example.com", "hash", "member", 1, 0, 0, now, now),
        )
        await db.commit()
    return {"admin_id": admin_id, "target_id": target_id}


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------


class TestRoleChangeAudit:
    async def test_records_role_change(
        self, audit_service: AuditService, seed_users: dict[str, str]
    ) -> None:
        record = await audit_service.record_role_change(
            actor_user_id=seed_users["admin_id"],
            actor_username="admin_alice",
            target_user_id=seed_users["target_id"],
            target_username="target_bob",
            role_before="member",
            role_after="operator",
            reason="晋升",
            source=SOURCE_MANUAL,
        )
        assert record["event_type"] == EVENT_ROLE_CHANGE
        assert record["role_before"] == "member"
        assert record["role_after"] == "operator"
        assert record["actor_username"] == "admin_alice"
        assert record["target_username"] == "target_bob"
        assert record["reason"] == "晋升"

    async def test_persists_to_db(
        self, audit_service: AuditService, auth_db: Path, seed_users: dict[str, str]
    ) -> None:
        await audit_service.record_role_change(
            actor_user_id=seed_users["admin_id"],
            actor_username="admin_alice",
            target_user_id=seed_users["target_id"],
            target_username="target_bob",
            role_before="member",
            role_after="admin",
            reason="提升管理员",
        )
        async with aiosqlite.connect(str(auth_db)) as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute(
                "SELECT * FROM auth_audit_log WHERE event_type = ? AND target_user_id = ?",
                (EVENT_ROLE_CHANGE, seed_users["target_id"]),
            )
            rows = await cursor.fetchall()
        assert len(rows) == 1
        row = audit_log_row_to_dict(rows[0])
        assert row["role_before"] == "member"
        assert row["role_after"] == "admin"


class TestDeprovisionAudit:
    async def test_records_deprovision(
        self, audit_service: AuditService, seed_users: dict[str, str]
    ) -> None:
        record = await audit_service.record_deprovision(
            actor_user_id=seed_users["admin_id"],
            actor_username="admin_alice",
            target_user_id=seed_users["target_id"],
            target_username="target_bob",
            reason="离职",
            source=SOURCE_MANUAL,
        )
        assert record["event_type"] == EVENT_DEPROVISION
        assert record["reason"] == "离职"
        assert record["source"] == SOURCE_MANUAL

    async def test_break_glass_source(
        self, audit_service: AuditService, seed_users: dict[str, str]
    ) -> None:
        """break_glass 应急通道 — source 记录为 break_glass。"""
        record = await audit_service.record_deprovision(
            actor_user_id=seed_users["admin_id"],
            actor_username="admin_alice",
            target_user_id=seed_users["target_id"],
            target_username="target_bob",
            reason="应急禁用",
            source=SOURCE_BREAKGLASS,
        )
        assert record["source"] == SOURCE_BREAKGLASS

    async def test_persists_to_db(
        self, audit_service: AuditService, auth_db: Path, seed_users: dict[str, str]
    ) -> None:
        await audit_service.record_deprovision(
            actor_user_id=seed_users["admin_id"],
            actor_username="admin_alice",
            target_user_id=seed_users["target_id"],
            target_username="target_bob",
            reason="审计测试",
        )
        async with aiosqlite.connect(str(auth_db)) as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute(
                "SELECT * FROM auth_audit_log WHERE event_type = ?",
                (EVENT_DEPROVISION,),
            )
            rows = await cursor.fetchall()
        assert len(rows) == 1


class TestListForTarget:
    async def test_lists_audit_history(
        self, audit_service: AuditService, seed_users: dict[str, str]
    ) -> None:
        """按目标用户查询审计历史（admin 视图）。"""
        target_id = seed_users["target_id"]
        await audit_service.record_role_change(
            actor_user_id=seed_users["admin_id"],
            actor_username="admin_alice",
            target_user_id=target_id,
            target_username="target_bob",
            role_before="member",
            role_after="operator",
            reason="第一次",
        )
        await audit_service.record_deprovision(
            actor_user_id=seed_users["admin_id"],
            actor_username="admin_alice",
            target_user_id=target_id,
            target_username="target_bob",
            reason="第二次",
        )
        records = await audit_service.list_for_target(target_id)
        assert len(records) == 2
        # 按 created_at DESC 排序 — deprovision 在前
        assert records[0]["event_type"] == EVENT_DEPROVISION
        assert records[1]["event_type"] == EVENT_ROLE_CHANGE


class TestGetAuditService:
    def test_singleton(self) -> None:
        svc1 = get_audit_service()
        svc2 = get_audit_service()
        assert svc1 is svc2