"""Comprehensive tests for Dashboard stats endpoint.""" import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from tests.conftest import auth_header from app.models.ticket import Ticket from app.models.instance import Instance from app.models.user import User from app.models.audit_log import AuditLog from app.core.security import hash_password BASE = "/api/v1/dashboard/stats" class TestDashboard: async def test_get_stats(self, client: AsyncClient, admin_token): resp = await client.get(BASE, headers=auth_header(admin_token)) assert resp.status_code == 200 data = resp.json() assert "ticket_total" in data assert "instance_total" in data assert "user_total" in data async def test_unauthenticated(self, client: AsyncClient): resp = await client.get(BASE) assert resp.status_code in (401, 403) # ─── DETAILED STATS TESTS ────────────────────────────────────────────── @pytest.mark.asyncio async def test_stats_all_zero_on_empty_db(client: AsyncClient, admin_token: str): """Fresh DB with only the admin user should show zeros except user_total=1.""" resp = await client.get(BASE, headers=auth_header(admin_token)) assert resp.status_code == 200 data = resp.json() assert data["ticket_total"] == 0 assert data["ticket_pending"] == 0 assert data["ticket_month"] == 0 assert data["ticket_executing"] == 0 assert data["instance_total"] == 0 assert data["instance_online"] == 0 assert data["user_total"] == 1 # admin_user from fixture assert data["queries_today"] == 0 @pytest.mark.asyncio async def test_stats_fields_structure(client: AsyncClient, admin_token: str): """Verify all expected fields are present and are integers.""" resp = await client.get(BASE, headers=auth_header(admin_token)) data = resp.json() expected_keys = { "ticket_total", "ticket_pending", "ticket_month", "ticket_executing", "instance_total", "instance_online", "user_total", "queries_today", "query_stats", } assert set(data.keys()) == expected_keys for key in expected_keys: if key == "query_stats": assert isinstance(data[key], (dict, type(None))), f"{key} should be dict or None" else: assert isinstance(data[key], int), f"{key} should be int" @pytest.mark.asyncio async def test_stats_reflect_ticket_counts( client: AsyncClient, db: AsyncSession, admin_token: str, test_instance: Instance ): """Create tickets with various statuses and verify counts.""" for status in ["pending", "reviewing", "executing", "success"]: ticket = Ticket( title=f"工单-{status}", db_type="mysql", instance_id=test_instance.id, target_database="test_db", sql_content="SELECT 1", status=status, applicant_id=1, ) db.add(ticket) await db.commit() resp = await client.get(BASE, headers=auth_header(admin_token)) data = resp.json() assert data["ticket_total"] == 4 # pending + reviewing = 2 assert data["ticket_pending"] == 2 assert data["ticket_executing"] == 1 # all 4 are within the last month assert data["ticket_month"] == 4 @pytest.mark.asyncio async def test_stats_reflect_instance_counts( client: AsyncClient, db: AsyncSession, admin_token: str ): """Create instances with different statuses.""" for i, status in enumerate(["connected", "disconnected", "connected"]): inst = Instance( name=f"inst-{i}", db_type="mysql", host=f"127.0.0.{i}", port=3306, username="root", password="pw", database="db", env="dev", status=status, ) db.add(inst) await db.commit() resp = await client.get(BASE, headers=auth_token_or_admin(admin_token)) data = resp.json() assert data["instance_total"] == 3 assert data["instance_online"] == 2 # connected count @pytest.mark.asyncio async def test_stats_reflect_user_count( client: AsyncClient, db: AsyncSession, admin_token: str ): """Add an extra active user and verify user_total.""" extra_user = User( username="extra", display_name="额外用户", role="developer", hashed_password="fake_hash", is_active=True, ) db.add(extra_user) await db.commit() resp = await client.get(BASE, headers=auth_header(admin_token)) data = resp.json() # admin + extra = 2 assert data["user_total"] == 2 @pytest.mark.asyncio async def test_stats_excludes_inactive_users( client: AsyncClient, db: AsyncSession, admin_token: str ): """Inactive users should not be counted in user_total.""" inactive = User( username="inactive", display_name="禁用用户", role="developer", hashed_password="fake_hash", is_active=False, ) db.add(inactive) await db.commit() resp = await client.get(BASE, headers=auth_header(admin_token)) data = resp.json() # Only admin (active), inactive user should be excluded assert data["user_total"] == 1 @pytest.mark.asyncio async def test_stats_reflect_queries_today( client: AsyncClient, db: AsyncSession, admin_token: str ): """Add query histories and verify queries_today counts from QueryHistory.""" from app.models.query_history import QueryHistory for i in range(5): qh = QueryHistory( user_id=1, username="admin", instance_id=1, instance_name="test", db_type="mysql", database_name="test_db", query_content="SELECT 1", query_type="SELECT", execution_time_ms=100, result_rows=1, is_success=True, ) db.add(qh) await db.commit() resp = await client.get(BASE, headers=auth_header(admin_token)) data = resp.json() assert data["queries_today"] == 5 @pytest.mark.asyncio async def test_stats_developer_can_access(client: AsyncClient, developer_token: str): """Dashboard stats should be accessible to any authenticated user.""" resp = await client.get(BASE, headers=auth_header(developer_token)) assert resp.status_code == 200 @pytest.mark.asyncio async def test_stats_guest_can_access(client: AsyncClient, guest_token: str): """Dashboard stats should be accessible to any authenticated user.""" resp = await client.get(BASE, headers=auth_header(guest_token)) assert resp.status_code == 200 @pytest.mark.asyncio async def test_stats_no_auth_returns_401(client: AsyncClient): resp = await client.get(BASE) assert resp.status_code == 401 # ─── Helper ──────────────────────────────────────────────────────────── def auth_token_or_admin(token: str) -> dict: """Alias for auth_header to avoid confusion in test body.""" return auth_header(token)