Download test/test_error_message_service.py from zlcy/study: direct link, hf CLI and curl.
- Browser
- Download file 26.8 kB
-
https://huggingface.co/spaces/zlcy/study/resolve/main/test/test_error_message_service.py
- Command line
-
hf download hf://spaces/zlcy/study/test/test_error_message_service.py
-
curl -L -o test_error_message_service.py https://huggingface.co/spaces/zlcy/study/resolve/main/test/test_error_message_service.py
26.8 kB
| """报错文案映射服务测试。 | |
| 覆盖三块: | |
| 1. 关键词归一化与规则归一化(脏数据要能安全丢弃 / 去重); | |
| 2. translate 的命中优先级(命中即替换、未命中记待办并保留原文); | |
| 3. 未配置报错的收集、去重计数、清空与上限淘汰。 | |
| """ | |
| import json | |
| import tempfile | |
| import unittest | |
| from pathlib import Path | |
| class KeywordNormalizationTests(unittest.TestCase): | |
| def test_accepts_list_and_separator_strings(self) -> None: | |
| from services.error_message_service import _normalize_keywords | |
| self.assertEqual(_normalize_keywords(["a1", "b2"]), ["a1", "b2"]) | |
| # 换行 / 逗号 / 中文逗号 / 顿号都应被拆开 | |
| self.assertEqual(_normalize_keywords("a1\nb2,c3,d4、e5"), ["a1", "b2", "c3", "d4", "e5"]) | |
| def test_drops_short_and_duplicate_keywords(self) -> None: | |
| from services.error_message_service import _normalize_keywords | |
| # 短于 MIN_KEYWORD_CHARS 的关键词会误伤正常报错,必须丢掉;重复项按小写去重 | |
| self.assertEqual(_normalize_keywords(["a", "ab", "AB", "cd", ""]), ["ab", "cd"]) | |
| def test_ignores_non_string_input(self) -> None: | |
| from services.error_message_service import _normalize_keywords | |
| self.assertEqual(_normalize_keywords(None), []) | |
| self.assertEqual(_normalize_keywords(123), []) | |
| class RuleNormalizationTests(unittest.TestCase): | |
| def test_drops_rules_missing_keywords_or_display(self) -> None: | |
| from services.error_message_service import normalize_rules | |
| rules = normalize_rules([ | |
| {"keywords": ["throttled"], "display": "太挤了"}, | |
| {"keywords": [], "display": "没关键词"}, | |
| {"keywords": ["x1"], "display": ""}, | |
| "not-a-dict", | |
| ]) | |
| self.assertEqual(len(rules), 1) | |
| self.assertEqual(rules[0]["keywords"], ["throttled"]) | |
| def test_enabled_defaults_and_string_coercion(self) -> None: | |
| from services.error_message_service import normalize_rules | |
| rules = normalize_rules([ | |
| {"id": "r1", "keywords": ["aa"], "display": "x"}, | |
| {"id": "r2", "keywords": ["bb"], "display": "y", "enabled": "false"}, | |
| {"id": "r3", "keywords": ["cc"], "display": "z", "enabled": "yes"}, | |
| ]) | |
| self.assertTrue(rules[0]["enabled"]) | |
| self.assertFalse(rules[1]["enabled"]) | |
| self.assertTrue(rules[2]["enabled"]) | |
| def test_duplicate_ids_get_suffixed(self) -> None: | |
| from services.error_message_service import normalize_rules | |
| rules = normalize_rules([ | |
| {"id": "dup", "keywords": ["aa"], "display": "x"}, | |
| {"id": "dup", "keywords": ["bb"], "display": "y"}, | |
| ]) | |
| self.assertEqual([rule["id"] for rule in rules], ["dup", "dup-2"]) | |
| def test_missing_id_gets_fallback(self) -> None: | |
| from services.error_message_service import normalize_rules | |
| rules = normalize_rules([{"keywords": ["aa"], "display": "x"}]) | |
| self.assertEqual(rules[0]["id"], "rule-1") | |
| class TranslateTests(unittest.TestCase): | |
| """translate 的命中 / 未命中行为。用临时文件隔离未配置报错的写入。""" | |
| def setUp(self) -> None: | |
| self._tmp = tempfile.TemporaryDirectory() | |
| self._store = Path(self._tmp.name) / "unconfigured.json" | |
| def tearDown(self) -> None: | |
| self._tmp.cleanup() | |
| def _service(self, rules): | |
| from services.error_message_service import ErrorMessageService | |
| service = ErrorMessageService(store_path=self._store) | |
| service._rules = lambda: __import__( | |
| "services.error_message_service", fromlist=["normalize_rules"] | |
| ).normalize_rules(rules) | |
| return service | |
| def test_hit_returns_display_text(self) -> None: | |
| service = self._service([{"keywords": ["upload_protection_limit_exceeded"], "display": "上传额度已满"}]) | |
| raw = '/backend-api/files failed: status=429, body={"reason": "upload_protection_limit_exceeded"}' | |
| self.assertEqual(service.translate(raw), "上传额度已满") | |
| def test_match_is_case_insensitive(self) -> None: | |
| service = self._service([{"keywords": ["Throttled"], "display": "太挤了"}]) | |
| self.assertEqual(service.translate("upstream THROTTLED now"), "太挤了") | |
| def test_disabled_rule_is_skipped(self) -> None: | |
| service = self._service([{"keywords": ["throttled"], "display": "太挤了", "enabled": False}]) | |
| self.assertEqual(service.translate("throttled"), "throttled") | |
| def test_miss_keeps_original_and_records_todo(self) -> None: | |
| service = self._service([{"keywords": ["throttled"], "display": "太挤了"}]) | |
| raw = "upstream connect failed: 由于目标计算机积极拒绝" | |
| self.assertEqual(service.translate(raw, channel="api"), raw) | |
| items = service.list_unconfigured() | |
| self.assertEqual(len(items), 1) | |
| self.assertEqual(items[0]["channel"], "api") | |
| self.assertEqual(items[0]["count"], 1) | |
| def test_empty_message_is_ignored(self) -> None: | |
| service = self._service([]) | |
| self.assertEqual(service.translate(""), "") | |
| self.assertEqual(service.translate(None), "") | |
| self.assertEqual(service.list_unconfigured(), []) | |
| def test_translate_detail_only_touches_text_fields(self) -> None: | |
| service = self._service([{"keywords": ["throttled"], "display": "太挤了"}]) | |
| detail = {"code": "throttled", "message": "throttled for you", "type": "throttled"} | |
| result = service.translate_detail(detail) | |
| self.assertEqual(result["message"], "太挤了") | |
| # code / type 是程序判断用的,不能被替换 | |
| self.assertEqual(result["code"], "throttled") | |
| self.assertEqual(result["type"], "throttled") | |
| class FallbackDisplayTests(unittest.TestCase): | |
| """未命中规则时的「兜底成通用文案」开关(后台可配,默认关)。 | |
| 后台那个勾选框写着「勾选后未命中规则的报错前台统一显示『操作失败,请稍后重试』」, | |
| 所以这里把开关的两个状态都钉住:关 = 原样显示(便于补规则),开 = 通用文案。 | |
| """ | |
| def setUp(self) -> None: | |
| self._tmp = tempfile.TemporaryDirectory() | |
| self._store = Path(self._tmp.name) / "unconfigured.json" | |
| def tearDown(self) -> None: | |
| self._tmp.cleanup() | |
| def _service(self, rules, *, fallback: bool): | |
| from services.error_message_service import ErrorMessageService, normalize_rules | |
| service = ErrorMessageService(store_path=self._store) | |
| service._rules = lambda: normalize_rules(rules) | |
| # 用实例属性覆盖静态方法,避免依赖真实配置(测试进程里 config 的共享设置不可控)。 | |
| service.fallback_enabled = lambda: fallback | |
| return service | |
| def test_unmatched_keeps_raw_text_when_disabled(self) -> None: | |
| service = self._service([], fallback=False) | |
| raw = "upstream connect failed: 由于目标计算机积极拒绝" | |
| self.assertEqual(service.translate(raw, channel="image_task"), raw) | |
| def test_unmatched_uses_generic_copy_when_enabled(self) -> None: | |
| from services.error_message_service import FALLBACK_DISPLAY | |
| service = self._service([], fallback=True) | |
| self.assertEqual(service.translate("upstream boom"), FALLBACK_DISPLAY) | |
| # ⚠️ 兜底只换前台文案:原文照样进「未配置报错」台账,管理员还能照着补规则。 | |
| items = service.list_unconfigured() | |
| self.assertEqual(len(items), 1) | |
| self.assertIn("upstream boom", items[0]["sample"]) | |
| def test_matched_rule_wins_over_fallback(self) -> None: | |
| service = self._service([{"keywords": ["throttled"], "display": "太挤了"}], fallback=True) | |
| self.assertEqual(service.translate("upstream THROTTLED now"), "太挤了") | |
| def test_disabled_rule_under_fallback_falls_back(self) -> None: | |
| """停用的规则不算命中:开着兜底时不能因为它而漏出技术报错。""" | |
| from services.error_message_service import FALLBACK_DISPLAY | |
| service = self._service( | |
| [{"keywords": ["throttled"], "display": "太挤了", "enabled": False}], | |
| fallback=True, | |
| ) | |
| self.assertEqual(service.translate("throttled"), FALLBACK_DISPLAY) | |
| def test_empty_message_stays_empty(self) -> None: | |
| service = self._service([], fallback=True) | |
| self.assertEqual(service.translate(""), "") | |
| self.assertEqual(service.translate(None), "") | |
| def test_flag_reads_from_config(self) -> None: | |
| from unittest.mock import patch | |
| from services.config import config | |
| from services.error_message_service import ErrorMessageService | |
| with patch.object(type(config), "error_message_fallback_enabled", property(lambda self: True)): | |
| self.assertTrue(ErrorMessageService.fallback_enabled()) | |
| with patch.object(type(config), "error_message_fallback_enabled", property(lambda self: False)): | |
| self.assertFalse(ErrorMessageService.fallback_enabled()) | |
| class UnconfiguredCollectionTests(unittest.TestCase): | |
| def setUp(self) -> None: | |
| self._tmp = tempfile.TemporaryDirectory() | |
| self._store = Path(self._tmp.name) / "unconfigured.json" | |
| def tearDown(self) -> None: | |
| self._tmp.cleanup() | |
| def _service(self): | |
| from services.error_message_service import ErrorMessageService | |
| service = ErrorMessageService(store_path=self._store) | |
| service._rules = lambda: [] | |
| return service | |
| def test_same_shape_errors_share_one_fingerprint(self) -> None: | |
| """同一类报错每次 uuid / 数字不同,不应各记一条。""" | |
| service = self._service() | |
| service.translate("/api/x failed id=550e8400-e29b-41d4-a716-446655440000 code=429") | |
| service.translate("/api/x failed id=6ba7b810-9dad-11d1-80b4-00c04fd430c8 code=500") | |
| items = service.list_unconfigured() | |
| self.assertEqual(len(items), 1) | |
| self.assertEqual(items[0]["count"], 2) | |
| def test_list_sorted_by_count_desc(self) -> None: | |
| service = self._service() | |
| service.translate("alpha error occurred") | |
| service.translate("beta error occurred") | |
| service.translate("beta error occurred") | |
| service.translate("beta error occurred") | |
| items = service.list_unconfigured() | |
| self.assertEqual(items[0]["count"], 3) | |
| self.assertEqual(items[1]["count"], 1) | |
| def test_list_includes_keyword_candidates(self) -> None: | |
| """待办列表要带上可直接用的候选关键词,省得管理员手抠。""" | |
| service = self._service() | |
| service.translate('beta failed: status=429, body={"reason": "some_specific_reason"}') | |
| items = service.list_unconfigured() | |
| self.assertEqual(len(items), 1) | |
| self.assertIn("some_specific_reason", items[0]["keywords"]) | |
| self.assertIn("status=429", items[0]["keywords"]) | |
| def test_persists_across_instances(self) -> None: | |
| from services.error_message_service import ErrorMessageService | |
| service = self._service() | |
| service.translate("some unique error text") | |
| reloaded = ErrorMessageService(store_path=self._store) | |
| reloaded._rules = lambda: [] | |
| self.assertEqual(len(reloaded.list_unconfigured()), 1) | |
| def test_clear_selected_and_all(self) -> None: | |
| service = self._service() | |
| service.translate("alpha error occurred") | |
| service.translate("beta error occurred") | |
| items = service.list_unconfigured() | |
| target = items[0]["fingerprint"] | |
| self.assertEqual(service.clear_unconfigured([target]), 1) | |
| self.assertEqual(len(service.list_unconfigured()), 1) | |
| self.assertEqual(service.clear_unconfigured(), 1) | |
| self.assertEqual(service.list_unconfigured(), []) | |
| def test_entry_cap_prunes_oldest(self) -> None: | |
| from services import error_message_service as module | |
| service = self._service() | |
| original_cap = module._MAX_UNCONFIGURED_ENTRIES | |
| module._MAX_UNCONFIGURED_ENTRIES = 10 | |
| try: | |
| for index in range(30): | |
| service.translate(f"unique error number {index:04d} text") | |
| self.assertLessEqual(len(service.list_unconfigured()), 10) | |
| finally: | |
| module._MAX_UNCONFIGURED_ENTRIES = original_cap | |
| def test_corrupt_store_is_tolerated(self) -> None: | |
| self._store.write_text("{ this is not json", encoding="utf-8") | |
| service = self._service() | |
| # 坏文件不能让请求链路崩,且能继续写入 | |
| service.translate("brand new error text") | |
| self.assertEqual(len(service.list_unconfigured()), 1) | |
| def test_sample_is_truncated(self) -> None: | |
| from services import error_message_service as module | |
| service = self._service() | |
| service.translate("x" * (module._MAX_SAMPLE_CHARS + 500)) | |
| items = service.list_unconfigured() | |
| self.assertEqual(len(items[0]["sample"]), module._MAX_SAMPLE_CHARS) | |
| class KeywordCandidateExtractionTests(unittest.TestCase): | |
| """本地抽取:AI 不可用时的兜底,必须能捞出高辨识度片段。""" | |
| def test_extracts_from_real_upload_throttle_payload(self) -> None: | |
| from services.error_message_service import extract_keyword_candidates | |
| raw = ( | |
| '/backend-api/files failed: status=429, body={"detail": {"code": "throttled", ' | |
| '"error_code": "throttled", "message": "你已达到文件上传上限。请4小时 内重试。", ' | |
| '"type": "throttled", "reason": "upload_protection_limit_exceeded"}}' | |
| ) | |
| candidates = extract_keyword_candidates(raw) | |
| # 最精确的标识符排在最前 | |
| self.assertEqual(candidates[0], "upload_protection_limit_exceeded") | |
| self.assertIn("/backend-api/files", candidates) | |
| self.assertIn("status=429", candidates) | |
| def test_skips_json_key_names_and_generic_words(self) -> None: | |
| from services.error_message_service import extract_keyword_candidates | |
| raw = '{"error_code": "abc12345", "message": "failed", "reason": "xyz98765"}' | |
| candidates = extract_keyword_candidates(raw) | |
| # 字段名本身不参与匹配,否则等于没筛 | |
| for noise in ("error_code", "message", "reason", "failed", "status"): | |
| self.assertNotIn(noise, candidates) | |
| def test_extracts_curl_noise(self) -> None: | |
| from services.error_message_service import extract_keyword_candidates | |
| self.assertIn("curl: (16)", extract_keyword_candidates("upstream curl: (16) recv failure")) | |
| def test_deduplicates_and_handles_empty(self) -> None: | |
| from services.error_message_service import extract_keyword_candidates | |
| self.assertEqual(extract_keyword_candidates(""), []) | |
| self.assertEqual(extract_keyword_candidates(None), []) | |
| repeated = extract_keyword_candidates("status=429 then again status=429") | |
| self.assertEqual(repeated.count("status=429"), 1) | |
| def test_no_json_syntax_fragments_in_candidates(self) -> None: | |
| """候选词里不能出现引号/冒号/花括号 —— 那是报文语法,不是错误标识。""" | |
| from services.error_message_service import extract_keyword_candidates | |
| raw = ( | |
| '/backend-api/files failed: status=429, body={"detail": {"code": "throttled", ' | |
| '"reason": "upload_protection_limit_exceeded"}}' | |
| ) | |
| candidates = extract_keyword_candidates(raw) | |
| # 注意不能一律禁冒号:`curl: (16)` 这种是合法关键词,只禁 JSON 的 `": "` 片段。 | |
| for candidate in candidates: | |
| for noise in ('"', "'", "{", "}", '": "'): | |
| self.assertNotIn(noise, candidate, msg=f"{candidate!r} 里混进了 JSON 语法碎片") | |
| class ReportedUploadErrorTests(unittest.TestCase): | |
| """用户实际上报的 /uploaded 500 报错:从「粘贴原文」到「规则真能命中」必须闭环。 | |
| 这条报错是多模态里最常见的 500(file_id 每次不同),用来守住: | |
| 1. 本地抽取能捞出高辨识度片段; | |
| 2. 模型编造的关键词会被丢弃; | |
| 3. 归纳出来的规则确实能命中原始报错(子串匹配)。 | |
| """ | |
| RAW = ( | |
| "/backend-api/files/file_000000006bcc81f5b6f6f32ff45d7f55/uploaded failed: status=500, " | |
| 'body={"detail": {"code": "internal_error", "error_code": "internal_error"}}' | |
| ) | |
| def setUp(self) -> None: | |
| self._tmp = tempfile.TemporaryDirectory() | |
| self._store = Path(self._tmp.name) / "unconfigured.json" | |
| def tearDown(self) -> None: | |
| self._tmp.cleanup() | |
| def test_extracts_usable_candidates(self) -> None: | |
| from services.error_message_service import extract_keyword_candidates | |
| candidates = extract_keyword_candidates(self.RAW) | |
| self.assertIn("/backend-api/files", candidates) | |
| self.assertIn("status=500", candidates) | |
| self.assertIn("internal_error", candidates) | |
| def test_ai_suggestion_drops_hallucinated_keywords(self) -> None: | |
| from services.error_message_service import parse_ai_suggestion | |
| answer = ( | |
| '{"keywords": ["status=500", "internal_error", "definitely_absent", "error_code"], ' | |
| '"display": "文件上传失败,请稍后重试"}' | |
| ) | |
| result = parse_ai_suggestion(answer, raw=self.RAW) | |
| self.assertEqual(result["display"], "文件上传失败,请稍后重试") | |
| self.assertIn("status=500", result["keywords"]) | |
| self.assertIn("internal_error", result["keywords"]) | |
| # 原文里没有的词永远命中不了;JSON 字段名会命中所有报错,两者都要剔。 | |
| self.assertNotIn("definitely_absent", result["keywords"]) | |
| self.assertNotIn("error_code", result["keywords"]) | |
| def test_volatile_path_segments_are_trimmed(self) -> None: | |
| from services.error_message_service import _strip_volatile_path | |
| # file_id 每次都变,关键词必须停在它之前,否则规则只对这一次生效。 | |
| self.assertEqual( | |
| _strip_volatile_path("/backend-api/files/file_000000006bcc81f5b6f6f32ff45d7f55/uploaded"), | |
| "/backend-api/files", | |
| ) | |
| self.assertEqual(_strip_volatile_path("/backend-api/files"), "/backend-api/files") | |
| # 单段路径太泛,直接丢弃。 | |
| self.assertEqual(_strip_volatile_path("/files"), "") | |
| def test_generated_rule_matches_the_reported_error(self) -> None: | |
| from services.error_message_service import ErrorMessageService, extract_keyword_candidates | |
| service = ErrorMessageService(store_path=self._store) | |
| service._rules = lambda: __import__( | |
| "services.error_message_service", fromlist=["normalize_rules"] | |
| ).normalize_rules( | |
| [ | |
| { | |
| "keywords": extract_keyword_candidates(self.RAW), | |
| "display": "文件上传失败,请稍后重试", | |
| } | |
| ] | |
| ) | |
| # file_id 每次都不同,命中靠的是路径前缀 + 状态码,不是整条原文。 | |
| other = ( | |
| "/backend-api/files/file_00000000aaaaaaaaaaaaaaaaaaaaaaaa/uploaded failed: status=500, " | |
| 'body={"detail": {"code": "internal_error"}}' | |
| ) | |
| self.assertEqual(service.translate(self.RAW), "文件上传失败,请稍后重试") | |
| self.assertEqual(service.translate(other), "文件上传失败,请稍后重试") | |
| # 429 上传限流:和上面那条 500 报错共用 /backend-api/files 前缀。 | |
| THROTTLE = ( | |
| '/backend-api/files failed: status=429, body={"detail": {"code": "throttled", ' | |
| '"error_code": "throttled", "reason": "upload_protection_limit_exceeded"}}' | |
| ) | |
| def _service_with(self, rules): | |
| from services.error_message_service import ErrorMessageService | |
| service = ErrorMessageService(store_path=self._store) | |
| service._rules = lambda: __import__( | |
| "services.error_message_service", fromlist=["normalize_rules"] | |
| ).normalize_rules(rules) | |
| return service | |
| def test_upload_500_rule_does_not_swallow_throttle_error(self) -> None: | |
| """两条规则必须正交:500 规则不能吞掉 429 限流,反之亦然。 | |
| 匹配是「任一关键词命中」,所以关键词选宽一点就会让两条规则互相覆盖。 | |
| 这是「结合本项目」的核心:路径必须带上区分步骤的片段。 | |
| """ | |
| service = self._service_with( | |
| [ | |
| {"keywords": ["internal_error", "/uploaded failed: status=500"], "display": "文件上传失败,请重试"}, | |
| {"keywords": ["upload_protection_limit_exceeded", "你已达到文件上传上限"], "display": "文件上传次数已达上限"}, | |
| ] | |
| ) | |
| self.assertEqual(service.translate(self.RAW), "文件上传失败,请重试") | |
| self.assertEqual(service.translate(self.THROTTLE), "文件上传次数已达上限") | |
| def test_broad_path_prefix_swallows_other_errors(self) -> None: | |
| """反面用例:只有裸前缀 /backend-api/files 时,500 规则会吞掉 429 限流报错。 | |
| 这条用例存在的意义是防止有人把提示词改回「只写接口路径前缀」—— | |
| 那样前台会把「配额用满」误报成「服务内部错误」,用户会一直重试而不是等换号。 | |
| """ | |
| service = self._service_with( | |
| [{"keywords": ["/backend-api/files"], "display": "服务器内部错误,请稍后重试"}] | |
| ) | |
| self.assertEqual(service.translate(self.RAW), "服务器内部错误,请稍后重试") | |
| # 这就是误伤:限流报错也被换成了 500 的文案。 | |
| self.assertEqual(service.translate(self.THROTTLE), "服务器内部错误,请稍后重试") | |
| class AiSuggestionParsingTests(unittest.TestCase): | |
| def test_parses_plain_json(self) -> None: | |
| from services.error_message_service import parse_ai_suggestion | |
| result = parse_ai_suggestion('{"keywords": ["aa", "bb"], "display": "出错了"}') | |
| self.assertEqual(result["keywords"], ["aa", "bb"]) | |
| self.assertEqual(result["display"], "出错了") | |
| def test_strips_markdown_code_fence(self) -> None: | |
| from services.error_message_service import parse_ai_suggestion | |
| result = parse_ai_suggestion('```json\n{"keywords": ["aa"], "display": "出错了"}\n```') | |
| self.assertEqual(result["keywords"], ["aa"]) | |
| def test_tolerates_surrounding_prose(self) -> None: | |
| from services.error_message_service import parse_ai_suggestion | |
| result = parse_ai_suggestion('好的,结果如下:\n{"keywords": ["aa"], "display": "出错了"}\n希望有帮助') | |
| self.assertEqual(result["display"], "出错了") | |
| def test_drops_keywords_absent_from_raw(self) -> None: | |
| """关键在原文里必须真实出现,否则规则永远命中不了。""" | |
| from services.error_message_service import parse_ai_suggestion | |
| result = parse_ai_suggestion( | |
| '{"keywords": ["upload_protection_limit_exceeded", "hallucinated_token"], "display": "太挤了"}', | |
| raw="/backend-api/files failed: upload_protection_limit_exceeded", | |
| ) | |
| self.assertEqual(result["keywords"], ["upload_protection_limit_exceeded"]) | |
| def test_bad_output_returns_empty(self) -> None: | |
| from services.error_message_service import parse_ai_suggestion | |
| for bad in ("", None, "抱歉我无法处理", "[1,2,3]", "{ not json"): | |
| self.assertEqual(parse_ai_suggestion(bad, raw="x")["keywords"], []) | |
| self.assertEqual(parse_ai_suggestion(bad, raw="x")["display"], "") | |
| def test_prompt_forbids_generic_keywords(self) -> None: | |
| from services.error_message_service import ERROR_SUMMARY_SYSTEM_PROMPT | |
| for word in ("error", "failed", "exception"): | |
| self.assertIn(word, ERROR_SUMMARY_SYSTEM_PROMPT) | |
| def test_sanitizes_json_fragment_keywords(self) -> None: | |
| """模型会把 `reason": "upload_xxx` 这种截断片段整条当关键词。 | |
| 它虽然能把当前这条报错匹配上,但只要上游 JSON 的字段顺序或空格一变 | |
| 就失效;归纳链路必须只保留「值」本身。 | |
| """ | |
| from services.error_message_service import parse_ai_suggestion | |
| raw = '/backend-api/files failed: status=429, body={"detail": {"code": "throttled", "reason": "upload_protection_limit_exceeded"}}' | |
| result = parse_ai_suggestion( | |
| json.dumps({ | |
| "keywords": ['reason": "upload_protection_limit_exceeded', "status=429"], | |
| "display": "上传频率过高,请稍后再试", | |
| }), | |
| raw=raw, | |
| ) | |
| self.assertEqual(result["keywords"], ["upload_protection_limit_exceeded", "status=429"]) | |
| def test_prompt_carries_project_context(self) -> None: | |
| """归纳要「结合本项目」:必须告诉模型匹配语义、报文格式和路径步骤要求。 | |
| 少了这些,模型会把 /backend-api/files 这种通用前缀当关键词; | |
| 而匹配是「任一命中」,一条规则就会吞掉一大片语义完全不同的报错。 | |
| """ | |
| from services.error_message_service import ERROR_SUMMARY_SYSTEM_PROMPT | |
| self.assertIn("任意一个", ERROR_SUMMARY_SYSTEM_PROMPT) | |
| self.assertIn("failed: status=", ERROR_SUMMARY_SYSTEM_PROMPT) | |
| for step in ("/uploaded", "/download", "/library"): | |
| self.assertIn(step, ERROR_SUMMARY_SYSTEM_PROMPT) | |
| self.assertIn("file_id", ERROR_SUMMARY_SYSTEM_PROMPT) | |
| class ConfigIntegrationTests(unittest.TestCase): | |
| def test_config_exposes_rule_helpers(self) -> None: | |
| """config 的归一化入口要能安全处理脏数据。""" | |
| from services.config import _normalize_error_message_rules | |
| self.assertEqual(_normalize_error_message_rules(None), []) | |
| self.assertEqual(_normalize_error_message_rules([{"keywords": [], "display": ""}]), []) | |
| result = _normalize_error_message_rules([{"keywords": ["aa"], "display": "bb"}]) | |
| self.assertEqual(result[0]["keywords"], ["aa"]) | |
| self.assertEqual(result[0]["display"], "bb") | |
| def test_shared_settings_keys_registered(self) -> None: | |
| from services.config import SHARED_SETTINGS_KEYS | |
| self.assertIn("error_message_rules", SHARED_SETTINGS_KEYS) | |
| self.assertIn("error_message_fallback_enabled", SHARED_SETTINGS_KEYS) | |
| if __name__ == "__main__": | |
| unittest.main() | |