from __future__ import annotations import sys import types import unittest 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总交期": 1, "AM516工作日交期": 1, "AM516最终交货日期": "2026-09-01", "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, "add_received_reaction", ) as react, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message(object(), data) build_report.assert_not_called() react.assert_not_called() reply.assert_not_called() def test_invalid_inputs_get_guidance_without_query_or_reaction(self) -> None: for chat_type, text in [ ("p2p", "你好"), ("p2p", None), ("p2p", "还有吗"), ("group", "交期怎么查"), ("group", "eRob110H160I-FHM-18ET[V6]"), ("group", "eRob110H160I-FHM-18ET[V6] 10台或20台"), ("group", "eRob110H160I-FHM-18ET[V6] 10台 eRob142H100I-BHM-18ET[V4] 20台"), ]: with self.subTest(chat_type=chat_type, text=text): data = message_event("om_help", "oc_help", "ou_help") data.event.message.chat_type = chat_type with ( mock.patch.object(bridge, "message_text", return_value=text), mock.patch.object(bridge, "build_report") as build, mock.patch.object(bridge, "add_received_reaction") as react, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message("client", data) build.assert_not_called() react.assert_not_called() reply.assert_called_once() self.assertIn("提问示例", reply.call_args.args[2]) 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, "add_received_reaction", return_value=False, ) as react, mock.patch.object(bridge, "build_report", return_value="report") as build_report, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message("client", data) react.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_received_reaction_targets_original_message_with_onit(self) -> None: class ReactionBodyBuilder: def __init__(self) -> None: self.value: dict[str, str] | None = None def reaction_type(self, value: dict[str, str]) -> "ReactionBodyBuilder": self.value = value return self def build(self) -> SimpleNamespace: return SimpleNamespace(reaction_type=self.value) class ReactionRequestBuilder: def __init__(self) -> None: self.message_id_value: str | None = None self.body: SimpleNamespace | None = None def message_id(self, value: str) -> "ReactionRequestBuilder": self.message_id_value = value return self def request_body(self, value: SimpleNamespace) -> "ReactionRequestBuilder": self.body = value return self def build(self) -> SimpleNamespace: return SimpleNamespace( message_id=self.message_id_value, request_body=self.body, ) body_type = SimpleNamespace(builder=ReactionBodyBuilder) request_type = SimpleNamespace(builder=ReactionRequestBuilder) create_reaction = mock.Mock( return_value=SimpleNamespace( success=lambda: True, code=0, msg="ok", ) ) client = SimpleNamespace( im=SimpleNamespace( v1=SimpleNamespace( message_reaction=SimpleNamespace(create=create_reaction), ) ) ) with ( mock.patch.object( bridge.lark.im.v1, "CreateMessageReactionRequestBody", body_type, create=True, ), mock.patch.object( bridge.lark.im.v1, "CreateMessageReactionRequest", request_type, create=True, ), ): result = bridge.add_received_reaction(client, "om_original") self.assertTrue(result) sent_request = create_reaction.call_args.args[0] self.assertEqual(sent_request.message_id, "om_original") self.assertEqual( sent_request.request_body.reaction_type, {"emoji_type": "OnIt"}, ) def test_received_reaction_exception_does_not_escape(self) -> None: broken_body_type = SimpleNamespace( builder=mock.Mock(side_effect=RuntimeError("reaction unavailable")) ) with ( mock.patch.object( bridge.lark.im.v1, "CreateMessageReactionRequestBody", broken_body_type, create=True, ), mock.patch.object(bridge.logging, "exception") as log_exception, ): result = bridge.add_received_reaction(object(), "om_original") self.assertFalse(result) log_exception.assert_called_once() 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_gets_guidance(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, "add_received_reaction", ) as react, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message("client", data) build_report.assert_not_called() react.assert_not_called() reply.assert_called_once() self.assertIn("当前没有可继续的查询", reply.call_args.args[2]) 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, "add_received_reaction", ) as react, 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( react.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, "add_received_reaction", ) as react, 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) ) react.assert_called_once_with("client", "om_small") self.assertEqual(reply.call_count, 2) self.assertEqual(reply.call_args_list[0], mock.call("client", "om_small", "report")) self.assertIn("当前没有可继续的查询", reply.call_args.args[2]) 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 UnavailableReasonTests(unittest.TestCase): def test_missing_empty_and_invalid_required_fields_have_specific_reasons(self) -> None: cases = [ ("AM516工作日交期", "missing", "缺少“AM516工作日交期”字段"), ("AM516工作日交期", None, "“AM516工作日交期”返回空值"), ("AM516工作日交期", -1, "应为非负整数工作日"), ("AM516工作日交期", True, "应为非负整数工作日"), ("AM516最终交货日期", "missing", "缺少“AM516最终交货日期”字段"), ("AM516最终交货日期", "", "“AM516最终交货日期”返回空值"), ("AM516最终交货日期", "2026-02-30", "应为有效的 YYYY-MM-DD 日期"), ] for field, value, expected in cases: with self.subTest(field=field, value=value): payload = successful_payload() if value == "missing": del payload["data"]["交期汇总"][field] else: payload["data"]["交期汇总"][field] = value reason = bridge.lead_time_details(payload, 10)["unavailable_reason"] self.assertIn(expected, reason) self.assertNotIn("型号不存在", reason) self.assertIn("无法替代上述必需字段", reason) def test_missing_data_summary_and_bom_are_reported_from_response_evidence(self) -> None: for payload, expected in ( ({}, "缺少 data 字段"), ({"data": None}, "data 为空"), ({"data": []}, "data 格式错误"), ({"data": {}}, "缺少“交期汇总”字段"), ({"data": {"交期汇总": None}}, "“交期汇总”为空"), ({"data": {"交期汇总": []}}, "“交期汇总”格式错误"), ): with self.subTest(payload=payload): self.assertIn(expected, bridge.lead_time_details(payload, 10)["unavailable_reason"]) payload = successful_payload() del payload["data"]["交期汇总"]["AM516工作日交期"] payload["data"]["BOM信息"]["是否有BOM(False则需提醒无BOM)"] = False self.assertIn("接口同时标记无有效 BOM", bridge.lead_time_details(payload, 10)["unavailable_reason"]) def test_controlled_query_failure_reasons_do_not_echo_raw_error_content(self) -> None: cases = [ ({"ok": False, "error": "missing API key: SECRET_MARKER"}, "未配置交期预测 API Key"), ({"ok": False, "status_code": 401}, "身份认证失败"), ({"ok": False, "status_code": 403}, "拒绝访问"), ({"ok": False, "status_code": 404}, "接口未说明是否为型号不存在"), ({"ok": False, "status_code": 429}, "请求频率受限"), ({"ok": False, "status_code": 503}, "服务异常"), ({"ok": False, "error": "TLS certificate verification failed"}, "TLS 证书验证失败"), ({"ok": False, "error": "request failed: nodename nor servname provided SECRET_MARKER"}, "域名解析失败"), ({"ok": False, "error": "request failed: timed out SECRET_MARKER"}, "请求超时"), ({"success": False, "message": "SECRET_MARKER"}, "未提供可安全展示的具体原因"), ] request = bridge.RequestInput(model="eRob70H120I-BHM-18CT[V5]", quantity=25) for payload, expected in cases: with self.subTest(payload=payload): payload["body_preview"] = "SECRET_MARKER sensitive raw data" completed = SimpleNamespace( stdout=bridge.json.dumps(payload), stderr="SECRET_MARKER", returncode=1 if payload.get("ok") is False else 0, ) with mock.patch.object(bridge.subprocess, "run", return_value=completed): with self.assertRaises(RuntimeError) as caught: bridge.run_controlled_query(request) self.assertIn(expected, str(caught.exception)) self.assertNotIn("SECRET_MARKER", str(caught.exception)) def test_stopped_feishu_reply_includes_actual_missing_field_and_no_record(self) -> None: payload = successful_payload() del payload["data"]["交期汇总"]["AM516工作日交期"] data = message_event("om_reason", "oc_reason", "ou_reason") with ( mock.patch.object(bridge, "message_text", return_value="eRob70H120I-BHM-18CT[V5] 25台"), mock.patch.object(bridge, "add_received_reaction"), mock.patch.object(bridge, "run_controlled_query", return_value=payload), mock.patch.object(bridge, "write_record") as write_record, mock.patch.object(bridge, "reply") as reply, ): bridge.handle_message("client", data) report = reply.call_args.args[2] self.assertIn("原因：缺少“AM516工作日交期”字段", report) self.assertIn("处理建议", report) self.assertEqual(report.count(bridge.DISCLAIMER), 1) write_record.assert_not_called() def test_each_unavailable_candidate_keeps_its_own_reason(self) -> None: request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10) specs = bridge.generate_candidate_specs(request.model) def query(query_request: bridge.RequestInput) -> dict[str, object]: payload = successful_payload() if query_request.model == specs[0]["model"]: raise bridge.CandidateUnavailableError("HTTP 422：请求参数未通过接口校验。") if query_request.model == specs[1]["model"]: del payload["data"]["交期汇总"]["AM516最终交货日期"] return payload with ( mock.patch.object(bridge, "run_controlled_query", side_effect=query), mock.patch.object(bridge, "write_record"), ): report = bridge.build_report(request) self.assertIn(f"- {specs[0]['model']}：原因：HTTP 422", report) self.assertIn(f"- {specs[1]['model']}：原因：缺少“AM516最终交货日期”字段", report) self.assertNotIn("不存在或接口未返回", report) class ShortageMaterialNamesTests(unittest.TestCase): def test_all_shortage_names_are_shown_without_non_reducer_raw_fields(self) -> None: payload = successful_payload() payload["data"]["采购信息"]["缺料明细"] = [ {"缺料物料": "20.30.00157", "物料名称": "减速器", "库存数量": 11}, {"缺料物料": "20.40.00999", "物料名称": " 编码器 ", "库存数量": 9876}, {"缺料物料": "20.50.00888", "缺料物料名称": "轴承", "在途数量": 8765}, {"缺料物料": "20.40.00777", "物料名称": "编码器"}, ] lines = bridge.lead_time_details(payload, requested_quantity=10)["evidence_lines"] self.assertEqual(lines[0], "- 缺料物料明细：减速器；编码器；轴承") self.assertIn("减速器缺料展开：物料编码20.30.00157；库存数量11；", lines[1]) for raw_value in ("20.40.00999", "20.50.00888", "20.40.00777", "9876", "8765"): self.assertNotIn(raw_value, "\n".join(lines)) def test_missing_names_never_fall_back_to_codes_or_claim_no_shortage(self) -> None: for valid_name in (False, True): with self.subTest(valid_name=valid_name): payload = successful_payload() shortages = [ {"缺料物料": "20.40.00999"}, {"物料名称": " "}, {"物料名称": 123}, None, ] if valid_name: shortages.append({"物料名称": "轴承"}) payload["data"]["采购信息"]["缺料明细"] = shortages lines = bridge.lead_time_details(payload, requested_quantity=10)["evidence_lines"] expected = "4项缺料记录未返回物料名称，需人工复核" if valid_name: expected = f"轴承（另有{expected}）" self.assertEqual(lines, [f"- 缺料物料明细：{expected}"]) def test_empty_missing_and_invalid_bom_remain_distinct(self) -> None: payload = successful_payload() lines = bridge.lead_time_details(payload, requested_quantity=10)["evidence_lines"] self.assertEqual(lines, ["- 缺料物料明细：API未返回缺料记录。"]) del payload["data"]["采购信息"]["缺料明细"] lines = bridge.lead_time_details(payload, requested_quantity=10)["evidence_lines"] self.assertIn("接口未返回可识别缺料汇总", lines[0]) payload["data"]["BOM信息"]["是否有BOM(False则需提醒无BOM)"] = False payload["data"]["采购信息"]["缺料明细"] = [{"物料名称": "轴承"}] lines = bridge.lead_time_details(payload, requested_quantity=10)["evidence_lines"] self.assertIn("API未返回有效BOM", lines[0]) self.assertNotIn("轴承", "\n".join(lines)) def test_initial_and_continuation_reports_keep_names_under_their_model(self) -> None: request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10) def query(query_request: bridge.RequestInput) -> dict[str, object]: payload = successful_payload() payload["data"]["采购信息"]["缺料明细"] = [ {"物料名称": f"样例物料-{query_request.model}"} ] return payload for offset in (0, bridge.CANDIDATE_BATCH_SIZE): with ( self.subTest(offset=offset), mock.patch.object(bridge, "run_controlled_query", side_effect=query) as run_query, mock.patch.object(bridge, "write_record"), ): report = bridge.build_report(request, candidate_offset=offset) self.assertEqual(report.count("- 缺料物料明细："), run_query.call_count) for call in run_query.call_args_list: model = call.args[0].model heading = report.index(f"{model}（") detail = report.index(f"- 缺料物料明细：样例物料-{model}", heading) end = report.index("- 需客户确认的关键差异：", heading) self.assertLess(detail, end) class ApiWorkdayOutputTests(unittest.TestCase): def test_api_workdays_and_date_override_legacy_fields(self) -> None: payload = successful_payload() summary = payload["data"]["交期汇总"] summary["AM516预测交期"] = "2026-10-03" summary["AM516总交期"] = 31 summary["AM516工作日交期"] = 8 summary["AM516最终交货日期"] = "2026-10-08" details = bridge.lead_time_details(payload, requested_quantity=10) self.assertEqual(details["total_days"], 31) self.assertEqual(details["workdays"], 8) self.assertEqual(details["delivery_date"], "2026-10-08") self.assertEqual( details["prediction"], "AM516预测交期：8个工作日，最终交货日期：2026-10-08", ) def test_new_fields_work_without_natural_days_or_local_calendar(self) -> None: payload = successful_payload() summary = payload["data"]["交期汇总"] del summary["AM516总交期"] summary["AM516工作日交期"] = 8 summary["AM516最终交货日期"] = "2027-01-15" details = bridge.lead_time_details(payload, requested_quantity=10) self.assertTrue(details["has_lead_time"]) self.assertEqual(details["workdays"], 8) self.assertEqual(details["delivery_date"], "2027-01-15") def test_zero_workdays_are_preserved(self) -> None: payload = successful_payload() payload["data"]["交期汇总"]["AM516工作日交期"] = 0 details = bridge.lead_time_details(payload, requested_quantity=10) self.assertTrue(details["has_lead_time"]) self.assertEqual( details["prediction_value"], "0个工作日，最终交货日期：2026-09-01" ) def test_missing_new_fields_do_not_fall_back_to_old_fields(self) -> None: for field in ("AM516工作日交期", "AM516最终交货日期"): with self.subTest(field=field): payload = successful_payload() summary = payload["data"]["交期汇总"] summary["AM516预测交期"] = "2026-09-01" summary["总交期"] = 15 del summary[field] details = bridge.lead_time_details(payload, requested_quantity=10) self.assertFalse(details["has_lead_time"]) self.assertIsNone(details["prediction_value"]) self.assertIn("未返回有效工作日交期和最终交货日期", details["prediction"]) def test_invalid_workdays_and_dates_are_not_displayed_as_valid(self) -> None: invalid_values = { "AM516工作日交期": [None, True, -1, 1.5, "8", float("nan"), float("inf")], "AM516最终交货日期": [None, True, 20261008, "", "2026-02-30", "20261008"], } for field, values in invalid_values.items(): for value in values: with self.subTest(field=field, value=value): payload = successful_payload() payload["data"]["交期汇总"][field] = value details = bridge.lead_time_details(payload, requested_quantity=10) self.assertFalse(details["has_lead_time"]) self.assertIsNone(details["prediction_value"]) def test_report_and_record_use_api_workdays_for_fastest_candidate(self) -> None: request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10) fastest_model = bridge.generate_candidate_specs(request.model)[0]["model"] def query(query_request: bridge.RequestInput) -> dict[str, object]: payload = successful_payload() summary = payload["data"]["交期汇总"] fastest = query_request.model == fastest_model # Deliberately disagree with natural-day ranking to detect old conversion. summary["AM516总交期"] = 31 if fastest else 1 summary["AM516工作日交期"] = 8 if fastest else 18 summary["AM516最终交货日期"] = "2026-10-08" if fastest else "2026-10-21" return payload with ( mock.patch.object(bridge, "run_controlled_query", side_effect=query), mock.patch.object(bridge, "write_record") as write_record, ): report = bridge.build_report(request) record = bridge.render_record(*write_record.call_args.args) self.assertIn(f"最快候选型号：{fastest_model}", report) self.assertIn("最快候选交期：8个工作日，最终交货日期：2026-10-08", report) for output in (report, record): self.assertIn("8个工作日，最终交货日期：2026-10-08", output) self.assertIn("18个工作日，最终交货日期：2026-10-21", output) self.assertNotIn("自然日", output) self.assertNotIn("折算", output) def test_baseline_missing_new_fields_stops_before_recording(self) -> None: payload = successful_payload() del payload["data"]["交期汇总"]["AM516工作日交期"] request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10) with ( mock.patch.object(bridge, "run_controlled_query", return_value=payload), mock.patch.object(bridge, "write_record") as write_record, ): with self.assertRaisesRegex(RuntimeError, "工作日交期和最终交货日期"): bridge.build_report(request) write_record.assert_not_called() if __name__ == "__main__": unittest.main()