""" 系统异常：话术、判定边界、无单不落库、暂停、告警卡、重试清列。 """ from __future__ import annotations import unittest from datetime import datetime from zoneinfo import ZoneInfo from agent.ledger.memory_ledger import MemoryLedger from agent.policy import inquiry_copy as copy from agent.policy.system_exception import ( STEP_GROUP, STEP_LLM, STEP_LLM_NO_TICKET, STEP_TMS_CABIN, STEP_TMS_QUOTE, TYPE_GROUP, TYPE_LLM, TYPE_TMS, SystemExceptionEvent, call_with_retries, confirm_system_exception, extract_work_order_nos, is_tms_tech_failure, should_mark_ticket, ticket_is_paused, ) class TestSystemExceptionCopy(unittest.TestCase): def test_fallback_three_types(self) -> None: self.assertEqual(copy.system_exception_fallback(TYPE_LLM), "大模型服务异常，IT排查中...") self.assertEqual(copy.system_exception_fallback(TYPE_TMS), "TMS接口异常，IT运维排查中...") self.assertEqual( copy.system_exception_fallback(TYPE_GROUP, names=["赵六"]), "无法添加 赵六 同事，请联系管理员检查企业微信配置。", ) self.assertIn("李四、王五", copy.system_exception_fallback(TYPE_GROUP, names=["李四", "王五"])) def test_recovery_four_steps(self) -> None: self.assertEqual(copy.system_exception_recovery(STEP_TMS_QUOTE), copy.SYS_EXC_RECOVERY_TMS) self.assertEqual(copy.system_exception_recovery(STEP_LLM), copy.SYS_EXC_RECOVERY_LLM) self.assertEqual(copy.system_exception_recovery(STEP_TMS_CABIN), copy.SYS_EXC_RECOVERY_CABIN) self.assertEqual(copy.system_exception_recovery(STEP_GROUP), copy.SYS_EXC_RECOVERY_GROUP) def test_impact_by_step(self) -> None: self.assertEqual(copy.system_exception_impact(STEP_LLM_NO_TICKET), copy.IMPACT_LLM_NO_TICKET) self.assertEqual(copy.system_exception_impact(STEP_LLM), copy.IMPACT_LLM) self.assertEqual(copy.system_exception_impact(STEP_TMS_QUOTE), copy.IMPACT_TMS_QUOTE) self.assertEqual(copy.system_exception_impact(STEP_TMS_CABIN), copy.IMPACT_TMS_CABIN) self.assertEqual(copy.system_exception_impact(STEP_GROUP), copy.IMPACT_GROUP) class TestSystemExceptionRules(unittest.TestCase): def test_tms_no_price_is_not_tech(self) -> None: self.assertFalse( is_tms_tech_failure( {"ok": True, "classification": "TMS_NO_QUOTE_1002", "has_price": False} ) ) self.assertFalse( is_tms_tech_failure({"ok": False, "classification": "TMS_QUERY_INCOMPLETE"}) ) self.assertTrue( is_tms_tech_failure({"ok": False, "classification": "TMS_TECH_FAILURE"}) ) self.assertTrue(is_tms_tech_failure({"ok": False, "error": "timeout"})) def test_closed_or_blank_not_marked(self) -> None: self.assertFalse(should_mark_ticket(work_order_no="", ticket_status="")) self.assertFalse(should_mark_ticket(work_order_no="WO1", ticket_status="已关闭")) self.assertTrue(should_mark_ticket(work_order_no="WO1", ticket_status="询价中")) self.assertTrue(should_mark_ticket(work_order_no="WO1", ticket_status="转人工")) self.assertTrue(should_mark_ticket(work_order_no="WO1", ticket_status="已成交")) def test_extract_work_order_nos(self) -> None: self.assertEqual(extract_work_order_nos("请看 WO202609210001"), ["WO202609210001"]) self.assertEqual(extract_work_order_nos("虚拟 VT202609210001 不算"), []) def test_retries_succeed_on_second(self) -> None: hits = {"n": 0} def flaky() -> dict: hits["n"] += 1 if hits["n"] < 2: return {"ok": False} return {"ok": True, "value": 1} out = call_with_retries(flaky, is_ok=lambda x: bool(x.get("ok")), attempts=3) self.assertTrue(out.get("ok")) self.assertEqual(hits["n"], 2) def test_retries_keep_last_failure(self) -> None: def always_fail() -> dict: return {"ok": False, "classification": "TMS_TECH_FAILURE"} out = call_with_retries(always_fail, is_ok=lambda x: bool(x.get("ok")), attempts=3) self.assertFalse(out.get("ok")) self.assertEqual(out.get("classification"), "TMS_TECH_FAILURE") class TestConfirmSystemException(unittest.TestCase): def setUp(self) -> None: self.ledger = MemoryLedger() created = self.ledger.create_ticket( sender_id="sales1", business_line="SEA", facts={"起运港": "上海", "目的港": "洛杉矶", "品名": "衣服"}, ) self.wo = str(created.get("work_order_no") or "") self.ledger.upsert_exception_notify_user({"name": "运维甲", "wecomId": "ops-a"}) def test_no_ticket_replies_and_alerts_but_does_not_create(self) -> None: replies: list[str] = [] queued: list[dict] = [] def enqueue(*, touser, content, dedupe_key, payload): queued.append({"touser": touser, "payload": payload, "content": content}) return True, "ok" before = len(self.ledger._tickets) out = confirm_system_exception( event=SystemExceptionEvent( exception_type=TYPE_LLM, step=STEP_LLM_NO_TICKET, reason="模型返回空结果", service="大模型识别", error_code="EMPTY", ), ledger=self.ledger, reply=replies.append, enqueue=enqueue, notify_ready=True, now=datetime(2026, 9, 21, 10, 0, 0, tzinfo=ZoneInfo("Asia/Shanghai")), ) self.assertTrue(out.get("ok")) self.assertFalse(out.get("marked")) self.assertEqual(len(self.ledger._tickets), before) self.assertEqual(replies[0], copy.SYS_EXC_LLM_FALLBACK) self.assertEqual(queued[0]["touser"], "ops-a") card = queued[0]["payload"]["template_card"] self.assertEqual(card["source"]["desc"], "询价机器人系统告警") self.assertEqual(card["main_title"]["desc"], "关联工单 无") values = {row["keyname"]: row["value"] for row in card["horizontal_content_list"]} self.assertEqual(values["影响范围"], copy.IMPACT_LLM_NO_TICKET) def test_mark_ticket_and_pause(self) -> None: replies: list[str] = [] out = confirm_system_exception( event=SystemExceptionEvent( exception_type=TYPE_TMS, step=STEP_TMS_QUOTE, reason="TMS /quote/v2 接口调用失败", service="TMS报价", error_code="QUOTE_CALCULATION", work_order_no=self.wo, ticket_status="询价中", ), ledger=self.ledger, reply=replies.append, enqueue=lambda **kwargs: (True, "ok"), notify_ready=False, ) self.assertTrue(out.get("marked")) ticket = self.ledger.get_ticket(work_order_no=self.wo) self.assertEqual(ticket.status, "询价中") self.assertEqual(ticket.system_exception, TYPE_TMS) self.assertEqual(ticket.system_exception_reason, "TMS /quote/v2 接口调用失败") self.assertTrue(ticket_is_paused(ticket)) self.assertEqual(replies[0], copy.SYS_EXC_TMS_FALLBACK) def test_closed_ticket_not_marked(self) -> None: self.ledger.transition(work_order_no=self.wo, to_status="已关闭") out = confirm_system_exception( event=SystemExceptionEvent( exception_type=TYPE_TMS, step=STEP_TMS_QUOTE, reason="x", service="TMS报价", error_code="TIMEOUT", work_order_no=self.wo, ticket_status="已关闭", ), ledger=self.ledger, reply=lambda _t: None, notify_ready=False, ) self.assertFalse(out.get("marked")) ticket = self.ledger.get_ticket(work_order_no=self.wo) self.assertFalse(ticket_is_paused(ticket)) def test_later_ticket_does_not_inherit_pre_ticket_exception(self) -> None: confirm_system_exception( event=SystemExceptionEvent( exception_type=TYPE_LLM, step=STEP_LLM_NO_TICKET, reason="空结果", service="大模型识别", error_code="EMPTY", ), ledger=self.ledger, reply=lambda _t: None, notify_ready=False, ) ticket = self.ledger.get_ticket(work_order_no=self.wo) self.assertFalse(ticket_is_paused(ticket)) def test_admin_retry_tms_success_clears(self) -> None: self.ledger.mark_system_exception( work_order_no=self.wo, system_exception=TYPE_TMS, system_exception_reason="TMS /quote/v2 接口调用失败", step=STEP_TMS_QUOTE, payload={"conversation_kind": "private", "conversation_target": "sales1"}, ) from agent.policy.system_exception import run_admin_retry talks: list[str] = [] def fake_send(*, kind, target, text, extra=None): talks.append(text) import agent.policy.system_exception as se orig = se._send_to_conversation se._send_to_conversation = fake_send try: out = run_admin_retry(self.wo, ledger=self.ledger) finally: se._send_to_conversation = orig self.assertTrue(out.get("ok")) ticket = self.ledger.get_ticket(work_order_no=self.wo) self.assertFalse(ticket_is_paused(ticket)) self.assertEqual(ticket.status, "已报价") self.assertTrue(ticket.quote) self.assertTrue(any("TMS已恢复" in x for x in talks)) joined = "\n".join(talks) self.assertTrue("USD" in joined or "拉产品" in joined or "已报价" in joined) def test_admin_retry_cabin_lock_runs_lock(self) -> None: self.ledger.mark_system_exception( work_order_no=self.wo, system_exception=TYPE_TMS, system_exception_reason="空运锁舱接口调用失败", step=STEP_TMS_CABIN, payload={ "conversation_kind": "group", "conversation_target": "chat-air", "extra": {"option_no": "AIR-OPT-01", "instruction": "锁舱"}, }, ) hits = {"n": 0} orig_lock = self.ledger.lock_cabin def counted(**kwargs): hits["n"] += 1 return orig_lock(**kwargs) self.ledger.lock_cabin = counted # type: ignore[method-assign] from agent.policy.system_exception import run_admin_retry talks: list[str] = [] import agent.policy.system_exception as se orig = se._send_to_conversation se._send_to_conversation = lambda **kwargs: talks.append(str(kwargs.get("text") or "")) try: out = run_admin_retry(self.wo, ledger=self.ledger) finally: se._send_to_conversation = orig self.assertTrue(out.get("ok")) self.assertGreaterEqual(hits["n"], 1) ticket = self.ledger.get_ticket(work_order_no=self.wo) self.assertFalse(ticket_is_paused(ticket)) self.assertTrue(any("空运舱位接口已恢复" in x for x in talks)) def test_extract_tech_fail_when_model_errors(self) -> None: from agent.llm.extract_text import extract_inquiry_snapshot class Boom: ok = False error = "timeout" tool_calls = [] content = "" snap = extract_inquiry_snapshot("上海到洛杉矶衣服", allow_b=True, chat_fn=lambda **_k: Boom()) self.assertTrue(snap.get("tech_fail")) self.assertEqual(snap.get("tech_error"), "timeout") if __name__ == "__main__": unittest.main()