from __future__ import annotations import sys import types import unittest from datetime import date from types import SimpleNamespace from unittest import mock if "truststore" not in sys.modules: try: import truststore # noqa: F401 except ModuleNotFoundError: truststore_stub = types.ModuleType("truststore") truststore_stub.inject_into_ssl = lambda: None sys.modules["truststore"] = truststore_stub if "lark_oapi" not in sys.modules: try: import lark_oapi # noqa: F401 except ModuleNotFoundError: lark_stub = types.ModuleType("lark_oapi") lark_stub.im = SimpleNamespace(v1=SimpleNamespace(P2ImMessageReceiveV1=object)) lark_stub.Client = object sys.modules["lark_oapi"] = lark_stub from shared.runtime import feishu_bot_bridge as bridge def successful_payload() -> dict[str, object]: return { "data": { "时间戳": "2026-08-31 12:00:00", "交期汇总": { "AM516预测交期": "2026-09-01", "AM516总交期": 1, "AM516主导因素": "现货", }, "库存信息": {"可用库存(扣除了待产数量)": 100}, "BOM信息": {"是否有BOM(False则需提醒无BOM)": True}, "采购信息": {"缺料明细": []}, } } def message_event(message_id: str, chat_id: str, open_id: str) -> SimpleNamespace: return SimpleNamespace( event=SimpleNamespace( message=SimpleNamespace(message_id=message_id, chat_id=chat_id), sender=SimpleNamespace(sender_id=SimpleNamespace(open_id=open_id)), ) ) class TriggerGateTests(unittest.TestCase): def setUp(self) -> None: with bridge._candidate_continuation_lock: bridge._candidate_continuations.clear() def test_complete_model_and_quantity_trigger(self) -> None: request = bridge.parse_trigger_request("eRob70H50I-BHM-18CTC[V5] 10台") self.assertEqual( request, bridge.RequestInput(model="eRob70H50I-BHM-18CTC[V5]", quantity=10), ) def test_natural_language_model_and_quantity_trigger(self) -> None: request = bridge.parse_trigger_request( "请查询 eRob142H100I-BHM-18ET[V4]，这个数量20台的交期" ) self.assertEqual( request, bridge.RequestInput(model="eRob142H100I-BHM-18ET[V4]", quantity=20), ) def test_unrelated_group_message_is_silent(self) -> None: self.assertIsNone(bridge.parse_trigger_request("这个机器人头像有点丑")) def test_quantity_without_complete_model_is_silent(self) -> None: self.assertIsNone(bridge.parse_trigger_request("关节模组要10台")) def test_model_without_quantity_is_silent(self) -> None: self.assertIsNone(bridge.parse_trigger_request("eRob70H50I-BHM-18CTC[V5]")) def test_ambiguous_quantity_is_silent(self) -> None: self.assertIsNone( bridge.parse_trigger_request( "eRob70H50I-BHM-18CTC[V5] 可能要10台，也可能要20台" ) ) def test_invalid_series_version_is_silent(self) -> None: self.assertIsNone(bridge.parse_trigger_request("eRob70H50I-BHM-18CTC[V6] 10台")) def test_ignored_message_does_not_build_or_reply(self) -> None: data = SimpleNamespace( event=SimpleNamespace(message=SimpleNamespace(message_id="om_ignored")) ) with ( mock.patch.object(bridge, "message_text", return_value="他怎么什么消息都回复啊"), mock.patch.object(bridge, "build_report") as build_report, mock.patch.object( bridge, "reply_received_acknowledgement", ) as acknowledge, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message(object(), data) build_report.assert_not_called() acknowledge.assert_not_called() reply.assert_not_called() def test_valid_message_builds_and_replies(self) -> None: data = SimpleNamespace( event=SimpleNamespace(message=SimpleNamespace(message_id="om_valid")) ) with ( mock.patch.object( bridge, "message_text", return_value="eRob70H50I-BHM-18CTC[V5] 10台", ), mock.patch.object( bridge, "reply_received_acknowledgement", ) as acknowledge, mock.patch.object(bridge, "build_report", return_value="report") as build_report, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message("client", data) acknowledge.assert_called_once_with("client", "om_valid") build_report.assert_called_once_with( bridge.RequestInput(model="eRob70H50I-BHM-18CTC[V5]", quantity=10) ) reply.assert_called_once_with("client", "om_valid", "report") def test_candidate_continuation_phrase_is_recognized(self) -> None: valid_phrases = ( "还有没有其他候选型号？", "还有吗？", "还有别的吗", "其他呢", "还有类似的吗", "还有什么推荐", "有没有其他选择", "还有其他方案吗", "再推荐几个", "可以再给我几个吗", "继续吧", "接着来", "下一批", "换一批", "多看几个候选", "more candidates", "any other options?", ) for phrase in valid_phrases: with self.subTest(phrase=phrase): self.assertTrue(bridge.is_candidate_continuation_request(phrase)) invalid_phrases = ( "这个机器人头像有点丑", "more interesting", "继续查询英语例句", "下一课", "再来一首歌", "有其他事情吗", ) for phrase in invalid_phrases: with self.subTest(phrase=phrase): self.assertFalse(bridge.is_candidate_continuation_request(phrase)) def test_continuation_without_same_conversation_context_is_silent(self) -> None: data = message_event("om_more", "oc_other", "ou_other") with ( mock.patch.object( bridge, "message_text", return_value="还有没有其他候选型号？", ), mock.patch.object(bridge, "build_report") as build_report, mock.patch.object( bridge, "reply_received_acknowledgement", ) as acknowledge, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message("client", data) build_report.assert_not_called() acknowledge.assert_not_called() reply.assert_not_called() def test_follow_up_continues_from_fifth_candidate_in_same_conversation(self) -> None: request_text = "eRob70H50I-FS-18CN[V5] 10台" request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10) initial_data = message_event("om_initial", "oc_same", "ou_same") follow_up_data = message_event("om_follow_up", "oc_same", "ou_same") with ( mock.patch.object( bridge, "message_text", side_effect=[request_text, "还有吗？"], ), mock.patch.object( bridge, "build_report", side_effect=["initial report", "continued report"], ) as build_report, mock.patch.object( bridge, "reply_received_acknowledgement", ) as acknowledge, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message("client", initial_data) bridge.handle_message("client", follow_up_data) self.assertEqual( build_report.call_args_list, [mock.call(request), mock.call(request, candidate_offset=4)], ) self.assertEqual( acknowledge.call_args_list, [ mock.call("client", "om_initial"), mock.call("client", "om_follow_up"), ], ) self.assertEqual( reply.call_args_list, [ mock.call("client", "om_initial", "initial report"), mock.call("client", "om_follow_up", "continued report"), ], ) def test_context_is_cleared_when_all_candidates_fit_in_first_batch(self) -> None: request_text = "eRob70H50I-BHM-18CTC[V5] 10台" initial_data = message_event("om_small", "oc_small", "ou_small") follow_up_data = message_event("om_small_more", "oc_small", "ou_small") with ( mock.patch.object( bridge, "message_text", side_effect=[request_text, "还有没有其他候选型号？"], ), mock.patch.object( bridge, "reply_received_acknowledgement", ) as acknowledge, mock.patch.object(bridge, "build_report", return_value="report") as build_report, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message("client", initial_data) bridge.handle_message("client", follow_up_data) build_report.assert_called_once_with( bridge.RequestInput(model="eRob70H50I-BHM-18CTC[V5]", quantity=10) ) acknowledge.assert_called_once_with("client", "om_small") reply.assert_called_once_with("client", "om_small", "report") def test_expired_candidate_context_is_removed(self) -> None: key = "oc_expired:ou_expired" request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10) bridge._candidate_continuations[key] = bridge.CandidateContinuation( request=request, next_offset=4, total_candidates=10, expires_at=0, ) self.assertIsNone(bridge.get_candidate_continuation(key)) self.assertNotIn(key, bridge._candidate_continuations) class CandidateBatchTests(unittest.TestCase): def setUp(self) -> None: self.request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10) self.specs = bridge.generate_candidate_specs(self.request.model) self.assertGreater(len(self.specs), bridge.CANDIDATE_BATCH_SIZE * 2) def test_initial_report_queries_baseline_and_only_four_candidates(self) -> None: with ( mock.patch.object( bridge, "run_controlled_query", return_value=successful_payload(), ) as run_query, mock.patch.object(bridge, "write_record"), ): report = bridge.build_report(self.request) queried_models = [call.args[0].model for call in run_query.call_args_list] self.assertEqual(len(queried_models), 1 + bridge.CANDIDATE_BATCH_SIZE) self.assertEqual(queried_models[0], self.request.model) self.assertCountEqual( queried_models[1:], [spec["model"] for spec in self.specs[: bridge.CANDIDATE_BATCH_SIZE]], ) self.assertIn("本次先推荐4个候选型号（原型号另作基准）", report) self.assertIn("可说“还有吗”“再推荐几个”“继续”等类似表达", report) self.assertNotIn(self.specs[bridge.CANDIDATE_BATCH_SIZE]["model"], report) def test_follow_up_report_queries_next_four_without_requerying_baseline(self) -> None: with ( mock.patch.object( bridge, "run_controlled_query", return_value=successful_payload(), ) as run_query, mock.patch.object(bridge, "write_record"), ): report = bridge.build_report( self.request, candidate_offset=bridge.CANDIDATE_BATCH_SIZE, ) queried_models = [call.args[0].model for call in run_query.call_args_list] expected_models = [ spec["model"] for spec in self.specs[ bridge.CANDIDATE_BATCH_SIZE : bridge.CANDIDATE_BATCH_SIZE * 2 ] ] self.assertEqual(len(queried_models), bridge.CANDIDATE_BATCH_SIZE) self.assertNotIn(self.request.model, queried_models) self.assertCountEqual(queried_models, expected_models) self.assertIn("本次继续推荐第5-8个候选型号", report) for model in expected_models: self.assertIn(model, report) for model in [spec["model"] for spec in self.specs[:4]]: self.assertNotIn(model, report) class WorkdayConversionTests(unittest.TestCase): def test_natural_day_conversion_rule(self) -> None: expected = { 0: 0, 1: 0, 2: 1, 3: 2, 4: 3, 5: 3, 6: 4, 7: 5, 18: 13, 23: 16, 26: 18, 31: 22, 33: 23, 36: 25, 43: 30, 48: 34, } for natural_days, workdays in expected.items(): with self.subTest(natural_days=natural_days): self.assertEqual( bridge.natural_days_to_workdays(natural_days), workdays, ) def test_2026_official_holiday_and_makeup_workdays(self) -> None: self.assertTrue(bridge.is_china_official_workday(date(2026, 9, 20))) self.assertFalse(bridge.is_china_official_workday(date(2026, 9, 25))) self.assertFalse(bridge.is_china_official_workday(date(2026, 10, 1))) self.assertTrue(bridge.is_china_official_workday(date(2026, 10, 10))) self.assertFalse(bridge.is_china_official_workday(date(2026, 10, 11))) def test_workday_dates_match_validation_samples(self) -> None: start = date(2026, 9, 2) samples = { 18: date(2026, 9, 20), 23: date(2026, 9, 23), 26: date(2026, 9, 28), 16: date(2026, 9, 17), 31: date(2026, 10, 9), 33: date(2026, 10, 10), 36: date(2026, 10, 13), 43: date(2026, 10, 20), 48: date(2026, 10, 26), } for natural_days, expected_date in samples.items(): with self.subTest(natural_days=natural_days): workdays = bridge.natural_days_to_workdays(natural_days) self.assertEqual( bridge.add_china_official_workdays(start, workdays), expected_date, ) def test_api_natural_days_drive_displayed_delivery_date(self) -> None: payload = successful_payload() summary = payload["data"]["交期汇总"] summary["AM516预测交期"] = "2026-10-03" summary["AM516总交期"] = 31 details = bridge.lead_time_details( payload, requested_quantity=10, calculation_date=date(2026, 9, 2), ) self.assertEqual(details["total_days"], 31) self.assertEqual(details["workdays"], 22) self.assertEqual(details["delivery_date"], "2026-10-09") self.assertEqual( details["prediction"], "AM516预测交期：2026-10-09（原始31个自然日，折算22个工作日）", ) def test_unconfigured_calendar_year_is_not_silently_estimated(self) -> None: with self.assertRaisesRegex( bridge.WorkdayCalendarUnavailableError, "2027年国家法定工作日历未配置", ): bridge.add_china_official_workdays(date(2026, 12, 31), 1) if __name__ == "__main__": unittest.main()