inquiry_robot/inquiry-agent/tests/test_event_exception.py

"""
事件异常纯规则：超时编码、点名对象、卡片接收人（不打外网）。
"""

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 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:
        text = copy.event_exception_session_text(
            exception_type="销售响应超时",
            work_order_no="WO1",
            names=["王五"],
            hours=1,
            mention=False,
            test_prefix=True,
        )
        self.assertTrue(text.startswith("[test] "))

    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"])


if __name__ == "__main__":
    unittest.main()