""" 事件异常纯规则：超时编码、点名对象、卡片接收人（不打外网）。 """ from __future__ import annotations import os import sys import unittest ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if ROOT not in sys.path: sys.path.insert(0, ROOT) from agent.policy import inquiry_copy as copy from agent.policy.event_exception import ( WAIT_ADOPT, WAIT_AIR_QUOTE, WAIT_DEAL, WAIT_PRODUCT_QUOTE, conversation_of, format_timeout_hours, pick_app_targets, pick_mentions, timeout_code, ) class TestTimeoutCode(unittest.TestCase): def test_sales_unified(self) -> None: self.assertEqual( timeout_code(wait_kind=WAIT_ADOPT, business_line="SEA", transport_class="急件"), "sea_sales", ) self.assertEqual( timeout_code(wait_kind=WAIT_DEAL, business_line="AIR", transport_class=""), "air_sales", ) self.assertEqual( timeout_code(wait_kind="lost_reason", business_line="LAND", transport_class="急件"), "land_sales", ) def test_air_route(self) -> None: self.assertEqual( timeout_code(wait_kind=WAIT_AIR_QUOTE, business_line="AIR", transport_class=""), "air_route", ) def test_sea_and_land_product_class(self) -> None: self.assertEqual( timeout_code( wait_kind=WAIT_PRODUCT_QUOTE, business_line="SEA", transport_class="Port to Port" ), "sea_product_ptp", ) self.assertEqual( timeout_code(wait_kind=WAIT_PRODUCT_QUOTE, business_line="SEA", transport_class="其他"), "sea_product_other", ) self.assertEqual( timeout_code(wait_kind=WAIT_PRODUCT_QUOTE, business_line="LAND", transport_class="急件"), "land_hk_express", ) self.assertEqual( timeout_code(wait_kind=WAIT_PRODUCT_QUOTE, business_line="SEA", transport_class=""), "", ) def test_hours_format(self) -> None: self.assertEqual(format_timeout_hours(0.25), "0.25") self.assertEqual(format_timeout_hours(48), "48") class TestMentions(unittest.TestCase): def test_product_all_unanswered(self) -> None: sales = { "wecomId": "s1", "name": "销售甲", "roleCode": "sales", "notifyEventException": 1, } members = [ {"wecomId": "p1", "name": "张三", "roleCode": "sea", "notifyEventException": 1}, {"wecomId": "p2", "name": "李四", "roleCode": "sea", "notifyEventException": 0}, {"wecomId": "s1", "name": "销售甲", "roleCode": "sales", "notifyEventException": 1}, ] hit = pick_mentions( wait_kind=WAIT_PRODUCT_QUOTE, sales=sales, members=members, business_line="SEA", ) self.assertEqual([p["name"] for p in hit], ["张三", "李四"]) apps = pick_app_targets(hit) self.assertEqual([p["wecomId"] for p in apps], ["p1"]) def test_adopt_is_private(self) -> None: self.assertEqual(conversation_of(wait_kind=WAIT_ADOPT, collab_chat_id="wrxxx"), "private") def test_deal_uses_group_when_present(self) -> None: self.assertEqual(conversation_of(wait_kind=WAIT_DEAL, collab_chat_id="wrxxx"), "group") self.assertEqual(conversation_of(wait_kind=WAIT_DEAL, collab_chat_id=""), "private") class TestMemoryEventWait(unittest.TestCase): def test_same_kind_keeps_start(self) -> None: from agent.ledger.memory_ledger import MemoryLedger book = MemoryLedger() created = book.create_ticket(sender_id="s1", business_line="SEA", facts={"运输分类": "其他"}) wo = created["work_order_no"] book.start_event_wait(work_order_no=wo, wait_kind="product_quote") first = book.get_ticket(work_order_no=wo).event_wait_started_at book.start_event_wait(work_order_no=wo, wait_kind="product_quote") self.assertEqual(book.get_ticket(work_order_no=wo).event_wait_started_at, first) def test_record_appends_and_alerts(self) -> None: from agent.ledger.memory_ledger import MemoryLedger book = MemoryLedger() created = book.create_ticket(sender_id="s1", business_line="AIR", facts={}) wo = created["work_order_no"] book.start_event_wait(work_order_no=wo, wait_kind="air_quote") book.record_event_exception(work_order_no=wo, event_exception="航线响应超时") book.record_event_exception(work_order_no=wo, event_exception="销售响应超时") t = book.get_ticket(work_order_no=wo) self.assertEqual(t.event_exception, "航线响应超时；销售响应超时") self.assertEqual(t.event_wait_alerted, 1) self.assertEqual(book.list_event_waits(), []) class TestEventExceptionCopy(unittest.TestCase): def test_group_air(self) -> None: text = copy.event_exception_session_text( exception_type="航线响应超时", work_order_no="WO202606180003", names=["张三"], hours=0.25, mention=True, ) self.assertEqual( text, "航线响应超时\n关联询价工单 WO202606180003。@张三已超过 0.25 工作小时 无响应，事件异常已记录工单，请及时处理。", ) def test_group_two_names(self) -> None: text = copy.event_exception_session_text( exception_type="产品响应超时", work_order_no="WO1", names=["张三", "李四"], hours=2, mention=True, ) self.assertIn("@张三 @李四已超过 2 工作小时", text) def test_private_sales_no_at(self) -> None: text = copy.event_exception_session_text( exception_type="销售响应超时", work_order_no="WO1", names=["王五"], hours=48, mention=False, ) self.assertIn("王五已超过 48 工作小时", text) self.assertNotIn("@王五", text) def test_test_prefix(self) -> None: """测服会话和应用卡片都不加 [test]。""" text = copy.event_exception_session_text( exception_type="销售响应超时", work_order_no="WO1", names=["王五"], hours=1, mention=False, test_prefix=True, ) self.assertFalse(text.startswith("[test]")) self.assertNotIn("[test]", text) payload = copy.event_exception_alert_payload( card_title="销售响应异常", work_order_no="WO1", hours=1, test_prefix=True, ) title = (payload.get("template_card") or {}).get("main_title", {}).get("title") or "" self.assertEqual(title, "销售响应异常") self.assertNotIn("[test]", title) def test_app_card(self) -> None: payload = copy.event_exception_alert_payload( card_title="销售响应异常", work_order_no="WO202609090199", hours=48, ) self.assertEqual(payload["wecom_app"], "notify") card = payload["template_card"] self.assertEqual(card["source"]["desc"], "询价机器人事件告警") self.assertEqual(card["main_title"]["title"], "销售响应异常") self.assertEqual(card["main_title"]["desc"], "关联询价工单 WO202609090199") self.assertIn("您已超过 48 工作小时无响应", card["sub_title_text"]) class TestPoll(unittest.TestCase): def test_fires_once_on_timeout(self) -> None: import importlib.util from datetime import datetime, timedelta from agent.ledger.memory_ledger import MemoryLedger from agent.policy.deal_followup import SHANGHAI path = os.path.join(ROOT, "agent", "jobs", "event_exception_poll.py") spec = importlib.util.spec_from_file_location("ee_poll", path) self.assertIsNotNone(spec) self.assertIsNotNone(spec.loader) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) poll_event_exceptions = mod.poll_event_exceptions book = MemoryLedger() book.set_sea_staff( [ { "wecomId": "p1", "name": "张三", "roleCode": "sea", "notifyEventException": 1, "status": "active", } ] ) created = book.create_ticket(sender_id="s1", business_line="SEA", facts={"运输分类": "其他"}) wo = created["work_order_no"] book.start_event_wait(work_order_no=wo, wait_kind="product_quote") t = book.get_ticket(work_order_no=wo) t.collab_chat_id = "wr1" t.collab_member_ids = ["p1"] t.event_wait_started_at = (datetime.now(SHANGHAI) - timedelta(hours=3)).strftime( "%Y-%m-%d %H:%M:%S" ) book._timeouts["sea_product_other"] = 0.25 groups: list[dict] = [] def send_group(*, chat_id: str, content: str, business_line: str = "") -> bool: groups.append({"chat_id": chat_id, "content": content, "line": business_line}) return True now = datetime.now(SHANGHAI) n = poll_event_exceptions( ledger=book, now=now, send_group=send_group, notify_ready=False ) self.assertEqual(n, 1) self.assertEqual(book.get_ticket(work_order_no=wo).event_exception, "产品响应超时") self.assertTrue(groups) self.assertIn("@张三", groups[0]["content"]) n2 = poll_event_exceptions( ledger=book, now=now, send_group=send_group, notify_ready=False ) self.assertEqual(n2, 0) if __name__ == "__main__": unittest.main()