"""管理后台统计接口测试:验证 AI 使用和专家沟通统计,确认不含具体内容。 运行:python3 test_admin_stats.py 前置:PostgreSQL 已启动,db_init.py 已执行。 """ import sys import json from fastapi.testclient import TestClient import api as api_module from api import app client = TestClient(app) def cleanup(): """清理测试数据。""" with api_module.user_store._conn() as conn: with conn.cursor() as cur: cur.execute("DELETE FROM chat_messages WHERE phone LIKE '139_test_admin_%'") cur.execute("DELETE FROM expert_consultations WHERE user_phone LIKE '139_test_admin_%'") print("[cleanup] cleared test data") def seed_data(): """插入测试数据:普通+保密模式对话,专家咨询。""" store = api_module.chat_store # 普通模式对话 store.save_message(session_id="admin_test_n1", phone="139_test_admin_001", mode="normal", role="user", content="普通提问1", scene="legal") store.save_message(session_id="admin_test_n1", phone="139_test_admin_001", mode="normal", role="assistant", content="普通回复1", scene="legal", risk_level="green", inference_location="cloud", model_name="kimi") store.save_message(session_id="admin_test_n2", phone="139_test_admin_002", mode="normal", role="user", content="普通提问2", scene="decision") store.save_message(session_id="admin_test_n2", phone="139_test_admin_002", mode="normal", role="assistant", content="普通回复2", scene="decision", risk_level="amber", inference_location="local", model_name="local-triage", fallback_used=True) # 保密模式对话 store.save_message(session_id="admin_test_c1", phone="139_test_admin_001", mode="confidential", role="user", content="保密提问", scene="emotion") store.save_message(session_id="admin_test_c1", phone="139_test_admin_001", mode="confidential", role="assistant", content="保密回复", scene="emotion", risk_level="red", risk_escalation=True, inference_location="cloud") # 专家咨询(直接插表) with api_module.user_store._conn() as conn: with conn.cursor() as cur: # 先确保有服务关系 cur.execute("SELECT id FROM experts LIMIT 1") expert_id = cur.fetchone()[0] cur.execute("SELECT id FROM expert_services LIMIT 1") svc_row = cur.fetchone() svc_id = svc_row[0] if svc_row else None if svc_id: cur.execute( "INSERT INTO expert_consultations (service_id, user_phone, expert_id, category, topic, status) " "VALUES (%s, %s, %s, 'legal', '测试咨询1', 'replied')", (svc_id, "139_test_admin_001", expert_id), ) cur.execute( "INSERT INTO expert_consultations (service_id, user_phone, expert_id, category, topic, status) " "VALUES (%s, %s, %s, 'legal', '测试咨询2', 'pending')", (svc_id, "139_test_admin_002", expert_id), ) print("[seed] inserted test data") def test_ai_usage_no_content(): """测试 AI 使用统计接口不含具体对话内容。""" res = client.get("/admin/ai-usage") assert res.status_code == 200, f"期望200,实际{res.status_code}" data = res.json() # 验证返回的字段 assert "total_questions" in data assert "active_users" in data assert "by_scene" in data assert "by_mode" in data assert "by_risk" in data assert "by_inference" in data assert "daily_trend" in data assert "by_user" in data # 验证不含具体内容:遍历整个响应,不应出现 "content" 或对话原文 raw = json.dumps(data, ensure_ascii=False) assert "普通提问1" not in raw, "统计响应不应包含对话原文" assert "保密提问" not in raw, "统计响应不应包含对话原文" assert "普通回复1" not in raw, "统计响应不应包含回复原文" # 验证 by_user 手机号脱敏 for u in data["by_user"]: if "139_test_admin" in u.get("phone_masked", ""): assert "****" in u["phone_masked"], f"手机号应脱敏: {u['phone_masked']}" assert "139_test_admin_001" not in u["phone_masked"], "不应暴露完整手机号" print("[pass] AI 使用统计不含具体内容,手机号已脱敏") def test_expert_activity_no_content(): """测试专家沟通统计接口不含具体咨询内容。""" res = client.get("/admin/expert-activity") assert res.status_code == 200, f"期望200,实际{res.status_code}" data = res.json() # 验证返回的字段 assert "total" in data assert "replied" in data assert "pending" in data assert "by_status" in data assert "by_category" in data assert "by_expert" in data assert "daily_trend" in data assert "by_user" in data # 验证不含具体内容 raw = json.dumps(data, ensure_ascii=False) assert "测试咨询1" not in raw, "统计响应不应包含咨询主题原文" assert "测试咨询2" not in raw, "统计响应不应包含咨询主题原文" # 验证 by_user 手机号脱敏 for u in data["by_user"]: if "139_test_admin" in u.get("phone_masked", ""): assert "****" in u["phone_masked"], f"手机号应脱敏: {u['phone_masked']}" print("[pass] 专家沟通统计不含具体内容,手机号已脱敏") def test_ai_usage_dimensions(): """测试 AI 使用统计维度正确。""" res = client.get("/admin/ai-usage") data = res.json() # 应有普通和保密两种模式 modes = {m["mode"]: m["count"] for m in data["by_mode"]} assert "normal" in modes, "应有普通模式统计" assert "confidential" in modes, "应有保密模式统计" # 应有 legal/decision/emotion 等场景 scenes = {s["scene"]: s["count"] for s in data["by_scene"]} assert "legal" in scenes, "应有法务场景" assert "decision" in scenes, "应有经营决策场景" # 应有风险等级 risks = {r["level"]: r["count"] for r in data["by_risk"]} assert "green" in risks or "amber" in risks or "red" in risks, "应有风险等级统计" print("[pass] AI 使用统计维度正确") def test_expert_activity_dimensions(): """测试专家沟通统计维度正确。""" res = client.get("/admin/expert-activity") data = res.json() # 应有状态分布 statuses = {s["status"]: s["count"] for s in data["by_status"]} assert "replied" in statuses or "pending" in statuses, "应有状态统计" # 应有分类分布 cats = {c["category"]: c["count"] for c in data["by_category"]} assert "legal" in cats, "应有法务分类" print("[pass] 专家沟通统计维度正确") def main(): cleanup() try: seed_data() test_ai_usage_no_content() test_expert_activity_no_content() test_ai_usage_dimensions() test_expert_activity_dimensions() print("\n===== 全部 4 项测试通过 =====") except AssertionError as e: print(f"\n[FAIL] {e}") sys.exit(1) finally: cleanup() if __name__ == "__main__": main()