BossAi/test_admin_stats.py

"""管理后台统计接口测试:验证 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()