wangwensheng/AM516
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 如实显示型号不存在提示并简化飞书异常回复

Browse Source
This commit is contained in:
ar51
2026-10-22 17:10:10 +08:00
parent d5eef9de57
commit a017c4b48c
4 changed files with 164 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
tests/test_feishu_bot_bridge.py
View File

@@ -586,15 +586,76 @@ class CandidateBatchTests(unittest.TestCase):
class UnavailableReasonTests(unittest.TestCase):
def test_model_not_found_notice_survives_controlled_query_and_stops_baseline(self) -> None:
# Reproduce the actual success=true / string data response in AR51's screenshot.
payload = {"success": True, "data": "产品型号不存在，请检查产品型号是否正确"}
request = bridge.RequestInput("eRob70H50I-FS-18CNC[V5]", 5)
completed = SimpleNamespace(stdout=bridge.json.dumps(payload), stderr="", returncode=0)
with (
mock.patch.object(bridge.subprocess, "run", return_value=completed) as run,
mock.patch.object(bridge, "write_record") as record,
):
with self.assertRaises(RuntimeError) as caught:
bridge.build_report(request)
self.assertIn("查询系统提示：产品型号不存在，请检查产品型号是否正确。", str(caught.exception))
self.assertNotIn("data", str(caught.exception))
self.assertNotIn("格式错误", str(caught.exception))
run.assert_called_once()
record.assert_not_called()
def test_model_not_found_notice_is_used_in_initial_and_continuation_replies(self) -> None:
request = bridge.RequestInput("eRob70H50I-FS-18CN[V5]", 5)
specs = bridge.generate_candidate_specs(request.model)
def query(query_request: bridge.RequestInput) -> dict[str, object]:
if query_request.model == request.model:
return successful_payload()
return {"success": True, "data": "产品型号不存在，请检查产品型号是否正确"}
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,
mock.patch.object(bridge, "write_record") as record,
):
report = bridge.build_report(request, candidate_offset=offset)
for spec in specs[offset : offset + bridge.CANDIDATE_BATCH_SIZE]:
self.assertIn(
f"- {spec['model']}：原因：查询系统提示：产品型号不存在，请检查产品型号是否正确。",
report,
)
self.assertEqual(run.call_count, bridge.CANDIDATE_BATCH_SIZE + (offset == 0))
self.assertEqual(len(record.call_args.args[2]), bridge.CANDIDATE_BATCH_SIZE)
self.assertNotIn("data 格式错误", report)
self.assertEqual(report.count(bridge.DISCLAIMER), 1)
def test_only_verified_business_notice_is_shown_without_echoing_arbitrary_text(self) -> None:
for success in (True, False):
payload = {"success": success, "data": "产品型号不存在，请检查产品型号是否正确"}
reason = (
bridge.lead_time_unavailable_reason(payload)
if success else bridge.query_failure_reason(payload)
)
self.assertEqual(reason, "查询系统提示：产品型号不存在，请检查产品型号是否正确。")
for data in (
"SECRET_MARKER sensitive raw data",
"产品型号不存在，请检查产品型号是否正确 SECRET_MARKER",
"",
):
reason = bridge.lead_time_unavailable_reason({"success": True, "data": data})
self.assertIn("未返回可用的交期信息", reason)
self.assertNotIn("SECRET_MARKER", reason)
self.assertNotIn("型号不存在", reason)
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 日期"),
("AM516工作日交期", "missing", "未返回工作日交期"),
("AM516工作日交期", None, "工作日交期为空"),
("AM516工作日交期", -1, "工作日交期无效"),
("AM516工作日交期", True, "工作日交期无效"),
("AM516最终交货日期", "missing", "未返回最终交货日期"),
("AM516最终交货日期", "", "最终交货日期为空"),
("AM516最终交货日期", "2026-02-30", "最终交货日期无效"),
]
for field, value, expected in cases:
with self.subTest(field=field, value=value):
@@ -606,36 +667,36 @@ class UnavailableReasonTests(unittest.TestCase):
reason = bridge.lead_time_details(payload, 10)["unavailable_reason"]
self.assertIn(expected, reason)
self.assertNotIn("型号不存在", reason)
self.assertIn("无法替代上述必需字段", 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": {"交期汇总": []}}, "“交期汇总”格式错误"),
({}, "未返回查询结果"),
({"data": None}, "查询结果为空"),
({"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"])
self.assertIn("缺少有效的产品物料清单", 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"}, "未提供可安全展示的具体原因"),
({"ok": False, "error": "missing API key: SECRET_MARKER"}, "交期查询尚未配置好"),
({"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"}, "无法安全连接"),
({"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:
@@ -649,6 +710,8 @@ class UnavailableReasonTests(unittest.TestCase):
bridge.run_controlled_query(request)
self.assertIn(expected, str(caught.exception))
self.assertNotIn("SECRET_MARKER", str(caught.exception))
for technical_term in ("HTTP", "TLS", "DNS", "JSON", "API Key", "success=false"):
self.assertNotIn(technical_term, str(caught.exception))
def test_stopped_feishu_reply_includes_actual_missing_field_and_no_record(self) -> None:
payload = successful_payload()
@@ -663,7 +726,7 @@ class UnavailableReasonTests(unittest.TestCase):
):
bridge.handle_message("client", data)
report = reply.call_args.args[2]
self.assertIn("原因：缺少“AM516工作日交期”字段", report)
self.assertIn("原因：查询系统未返回工作日交期", report)
self.assertIn("处理建议", report)
self.assertEqual(report.count(bridge.DISCLAIMER), 1)
write_record.assert_not_called()
@@ -686,7 +749,7 @@ class UnavailableReasonTests(unittest.TestCase):
):
report = bridge.build_report(request)
self.assertIn(f"- {specs[0]['model']}：原因：HTTP 422", report)
self.assertIn(f"- {specs[1]['model']}：原因：缺少“AM516最终交货日期”字段", report)
self.assertIn(f"- {specs[1]['model']}：原因：查询系统未返回最终交货日期", report)
self.assertNotIn("不存在或接口未返回", report)
@@ -933,7 +996,7 @@ class ApiWorkdayOutputTests(unittest.TestCase):
mock.patch.object(bridge, "run_controlled_query", return_value=payload),
mock.patch.object(bridge, "write_record") as write_record,
):
with self.assertRaisesRegex(RuntimeError, "缺少“AM516工作日交期”字段"):
with self.assertRaisesRegex(RuntimeError, "未返回工作日交期"):
bridge.build_report(request)
write_record.assert_not_called()