yunke/yunke-server/tests/test_management_anomalies.py

"""管理域确定性异常规则判别测试(批次B Task B3,v1 §5.4 异常卡片)。"""
import json

import pytest

from app.management.anomalies import detect_anomalies
from app.management.employee_metrics import PeriodMetrics
from app.management.metric_definitions import DEFAULT_SCORE_CONFIG


def _pm(**overrides):
    """全零/None 默认的 PeriodMetrics 构造器(22 字段,覆盖式改)。

    与 test_management_scoring._pm 同式(测试间不共享 helper,复制保持独立)。
    """
    base = {
        "user_id": "u1", "days_with_data": 0, "msg_sent": 0, "msg_recv": 0,
        "cust_threads": 0, "threads_answered": 0, "fr_sum_ms": 0,
        "fr_p50_ms": None, "fr_p90_ms": None,
        "reply_2h_count": 0, "reply_24h_count": 0,
        "active_friends": 0, "active_rooms_avg": 0.0,
        "new_customer": 0, "cooling_customer": 0, "dormant_customer": 0,
        "end_active_friends": 0, "reactivated_customer": 0,
        "call_count": 0, "call_seconds": 0, "answered_call": 0,
        "call_followup": 0,
    }
    base.update(overrides)
    return PeriodMetrics(**base)


def _detect(cur, base, *, n_days=7, user_id="u1", config=None):
    return detect_anomalies(
        user_id, cur, base, n_days=n_days,
        config=config if config is not None else DEFAULT_SCORE_CONFIG)


# ---- 1. reply_p90_worsened ----

def test_p90_worsened_warning_with_pct():
    """×1.6 → warning;标题百分比=(160-100)/100=60%;affected=未回串数。"""
    cur = _pm(cust_threads=20, threads_answered=12, fr_p90_ms=9_600_000)   # 160min
    base = _pm(cust_threads=40, threads_answered=40, fr_p90_ms=6_000_000)  # 100min
    cards = _detect(cur, base)
    assert [c["rule_code"] for c in cards] == ["reply_p90_worsened"]
    c = cards[0]
    assert c["severity"] == "warning"
    assert c["title"] == "回复 P90 较个人基线上升 60%"
    assert c["current_value"] == 9_600_000
    assert c["baseline_value"] == 6_000_000
    assert c["unit"] == "ms"
    assert c["sample_size"] == 20
    assert c["affected_count"] == 8  # 20 串 - 12 已回 = 期内未回串数
    assert c["drilldown"] == {"type": "metric",
                              "query": {"metric_code": "first_reply_p90",
                                        "user_id": "u1"}}


def test_p90_worsened_critical_at_3x():
    cur = _pm(cust_threads=20, threads_answered=20, fr_p90_ms=18_000_000)
    base = _pm(cust_threads=40, threads_answered=40, fr_p90_ms=6_000_000)
    cards = _detect(cur, base)
    assert len(cards) == 1
    assert cards[0]["severity"] == "critical"
    assert "上升 200%" in cards[0]["title"]


def test_p90_boundaries_strict_compare():
    """×1.5 恰好 → 严格大于不成立不触发;×2.5 恰好 → warning 非 critical;
    threads 恰好 = soft(10) → ≥soft 触发。"""
    base = _pm(cust_threads=40, fr_p90_ms=6_000_000)
    assert _detect(_pm(cust_threads=20, fr_p90_ms=9_000_000), base) == []
    cards = _detect(_pm(cust_threads=20, fr_p90_ms=15_000_000), base)
    assert [c["severity"] for c in cards] == ["warning"]
    cards = _detect(_pm(cust_threads=10, fr_p90_ms=9_600_000), base)
    assert [c["rule_code"] for c in cards] == ["reply_p90_worsened"]


def test_p90_below_soft_threads_not_fired():
    cur = _pm(cust_threads=5, fr_p90_ms=9_600_000)  # <soft 10
    base = _pm(cust_threads=40, threads_answered=40, fr_p90_ms=6_000_000)
    assert _detect(cur, base) == []


# ---- 2. 基线缺失 ----

def test_baseline_none_all_rules_silent():
    """新员工无个人基线:即便 p90 恶化/活动骤降/过载形态俱在也不出卡。"""
    cur = _pm(cust_threads=20, threads_answered=5, fr_p90_ms=30_000_000,
              active_friends=85, msg_sent=10, msg_recv=10)
    assert _detect(cur, None) == []


# ---- 3. reply_rate_dropped ----

def test_reply_rate_dropped_warning():
    """0.8→0.5(0.5 < 0.7×0.8=0.56)→ warning;百分比 37.5%→38%。"""
    cur = _pm(cust_threads=20, reply_2h_count=10, threads_answered=15)   # 0.5
    base = _pm(cust_threads=40, reply_2h_count=32, threads_answered=40)  # 0.8
    cards = _detect(cur, base)
    assert [c["rule_code"] for c in cards] == ["reply_rate_dropped"]
    c = cards[0]
    assert c["severity"] == "warning"
    assert c["title"] == "2h 回复率较个人基线下降 38%"
    assert c["current_value"] == pytest.approx(0.5)
    assert c["baseline_value"] == pytest.approx(0.8)
    assert c["unit"] == "0-1"
    assert c["sample_size"] == 20
    assert c["affected_count"] == 5
    assert c["drilldown"] == {"type": "metric",
                              "query": {"metric_code": "unreplied",
                                        "user_id": "u1"}}


def test_reply_rate_small_drop_not_fired():
    """0.8→0.6(0.6 ≥ 0.56)→ 不触发。"""
    cur = _pm(cust_threads=20, reply_2h_count=12)   # 0.6
    base = _pm(cust_threads=40, reply_2h_count=32)  # 0.8
    assert _detect(cur, base) == []


def test_rate_baseline_no_threads_not_fired():
    """基线无串样本 → 基线率 None → 不判(基线值 None 不触发)。"""
    cur = _pm(cust_threads=20, reply_2h_count=0)
    base = _pm(cust_threads=0)  # derive → reply_2h_rate None
    assert _detect(cur, base) == []


# ---- 4. activity_drop ----

def test_activity_drop_triggered():
    """当前 7 天日均 10 < 基线 28 天日均 30×0.5=15 → warning。"""
    cur = _pm(msg_sent=40, msg_recv=30, active_friends=25)      # 70/7=10 条/天
    base = _pm(msg_sent=600, msg_recv=240, days_with_data=20)   # 840/28=30 条/天
    cards = _detect(cur, base)
    assert [c["rule_code"] for c in cards] == ["activity_drop"]
    c = cards[0]
    assert c["severity"] == "warning"
    assert c["title"] == "日均消息量较个人基线下降 67%"  # 66.7→67
    assert c["current_value"] == pytest.approx(10.0)
    assert c["baseline_value"] == pytest.approx(30.0)
    assert c["unit"] == "条"
    assert c["sample_size"] == 25
    assert c["affected_count"] == 25
    assert c["drilldown"] == {"type": "employee", "query": {"user_id": "u1"}}


def test_activity_drop_sparse_baseline_not_fired():
    """基线 days_with_data=3 < min_days(5) → 基线不可比,不触发。"""
    base = _pm(msg_sent=600, msg_recv=240, days_with_data=3)
    cur = _pm(msg_sent=40, msg_recv=30)
    assert _detect(cur, base) == []


# ---- 5. workload_overload ----

def test_workload_overload_triggered():
    cards = _detect(_pm(active_friends=85), _pm())
    assert [c["rule_code"] for c in cards] == ["workload_overload"]
    c = cards[0]
    assert c["severity"] == "warning"
    assert c["title"] == "活跃客户 85 人超过负载上限 80 人"
    assert c["current_value"] == 85
    assert c["baseline_value"] is None  # 规则与个人基线无关,阈值在 band 配置
    assert c["unit"] == "人"
    assert c["sample_size"] == 85
    assert c["affected_count"] == 85
    assert c["drilldown"] == {"type": "employee", "query": {"user_id": "u1"}}


def test_workload_overload_at_limit_not_fired():
    assert _detect(_pm(active_friends=80), _pm()) == []  # 严格大于上限


# ---- 6. 配置驱动与组合 ----

def test_thresholds_read_from_config():
    """soft_threads/band high 改配置即改判——阈值不硬编码。"""
    cfg = json.loads(json.dumps(DEFAULT_SCORE_CONFIG))
    cfg["sample_guards"]["soft_threads"] = 25
    cfg["dimensions"]["workload"]["metrics"]["active_friends_band"]["high"] = 90
    cur = _pm(cust_threads=20, fr_p90_ms=9_600_000, active_friends=85)
    base = _pm(cust_threads=40, fr_p90_ms=6_000_000)
    # threads 20 < soft25 → p90/rate 不评;85 ≤ 90 → 不 overload
    assert _detect(cur, base, config=cfg) == []


def test_multiple_rules_stack_in_order():
    """p90 ×5 critical + activity 骤降 + overload 90>80 同时触发,按规则表序输出;
    两侧回复率同为 0(0<0 不成立)→ rate 规则不触发。"""
    cur = _pm(cust_threads=30, threads_answered=10, fr_p90_ms=30_000_000,
              active_friends=90, msg_sent=10, msg_recv=10)
    base = _pm(cust_threads=60, threads_answered=60, fr_p90_ms=6_000_000,
               msg_sent=600, msg_recv=240, days_with_data=20)
    cards = _detect(cur, base)
    assert [c["rule_code"] for c in cards] == [
        "reply_p90_worsened", "activity_drop", "workload_overload"]
    assert cards[0]["severity"] == "critical"


def test_card_structure_all_fields():
    """单卡 12 字段齐全且类型正确(v1 §5.4)。"""
    cur = _pm(cust_threads=20, reply_2h_count=20, threads_answered=12,
              fr_p90_ms=9_600_000, active_friends=30,
              msg_sent=80, msg_recv=30)  # 日均 110/7≈15.7 ≥ 15 → activity 不触发
    base = _pm(cust_threads=40, reply_2h_count=40, threads_answered=40,
               fr_p90_ms=6_000_000, active_friends=30,
               msg_sent=600, msg_recv=240, days_with_data=20)
    cards = _detect(cur, base, user_id="u9")
    assert len(cards) == 1
    c = cards[0]
    assert set(c) == {
        "rule_code", "severity", "employee_user_id", "title",
        "current_value", "baseline_value", "unit", "sample_size",
        "affected_count", "reason", "drilldown", "suggested_action",
    }
    assert isinstance(c["rule_code"], str)
    assert c["severity"] in ("warning", "critical")
    assert c["employee_user_id"] == "u9"
    assert isinstance(c["title"], str) and c["title"]
    assert isinstance(c["current_value"], (int, float))
    assert c["baseline_value"] is None or isinstance(
        c["baseline_value"], (int, float))
    assert isinstance(c["unit"], str)
    assert isinstance(c["sample_size"], int)
    assert isinstance(c["affected_count"], int)
    assert isinstance(c["reason"], str) and c["reason"]
    assert set(c["drilldown"]) == {"type", "query"}
    assert isinstance(c["drilldown"]["query"], dict)
    assert isinstance(c["suggested_action"], str) and c["suggested_action"]