shaguabijia-app-server/tests/test_compare_alert_format.py

"""AlertHit[] → 飞书消息:分组 / 截断 / 含关键词(text + post 两种格式)。"""
from __future__ import annotations

from datetime import datetime

from app.services.compare_alert import AlertHit
from app.services.compare_alert_format import ALERT_KEYWORD, format_alert_message, format_alert_post


def _hits(n, alert_type="T1"):
    return [
        AlertHit(
            trace_id=f"t{i}",
            alert_type=alert_type,
            reason="比价过程出错",
            app_version="v0.3.4",
            created_at=None,
            trace_url=None,
            user_id=None,
        )
        for i in range(n)
    ]


# ---- format_alert_message(纯文本,保留原有测试) ----

def test_contains_keyword_and_count():
    msg = format_alert_message(
        _hits(2), window_label="2026-08-04 08:00–08:30",
        max_detail_per_type=20, max_total=50,
    )
    assert ALERT_KEYWORD in msg
    assert "本期触发 2 条" in msg
    assert "系统技术失败 2 条" in msg
    assert "t0" in msg and "v0.3.4" in msg


def test_group_by_type():
    hits = _hits(1, "T1") + _hits(1, "T6") + _hits(1, "T5")
    msg = format_alert_message(hits, window_label="w", max_detail_per_type=20, max_total=50)
    assert "系统技术失败 1 条" in msg
    assert "商品识别失败 1 条" in msg
    assert "深度放弃(cancelled) 1 条" in msg


def test_per_type_truncation():
    msg = format_alert_message(_hits(25), window_label="w", max_detail_per_type=20, max_total=50)
    assert msg.count("t0") == 1
    assert "另有 5 条" in msg


def test_total_truncation_counts_only():
    msg = format_alert_message(_hits(60), window_label="w", max_detail_per_type=20, max_total=50)
    assert "系统技术失败 60 条" in msg
    assert "t0" not in msg
    assert "分析库" in msg


# ---- format_alert_post(富文本 post) ----

def _hits_with_meta(n, alert_type="T1", *, user_id=None, trace_url=None, created_at=None):
    return [
        AlertHit(
            trace_id=f"tr{i}",
            alert_type=alert_type,
            reason="比价过程出错",
            app_version="v1.2.3",
            created_at=created_at or datetime(2026, 8, 4, 10, 30),
            trace_url=trace_url,
            user_id=user_id,
        )
        for i in range(n)
    ]


def test_post_title_contains_keyword():
    hits = _hits_with_meta(1)
    title, content = format_alert_post(
        hits, window_label="2026-08-04 10:00", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    assert ALERT_KEYWORD in title


def test_post_content_is_list():
    hits = _hits_with_meta(2)
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    assert isinstance(content, list)
    assert len(content) >= 1
    # 每个段落是 list[dict]
    for para in content:
        assert isinstance(para, list)
        for elem in para:
            assert "tag" in elem


def test_post_summary_count():
    hits = _hits_with_meta(3, "T1") + _hits_with_meta(2, "T6")
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    # 摘要第一段含合计数和类型计数
    first_para_text = "".join(e.get("text", "") for e in content[0])
    assert "合计 5 条" in first_para_text
    assert "系统技术失败 3" in first_para_text
    assert "商品识别失败 2" in first_para_text


def test_post_phone_map_applied():
    hits = _hits_with_meta(1, user_id=42)
    title, content = format_alert_post(
        hits, window_label="w", phone_map={42: "13800138000"}, max_detail_per_type=20, max_total=50,
    )
    # 找明细行(非摘要非表头)中含手机号
    all_text = " ".join(
        e.get("text", "") for para in content for e in para
    )
    assert "13800138000" in all_text


def test_post_no_user_id_shows_dash():
    hits = _hits_with_meta(1, user_id=None)
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    all_text = " ".join(e.get("text", "") for para in content for e in para)
    assert "｜ - ｜" in all_text


def test_post_trace_url_becomes_a_element():
    hits = _hits_with_meta(1, trace_url="https://trace.example.com/tr0")
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    # 找 tag=a 的元素
    a_elements = [e for para in content for e in para if e.get("tag") == "a"]
    assert len(a_elements) == 1
    assert a_elements[0]["href"] == "https://trace.example.com/tr0"
    assert a_elements[0]["text"] == "trace"


def test_post_no_trace_url_shows_trace_id_prefix():
    hits = _hits_with_meta(1, trace_url=None)
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    # 无 trace_url 时:tag=text, text=trace_id[:16]
    detail_texts = [e.get("text", "") for para in content for e in para if e.get("tag") == "text"]
    # trace_id 是 tr0,截 16 位
    assert any("tr0" in t for t in detail_texts)


def test_post_total_truncation_no_detail():
    hits = _hits_with_meta(60)
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    # 超 max_total:只有摘要+截断提示,无表头、无明细
    all_text = " ".join(e.get("text", "") for para in content for e in para)
    assert "合计 60 条" in all_text
    assert "时间 ｜ 手机号" not in all_text
    assert "tr0" not in all_text
    assert "分析库" in all_text


def test_post_per_type_truncation():
    hits = _hits_with_meta(25)
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    all_text = " ".join(e.get("text", "") for para in content for e in para)
    assert "另有 5 条" in all_text


def test_post_created_at_formatting():
    hits = [
        AlertHit(
            trace_id="tx1",
            alert_type="T1",
            reason="测试",
            app_version="v2.0",
            created_at=datetime(2026, 8, 4, 10, 30),
            trace_url=None,
            user_id=None,
        )
    ]
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    all_text = " ".join(e.get("text", "") for para in content for e in para)
    assert "08-04 10:30" in all_text


def test_post_no_created_at_shows_dash():
    hits = [
        AlertHit(
            trace_id="tx2",
            alert_type="T1",
            reason="测试",
            app_version=None,
            created_at=None,
            trace_url=None,
            user_id=None,
        )
    ]
    title, content = format_alert_post(
        hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
    )
    all_text = " ".join(e.get("text", "") for para in content for e in para)
    # 无 created_at 时间显示 "-"
    assert "- ｜" in all_text


# ---- format_alert_card(schema 2.0 卡片) ----

from app.services.compare_alert_format import format_alert_card  # noqa: E402


def _card_hits(n, alert_type="T1", *, total_ms=None, step_count=None, user_id=None,
               trace_url=None, created_at=None):
    return [
        AlertHit(
            trace_id=f"card{i}",
            alert_type=alert_type,
            reason="比价过程出错",
            app_version="v1.5.0",
            created_at=created_at or datetime(2026, 8, 4, 10, 30),
            trace_url=trace_url,
            user_id=user_id,
            total_ms=total_ms,
            step_count=step_count,
        )
        for i in range(n)
    ]


def test_card_schema_and_header():
    card = format_alert_card(
        _card_hits(1),
        window_label="2026-08-04 10:00",
        phone_map={},
        interval_min=15,
        max_detail_per_type=20,
        max_total=50,
    )
    assert card["schema"] == "2.0"
    assert card["header"]["template"] == "red"
    assert ALERT_KEYWORD in card["header"]["title"]["content"]


def test_card_body_markdown_summary():
    card = format_alert_card(
        _card_hits(3),
        window_label="2026-08-04 10:00",
        phone_map={},
        interval_min=15,
        max_detail_per_type=20,
        max_total=50,
    )
    elements = card["body"]["elements"]
    md = elements[0]
    assert md["tag"] == "markdown"
    assert "近 15 分钟" in md["content"]
    assert "合计 3 条" in md["content"]


def test_card_has_table_six_columns():
    card = format_alert_card(
        _card_hits(2),
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    elements = card["body"]["elements"]
    # 第二个元素是 table
    assert len(elements) >= 2
    table = elements[1]
    assert table["tag"] == "table"
    cols = table["columns"]
    assert len(cols) == 6
    display_names = [c["display_name"] for c in cols]
    assert display_names == ["时间", "手机号", "用时", "失败原因", "版本", "trace"]
    # trace 列用 lark_md
    trace_col = next(c for c in cols if c["name"] == "trace")
    assert trace_col["data_type"] == "lark_md"
    # 其余列 data_type 均为 text
    for c in cols:
        if c["name"] != "trace":
            assert c["data_type"] == "text"


def test_card_cost_cell_format():
    """cost 格式: total_ms 和 step_count 均有值时 '{Ns} / {M步}'"""
    hits = [
        AlertHit(
            trace_id="t1",
            alert_type="T1",
            reason="r",
            app_version="v1",
            created_at=datetime(2026, 8, 4, 10, 0),
            trace_url=None,
            user_id=None,
            total_ms=602_000,
            step_count=157,
        )
    ]
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    row = table["rows"][0]
    assert row["cost"] == "602s / 157步"


def test_card_cost_only_ms():
    hits = [
        AlertHit(
            trace_id="t2",
            alert_type="T1",
            reason="r",
            app_version="v1",
            created_at=None,
            trace_url=None,
            user_id=None,
            total_ms=30_000,
            step_count=None,
        )
    ]
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    row = table["rows"][0]
    assert row["cost"] == "30s"


def test_card_cost_only_steps():
    hits = [
        AlertHit(
            trace_id="t3",
            alert_type="T1",
            reason="r",
            app_version="v1",
            created_at=None,
            trace_url=None,
            user_id=None,
            total_ms=None,
            step_count=42,
        )
    ]
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    row = table["rows"][0]
    assert row["cost"] == "42步"


def test_card_cost_none_when_both_missing():
    hits = _card_hits(1, total_ms=None, step_count=None)
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    row = table["rows"][0]
    assert row["cost"] == "-"


def test_card_total_truncation_no_table():
    """超 max_total 时只有 markdown 摘要，没有 table 元素"""
    card = format_alert_card(
        _card_hits(60),
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    elements = card["body"]["elements"]
    assert len(elements) == 1
    assert elements[0]["tag"] == "markdown"
    assert "comparison_record" in elements[0]["content"]


def test_card_empty_hits():
    """空 hits 返回含「本期无异常」的卡片（无 table）"""
    card = format_alert_card(
        [],
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    assert card["schema"] == "2.0"
    elements = card["body"]["elements"]
    assert len(elements) == 1
    assert elements[0]["tag"] == "markdown"
    assert "本期无异常" in elements[0]["content"]


def test_card_rows_count_respects_per_type_limit():
    """每类型最多 max_detail_per_type 条"""
    hits = _card_hits(25, "T1")
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    assert len(table["rows"]) == 20


def test_card_trace_url_becomes_markdown_link():
    hits = _card_hits(1, trace_url="https://trace.example.com/t0")
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    row = table["rows"][0]
    assert row["trace"] == "[链接](https://trace.example.com/t0)"


def test_card_no_trace_url_shows_trace_id_prefix():
    hits = _card_hits(1, trace_url=None)
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    row = table["rows"][0]
    # trace_id = "card0"[:12]
    assert row["trace"] == "card0"


def test_card_phone_from_map():
    hits = _card_hits(1, user_id=7)
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={7: "13912345678"},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    row = table["rows"][0]
    assert row["phone"] == "13912345678"


def test_card_version_not_truncated():
    """版本完整显示，不缩写"""
    hits = [
        AlertHit(
            trace_id="tv1",
            alert_type="T1",
            reason="r",
            app_version="v2.15.3-release",
            created_at=None,
            trace_url=None,
            user_id=None,
        )
    ]
    card = format_alert_card(
        hits,
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    row = table["rows"][0]
    assert row["ver"] == "v2.15.3-release"


def test_card_table_has_page_size_and_header_style():
    card = format_alert_card(
        _card_hits(1),
        window_label="w",
        phone_map={},
        interval_min=10,
        max_detail_per_type=20,
        max_total=50,
    )
    table = card["body"]["elements"][1]
    assert "page_size" in table
    assert "header_style" in table
    assert table["header_style"].get("background_style") == "grey"
    assert table["header_style"].get("bold") is True