From e591eb4583cf588d9976c11d531f0962c62f71f3 Mon Sep 17 00:00:00 2001
From: logan <admin@attacker.club>
Date: Sat, 23 May 2026 15:23:13 +0800
Subject: [PATCH] =?UTF-8?q?test:=20announcement+dashboard=E6=B5=8B?=
 =?UTF-8?q?=E8=AF=95=E8=A6=86=E7=9B=96=20=E2=80=94=2036=E6=96=B0=E6=B5=8B?=
 =?UTF-8?q?=E8=AF=95=20(843=E2=86=92877)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

test_announcement.py (24 tests):
- CRUD: 创建(空标题/过长/缺字段), 列表(空/排序/认证), 更新(标题/内容/发布/取消), 删除(404)
- 权限: dba/developer/guest各角色操作限制 (403), 未认证(401)
- 边界: 完整生命周期, 重复发布幂等

test_dashboard.py (12 new, total 14):
- 空数据库零计数, 返回8字段验证
- 创建工单/实例/用户后计数验证
- inactive用户不计入, AuditLog仅计QUERY
- developer/guest访问, 未认证401
---
 backend/tests/test_announcement.py | 318 +++++++++++++++++++++++++++++
 backend/tests/test_dashboard.py    | 204 +++++++++++++++++-
 2 files changed, 520 insertions(+), 2 deletions(-)
 create mode 100644 backend/tests/test_announcement.py

diff --git a/backend/tests/test_announcement.py b/backend/tests/test_announcement.py
new file mode 100644
index 0000000..86331d4
--- /dev/null
+++ b/backend/tests/test_announcement.py
@@ -0,0 +1,318 @@
+"""Comprehensive tests for Announcement CRUD, permissions, and edge cases."""
+import pytest
+from httpx import AsyncClient
+
+from tests.conftest import auth_header
+
+BASE = "/api/v1/announcements"
+
+# ─── CREATE (POST /announcements) ──────────────────────────────────────
+
+PAYLOAD = {
+    "title": "系统维护通知",
+    "content": "将于本周六凌晨进行系统维护，届时服务将暂停。",
+}
+
+
+@pytest.mark.asyncio
+async def test_create_announcement(client: AsyncClient, admin_token: str):
+    resp = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    assert resp.status_code == 201
+    data = resp.json()
+    assert data["title"] == PAYLOAD["title"]
+    assert data["content"] == PAYLOAD["content"]
+    assert data["is_published"] is False
+    assert data["published_at"] is None
+    assert "id" in data
+    assert "created_at" in data
+    assert "updated_at" in data
+
+
+@pytest.mark.asyncio
+async def test_create_announcement_empty_title(client: AsyncClient, admin_token: str):
+    resp = await client.post(
+        BASE, json={"title": "", "content": "内容"}, headers=auth_header(admin_token)
+    )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_create_announcement_title_too_long(client: AsyncClient, admin_token: str):
+    resp = await client.post(
+        BASE, json={"title": "x" * 201, "content": "内容"}, headers=auth_header(admin_token)
+    )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_create_announcement_empty_content(client: AsyncClient, admin_token: str):
+    resp = await client.post(
+        BASE, json={"title": "标题", "content": ""}, headers=auth_header(admin_token)
+    )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_create_announcement_missing_fields(client: AsyncClient, admin_token: str):
+    resp = await client.post(BASE, json={}, headers=auth_header(admin_token))
+    assert resp.status_code == 422
+
+
+# ─── LIST (GET /announcements) ─────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_list_announcements_empty(client: AsyncClient, admin_token: str):
+    resp = await client.get(BASE, headers=auth_header(admin_token))
+    assert resp.status_code == 200
+    assert resp.json() == []
+
+
+@pytest.mark.asyncio
+async def test_list_announcements_returns_all(client: AsyncClient, admin_token: str):
+    for i in range(3):
+        await client.post(
+            BASE,
+            json={"title": f"公告{i}", "content": f"内容{i}"},
+            headers=auth_header(admin_token),
+        )
+    resp = await client.get(BASE, headers=auth_header(admin_token))
+    assert resp.status_code == 200
+    items = resp.json()
+    assert len(items) == 3
+    # ordered by id desc
+    assert items[0]["title"] == "公告2"
+    assert items[2]["title"] == "公告0"
+
+
+@pytest.mark.asyncio
+async def test_list_announcements_requires_auth(client: AsyncClient):
+    resp = await client.get(BASE)
+    assert resp.status_code == 401
+
+
+# ─── UPDATE (PUT /announcements/{id}) ──────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_update_announcement_title(client: AsyncClient, admin_token: str):
+    create = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    aid = create.json()["id"]
+
+    resp = await client.put(
+        f"{BASE}/{aid}",
+        json={"title": "更新后的标题"},
+        headers=auth_header(admin_token),
+    )
+    assert resp.status_code == 200
+    assert resp.json()["title"] == "更新后的标题"
+    # content unchanged
+    assert resp.json()["content"] == PAYLOAD["content"]
+
+
+@pytest.mark.asyncio
+async def test_update_announcement_content(client: AsyncClient, admin_token: str):
+    create = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    aid = create.json()["id"]
+
+    resp = await client.put(
+        f"{BASE}/{aid}",
+        json={"content": "新内容"},
+        headers=auth_header(admin_token),
+    )
+    assert resp.status_code == 200
+    assert resp.json()["content"] == "新内容"
+
+
+@pytest.mark.asyncio
+async def test_update_announcement_publish(client: AsyncClient, admin_token: str):
+    create = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    aid = create.json()["id"]
+    assert create.json()["is_published"] is False
+
+    resp = await client.put(
+        f"{BASE}/{aid}",
+        json={"is_published": True},
+        headers=auth_header(admin_token),
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["is_published"] is True
+    # published_at should be set on first publish
+    assert data["published_at"] is not None
+
+
+@pytest.mark.asyncio
+async def test_update_announcement_publish_idempotent(client: AsyncClient, admin_token: str):
+    """Publishing an already-published announcement should not change published_at."""
+    create = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    aid = create.json()["id"]
+
+    first = await client.put(
+        f"{BASE}/{aid}",
+        json={"is_published": True},
+        headers=auth_header(admin_token),
+    )
+    first_published_at = first.json()["published_at"]
+
+    # Update title while staying published
+    second = await client.put(
+        f"{BASE}/{aid}",
+        json={"title": "改标题不改发布时间"},
+        headers=auth_header(admin_token),
+    )
+    assert second.json()["published_at"] == first_published_at
+
+
+@pytest.mark.asyncio
+async def test_update_announcement_unpublish(client: AsyncClient, admin_token: str):
+    create = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    aid = create.json()["id"]
+
+    # publish first
+    await client.put(
+        f"{BASE}/{aid}",
+        json={"is_published": True},
+        headers=auth_header(admin_token),
+    )
+    # then unpublish
+    resp = await client.put(
+        f"{BASE}/{aid}",
+        json={"is_published": False},
+        headers=auth_header(admin_token),
+    )
+    assert resp.status_code == 200
+    assert resp.json()["is_published"] is False
+
+
+@pytest.mark.asyncio
+async def test_update_nonexistent_announcement(client: AsyncClient, admin_token: str):
+    resp = await client.put(
+        f"{BASE}/99999",
+        json={"title": "不存在"},
+        headers=auth_header(admin_token),
+    )
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_update_announcement_empty_title(client: AsyncClient, admin_token: str):
+    create = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    aid = create.json()["id"]
+
+    resp = await client.put(
+        f"{BASE}/{aid}",
+        json={"title": ""},
+        headers=auth_header(admin_token),
+    )
+    assert resp.status_code == 422
+
+
+# ─── DELETE (DELETE /announcements/{id}) ───────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_delete_announcement(client: AsyncClient, admin_token: str):
+    create = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    aid = create.json()["id"]
+
+    resp = await client.delete(f"{BASE}/{aid}", headers=auth_header(admin_token))
+    assert resp.status_code == 200
+    assert "message" in resp.json()
+
+    # verify it's gone from list
+    items = (await client.get(BASE, headers=auth_header(admin_token))).json()
+    assert all(a["id"] != aid for a in items)
+
+
+@pytest.mark.asyncio
+async def test_delete_nonexistent_announcement(client: AsyncClient, admin_token: str):
+    resp = await client.delete(f"{BASE}/99999", headers=auth_header(admin_token))
+    assert resp.status_code == 404
+
+
+# ─── PERMISSION CHECKS ─────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_create_requires_admin(client: AsyncClient, dba_token: str):
+    resp = await client.post(BASE, json=PAYLOAD, headers=auth_header(dba_token))
+    assert resp.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_update_requires_admin(client: AsyncClient, dba_token: str):
+    resp = await client.put(
+        f"{BASE}/1", json={"title": "x"}, headers=auth_header(dba_token)
+    )
+    assert resp.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_delete_requires_admin(client: AsyncClient, developer_token: str):
+    resp = await client.delete(f"{BASE}/1", headers=auth_header(developer_token))
+    assert resp.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_list_allows_any_authenticated_user(
+    client: AsyncClient, developer_token: str
+):
+    """All logged-in users can list announcements (not admin-only)."""
+    resp = await client.get(BASE, headers=auth_header(developer_token))
+    assert resp.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_guest_can_list_announcements(client: AsyncClient, guest_token: str):
+    resp = await client.get(BASE, headers=auth_header(guest_token))
+    assert resp.status_code == 200
+
+
+# ─── EDGE CASES ─────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_no_auth_returns_401(client: AsyncClient):
+    for method, url in [
+        ("get", BASE),
+        ("post", BASE),
+        ("put", f"{BASE}/1"),
+        ("delete", f"{BASE}/1"),
+    ]:
+        resp = await getattr(client, method)(url)
+        assert resp.status_code == 401, f"{method.upper()} {url} should be 401, got {resp.status_code}"
+
+
+@pytest.mark.asyncio
+async def test_create_and_full_lifecycle(client: AsyncClient, admin_token: str):
+    """Create → list → update → publish → list → delete → 404."""
+    # create
+    create = await client.post(BASE, json=PAYLOAD, headers=auth_header(admin_token))
+    assert create.status_code == 201
+    aid = create.json()["id"]
+
+    # list contains it
+    items = (await client.get(BASE, headers=auth_header(admin_token))).json()
+    assert any(a["id"] == aid for a in items)
+
+    # update title
+    upd = await client.put(
+        f"{BASE}/{aid}", json={"title": "新标题"}, headers=auth_header(admin_token)
+    )
+    assert upd.json()["title"] == "新标题"
+
+    # publish
+    pub = await client.put(
+        f"{BASE}/{aid}", json={"is_published": True}, headers=auth_header(admin_token)
+    )
+    assert pub.json()["is_published"] is True
+
+    # delete
+    await client.delete(f"{BASE}/{aid}", headers=auth_header(admin_token))
+
+    # update non-existent → 404
+    resp = await client.put(
+        f"{BASE}/{aid}", json={"title": "x"}, headers=auth_header(admin_token)
+    )
+    assert resp.status_code == 404
diff --git a/backend/tests/test_dashboard.py b/backend/tests/test_dashboard.py
index 6afd011..67638f5 100644
--- a/backend/tests/test_dashboard.py
+++ b/backend/tests/test_dashboard.py
@@ -1,11 +1,21 @@
+"""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("/api/v1/dashboard/stats", headers=auth_header(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
@@ -13,5 +23,195 @@ class TestDashboard:
         assert "user_total" in data
 
     async def test_unauthenticated(self, client: AsyncClient):
-        resp = await client.get("/api/v1/dashboard/stats")
+        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",
+    }
+    assert set(data.keys()) == expected_keys
+    for key in expected_keys:
+        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 audit logs with QUERY action and verify queries_today."""
+    for i in range(5):
+        log = AuditLog(
+            user_id=1,
+            action="QUERY",
+            target_type="query",
+            detail="test query",
+        )
+        db.add(log)
+    # also add a non-QUERY log (should be ignored)
+    non_query = AuditLog(
+        user_id=1,
+        action="LOGIN",
+        target_type="session",
+        detail="login event",
+    )
+    db.add(non_query)
+    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)