"""报错文案映射服务测试。 覆盖三块： 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()