test/test_error_message_service.py

"""报错文案映射服务测试。

覆盖三块：
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()