320 lines
13 KiB
Python
320 lines
13 KiB
Python
from services import telegram_templates
|
||
from services.telegram_templates import _sanitize_telegram_html, price_decision, triaged_alert
|
||
|
||
|
||
def test_telegram_html_sanitizer_converts_br_tags_to_newlines():
|
||
msg = _sanitize_telegram_html("第一行<br>第二行<br/>第三行<BR />第四行")
|
||
|
||
assert "<br" not in msg.lower()
|
||
assert msg == "第一行\n第二行\n第三行\n第四行"
|
||
assert _sanitize_telegram_html("第一行<br>第二行", parse_mode=None) == "第一行<br>第二行"
|
||
|
||
|
||
def test_send_telegram_with_result_sanitizes_html_payload(monkeypatch):
|
||
sent_payloads = []
|
||
|
||
class Response:
|
||
ok = True
|
||
status_code = 200
|
||
text = "ok"
|
||
|
||
def fake_post(url, json=None, timeout=None):
|
||
sent_payloads.append({"url": url, "json": json, "timeout": timeout})
|
||
return Response()
|
||
|
||
monkeypatch.setattr(telegram_templates, "_get_bot_token", lambda: "telegram-token")
|
||
monkeypatch.setattr("requests.post", fake_post)
|
||
|
||
result = telegram_templates.send_telegram_with_result(
|
||
"第一行<br>第二行<br/>第三行",
|
||
chat_ids=[101, 202],
|
||
parse_mode="HTML",
|
||
)
|
||
|
||
assert result["ok"] is True
|
||
assert result["sent"] == 2
|
||
assert [item["json"]["chat_id"] for item in sent_payloads] == [101, 202]
|
||
assert all(item["json"]["text"] == "第一行\n第二行\n第三行" for item in sent_payloads)
|
||
assert all(item["json"]["parse_mode"] == "HTML" for item in sent_payloads)
|
||
|
||
|
||
def test_price_decision_accepts_report_url_and_escapes_dynamic_fields():
|
||
message, keyboard = price_decision(
|
||
product_name="精華 <script>",
|
||
product_sku="SKU<001>",
|
||
current_price=1200,
|
||
suggested_price=990,
|
||
reason="第一行<br>第二行<script>alert(1)</script>",
|
||
insight_id=42,
|
||
report_url="https://mo.wooo.work/report?a=1&b=<x>",
|
||
)
|
||
|
||
assert "<script>" not in message
|
||
assert "<br" not in message.lower()
|
||
assert "第一行\n第二行<script>alert(1)</script>" in message
|
||
assert "精華 <script>" in message
|
||
assert "SKU<001>" in message
|
||
assert 'href="https://mo.wooo.work/report?a=1&b=<x>"' in message
|
||
assert keyboard["inline_keyboard"][0][0]["callback_data"] == "momo:pa:42"
|
||
|
||
|
||
def test_ea_escalation_uses_structured_incident_brief():
|
||
msg, keyboard = triaged_alert(
|
||
base_event={
|
||
"event_type": "ea_escalation",
|
||
"title": "🐘 EA 升級審核 · 價格下滑警報",
|
||
"summary": "自主決策信心度 0.82 低於門檻，需人工批准",
|
||
"id": "ea_review_test",
|
||
},
|
||
tier_label="🐘 Elephant Alpha · L3 HITL",
|
||
ai_summary=(
|
||
"分析顯示 5 個代表性 SKU 的價格差異分別為 16.7%~38.3%，"
|
||
"且每件價差至多 370 元。"
|
||
),
|
||
ai_cause="觸發類型：價格下滑警報 | 信心度：0.82 | 參與模組：Hermes, NemoTron",
|
||
ai_actions=[
|
||
"[5900068] [derma Angel 護妍天使] 集中抗痘精華｜"
|
||
"MOMO $300 vs PChome $250 (+16.7%)｜"
|
||
"每件價差 NT$ 50｜"
|
||
"證據：高信心同款 / 總價可比 / 可直接價格告警 / score 0.86｜"
|
||
"建議人工確認 PChome identity_v2 後評估跟價或促銷｜"
|
||
"PChome DABC53-A9009OEF",
|
||
"[3518670] L'Occitane 歐舒丹 官方直營 乳油木｜"
|
||
"MOMO $1,220 vs PChome $850 (+30.3%)｜"
|
||
"每件價差 NT$ 370｜"
|
||
"證據：高信心同款 / 總價可比 / 可直接價格告警 / score 0.91｜"
|
||
"建議人工確認 PChome identity_v2 後評估跟價或促銷｜"
|
||
"PChome DDADKS-A900HIG5Y",
|
||
],
|
||
)
|
||
|
||
assert "🧭 <b>決策狀態</b>" in msg
|
||
assert "📊 <b>風險摘要</b>" in msg
|
||
assert "📋 <b>TOP 待審 SKU</b>" in msg
|
||
assert "✅ <b>建議處置</b>" in msg
|
||
assert "• 待審 SKU：<b>2</b> 件" in msg
|
||
assert "• 價差範圍：<b>+16.7%～+30.3%</b>" in msg
|
||
assert "• 最大單件價差：<b>NT$ 370</b>" in msg
|
||
assert "<b>1. [5900068]" in msg
|
||
assert "MOMO：<b>$300</b> PChome：<b>$250</b>" in msg
|
||
assert "證據：高信心同款 / 總價可比 / 可直接價格告警 / score 0.86" in msg
|
||
assert "PChome：<code>DABC53-A9009OEF</code>" in msg
|
||
assert " • [5900068]" not in msg
|
||
assert keyboard["inline_keyboard"][0][0]["callback_data"] == "momo:eig:ea_review_test"
|
||
|
||
|
||
def test_ea_escalation_generic_actions_do_not_render_as_sku_cards():
|
||
msg, _ = triaged_alert(
|
||
base_event={
|
||
"event_type": "ea_escalation",
|
||
"title": "🐘 EA 升級審核 · 程式碼異常偵測",
|
||
"summary": "低信心且缺少可格式化的具體行動",
|
||
"id": "ea_review_generic",
|
||
},
|
||
tier_label="🐘 Elephant Alpha · L3 HITL",
|
||
ai_summary="已隱藏 LLM plan 文字，避免把推測當成事實。",
|
||
ai_cause="觸發類型：程式碼異常偵測 | 信心度：0.62 | 缺少可直接審核的實證資料",
|
||
ai_actions=[
|
||
"檢查觸發條件：{\"scan_containers\": [\"momo-pro-system\"]}",
|
||
"不執行自動動作；請先在觀測台確認對應資料來源與最近錯誤紀錄。",
|
||
],
|
||
)
|
||
|
||
assert "📋 <b>待確認事項</b>" in msg
|
||
assert "📋 <b>TOP 待審 SKU</b>" not in msg
|
||
assert "• 待審 SKU" not in msg
|
||
assert "未取得實證前，不執行自動調價、修復或策略派發" in msg
|
||
|
||
|
||
def test_ea_escalation_renders_decision_envelope_and_uses_decision_id_callback():
|
||
msg, keyboard = triaged_alert(
|
||
base_event={
|
||
"event_type": "ea_escalation",
|
||
"title": "🐘 EA 升級審核 · 資源壓力治理",
|
||
"summary": "自主決策信心度 0.82 低於門檻，需人工批准",
|
||
"decision_envelope": {
|
||
"decision_id": "ea_resource_pressure_1942",
|
||
"decision_type": "resource_optimization",
|
||
"source_agent": "elephant_alpha",
|
||
"severity": "P2",
|
||
"confidence": 0.86,
|
||
"subject": {
|
||
"sku": "action_plans",
|
||
"name": "Elephant Alpha 資源壓力治理",
|
||
},
|
||
"evidence": [
|
||
{
|
||
"type": "queue",
|
||
"metric": "action_queue_size",
|
||
"value": "7/10",
|
||
"basis": "action_plans pending / pending_review",
|
||
"confidence": 1.0,
|
||
},
|
||
],
|
||
"recommended_action": {
|
||
"action": "human_review_backlog_triage",
|
||
"owner": "ops",
|
||
"requires_hitl": True,
|
||
},
|
||
"guardrails": {
|
||
"can_auto_execute": False,
|
||
"blocked_reason": "resource_optimization requires HITL",
|
||
"data_quality": "complete",
|
||
},
|
||
},
|
||
},
|
||
tier_label="🐘 Elephant Alpha · L3 HITL",
|
||
ai_summary="只採用 action_plans 與 CPU 實測值。",
|
||
ai_cause="觸發類型：資源壓力治理 | 信心度：0.86",
|
||
ai_actions=["檢查 priority <= 2 的 action_plans"],
|
||
)
|
||
|
||
assert "🧭 <b>決策信封</b>" in msg
|
||
assert "類型：<code>resource_optimization</code>" in msg
|
||
assert "資料品質：<code>complete</code> 自動執行：<b>不允許</b>" in msg
|
||
assert "SKU：<code>action_plans</code>" in msg
|
||
assert "human_review_backlog_triage" in msg
|
||
assert keyboard["inline_keyboard"][0][0]["callback_data"] == "momo:eig:ea_resource_pressure_1942"
|
||
|
||
|
||
def test_triaged_alert_renders_decision_envelope_contract():
|
||
msg, keyboard = triaged_alert(
|
||
base_event={
|
||
"event_type": "price_alert",
|
||
"title": "MOMO / PChome 價格威脅",
|
||
"summary": "高信心同款且 PChome 低價。",
|
||
"id": "decision_env_001",
|
||
"decision_envelope": {
|
||
"decision_type": "price_alert",
|
||
"severity": "P1",
|
||
"confidence": 0.86,
|
||
"subject": {
|
||
"sku": "SKU-1",
|
||
"name": "測試商品",
|
||
"competitor_product_id": "PC-1",
|
||
"competitor_product_name": "PChome 測試商品",
|
||
},
|
||
"evidence": [
|
||
{
|
||
"type": "match",
|
||
"metric": "match_score",
|
||
"value": 0.91,
|
||
"basis": "identity_v2 + price_alert_exact",
|
||
"freshness": "2026-05-24T10:00:00+08:00",
|
||
"confidence": 0.91,
|
||
},
|
||
{
|
||
"type": "price",
|
||
"metric": "gap_pct",
|
||
"value": "18.4%",
|
||
"basis": "latest price_records",
|
||
},
|
||
],
|
||
"recommended_action": {
|
||
"action": "human_review",
|
||
"owner": "ops",
|
||
"deadline": "2026-05-24T18:00:00+08:00",
|
||
"requires_hitl": True,
|
||
},
|
||
"expected_impact": {
|
||
"revenue_loss_7d": 42000,
|
||
"gap_amount": 120,
|
||
"risk_reduction": "high",
|
||
},
|
||
"guardrails": {
|
||
"can_auto_execute": False,
|
||
"blocked_reason": "price adjustment requires HITL",
|
||
"data_quality": "complete",
|
||
},
|
||
"trace": {
|
||
"ai_call_id": 123,
|
||
"action_plan_id": 456,
|
||
"model": "qwen3:14b",
|
||
"provider": "ollama_gcp_a",
|
||
},
|
||
},
|
||
},
|
||
tier_label="Hermes · P1",
|
||
ai_summary="建議進人工價格審核。",
|
||
)
|
||
|
||
assert "🧭 <b>決策信封</b>" in msg
|
||
assert "類型：<code>price_alert</code>" in msg
|
||
assert "嚴重度：<b>P1</b>" in msg
|
||
assert "信心度：<b>86%</b>" in msg
|
||
assert "資料品質：<code>complete</code>" in msg
|
||
assert "自動執行：<b>不允許</b>" in msg
|
||
assert "邊界：price adjustment requires HITL" in msg
|
||
assert "<b>標的</b>" in msg
|
||
assert "PChome" in msg
|
||
assert "<code>match_score</code> / 91%" in msg
|
||
assert "identity_v2 + price_alert_exact" in msg
|
||
assert "動作：<code>human_review</code>" in msg
|
||
assert "revenue_loss_7d=42000" in msg
|
||
assert "ai_call_id=123" in msg
|
||
assert keyboard["inline_keyboard"][0][0]["callback_data"] == "momo:eig:decision_env_001"
|
||
|
||
|
||
def test_triaged_alert_uses_decision_id_when_event_id_missing():
|
||
_msg, keyboard = triaged_alert(
|
||
base_event={
|
||
"event_type": "price_alert",
|
||
"title": "MOMO / PChome 價格威脅",
|
||
"summary": "高信心同款且 PChome 低價。",
|
||
"decision_envelope": {
|
||
"decision_id": "nemotron:price_alert:SKU-1:abcdef12",
|
||
"decision_type": "price_alert",
|
||
"severity": "P2",
|
||
"guardrails": {"can_auto_execute": False, "data_quality": "complete"},
|
||
},
|
||
},
|
||
tier_label="NemoTron · P2",
|
||
ai_summary="建議進人工價格審核。",
|
||
)
|
||
|
||
assert (
|
||
keyboard["inline_keyboard"][0][0]["callback_data"]
|
||
== "momo:eig:nemotron:price_alert:SKU-1:abcdef12"
|
||
)
|
||
|
||
|
||
def test_triaged_alert_clamps_long_ascii_callback_to_telegram_limit():
|
||
_msg, keyboard = triaged_alert(
|
||
base_event={
|
||
"event_type": "price_alert",
|
||
"title": "MOMO / PChome 價格威脅",
|
||
"summary": "高信心同款且 PChome 低價。",
|
||
"id": "ascii-event-id-" + ("x" * 120),
|
||
},
|
||
tier_label="NemoTron · P2",
|
||
ai_summary="建議進人工價格審核。",
|
||
)
|
||
|
||
callback_data = keyboard["inline_keyboard"][0][0]["callback_data"]
|
||
assert callback_data.startswith("momo:eig:")
|
||
assert len(callback_data.encode("utf-8")) <= 64
|
||
|
||
|
||
def test_triaged_alert_clamps_multibyte_callback_to_telegram_limit():
|
||
_msg, keyboard = triaged_alert(
|
||
base_event={
|
||
"event_type": "price_alert",
|
||
"title": "MOMO / PChome 價格威脅",
|
||
"summary": "高信心同款且 PChome 低價。",
|
||
"decision_envelope": {
|
||
"decision_id": "價格決策" * 20,
|
||
"decision_type": "price_alert",
|
||
"severity": "P2",
|
||
"guardrails": {"can_auto_execute": False, "data_quality": "complete"},
|
||
},
|
||
},
|
||
tier_label="NemoTron · P2",
|
||
ai_summary="建議進人工價格審核。",
|
||
)
|
||
|
||
callback_data = keyboard["inline_keyboard"][0][0]["callback_data"]
|
||
assert callback_data.startswith("momo:eig:")
|
||
assert callback_data != "momo:eig:unknown"
|
||
assert len(callback_data.encode("utf-8")) <= 64
|
||
assert callback_data.encode("utf-8").decode("utf-8") == callback_data
|