from __future__ import annotations import json from pathlib import Path import tempfile import unittest from ai_triage.triage import TriageEngine VALID_AI_RESULT = { "category": "product_error", "component": "dedicated-control-plane", "problem_key": "cluster-stuck-creating", "confidence": 0.9, "summary": "创建集群接口返回了内部错误", "evidence": ["POST /v1/clusters 实际返回 500，预期返回 200"], "root_cause": "现有证据表明集群创建链路发生了回归", "recommendations": ["根据 trace-123 检查服务端调用链"], "oncall_team": "global_control", } class StubACP: def __init__(self, result: str | Exception) -> None: self.result = result self.calls: list[tuple[str, str]] = [] def prompt(self, base_prompt: str, user_prompt: str) -> str: self.calls.append((base_prompt, user_prompt)) if isinstance(self.result, Exception): raise self.result return self.result class StubApiTeamRouter: def __init__(self, team: str | None) -> None: self.team = team self.calls: list[object] = [] def resolve(self, request_url: object) -> str | None: self.calls.append(request_url) return self.team class TriageEngineTests(unittest.TestCase): def knowledge_file(self, directory: str) -> Path: path = Path(directory) / "knowledge.md" path.write_text("Only scenario execution failures can be product errors.", encoding="utf-8") return path def test_accepts_strict_json_inside_markdown_fence(self) -> None: with tempfile.TemporaryDirectory() as directory: acp = StubACP("```json\n" + json.dumps(VALID_AI_RESULT) + "\n```") engine = TriageEngine(acp, self.knowledge_file(directory), owners={}) result = engine.classify({"request_url": "https://example.invalid/v1"}, scenario_step_allowed=True) self.assertEqual("product_error", result["category"]) self.assertEqual("case", result["owner_type"]) self.assertEqual("dedicated-control-plane", result["component"]) self.assertEqual("cluster-stuck-creating", result["problem_key"]) self.assertEqual("global_control", result["oncall_team"]) self.assertEqual("未配置", result["owner"]) self.assertIn("不可信", acp.calls[0][1]) def test_loads_curated_knowledge_files_into_the_base_prompt(self) -> None: with tempfile.TemporaryDirectory() as directory: primary = self.knowledge_file(directory) framework = Path(directory) / "scenario-framework.md" framework.write_text( "Framework fact: retries must use a bounded non-zero delay.", encoding="utf-8", ) acp = StubACP(json.dumps(VALID_AI_RESULT)) engine = TriageEngine( acp, primary, owners={}, knowledge_paths=(framework,), ) engine.classify({}, scenario_step_allowed=True) base_prompt, _ = acp.calls[0] self.assertIn("Only scenario execution failures", base_prompt) self.assertIn("Framework fact: retries", base_prompt) def test_prompt_requires_non_overlapping_summary_and_root_cause(self) -> None: with tempfile.TemporaryDirectory() as directory: acp = StubACP(json.dumps(VALID_AI_RESULT)) engine = TriageEngine(acp, self.knowledge_file(directory), owners={}) engine.classify({}, scenario_step_allowed=True) _, user_prompt = acp.calls[0] self.assertIn("所有面向用户的自然语言字段必须使用简体中文", user_prompt) self.assertIn("不得重复", user_prompt) self.assertIn( "evidence 和 recommendations 数组中的每一项都必须包含简体中文说明", user_prompt, ) def test_prompt_treats_related_metadata_as_evidence_not_instructions(self) -> None: with tempfile.TemporaryDirectory() as directory: acp = StubACP(json.dumps(VALID_AI_RESULT)) engine = TriageEngine(acp, self.knowledge_file(directory), owners={}) engine.classify( { "related_metadata": { "clusterId": ["cluster-1"], "restoreId": ["restore-1"], } }, scenario_step_allowed=True, ) _, user_prompt = acp.calls[0] self.assertIn("related_metadata", user_prompt) self.assertIn("不得编造", user_prompt) self.assertIn("cluster-1", user_prompt) def test_prompt_lists_only_current_oncall_teams_for_product_routing(self) -> None: with tempfile.TemporaryDirectory() as directory: acp = StubACP(json.dumps(VALID_AI_RESULT)) engine = TriageEngine( acp, self.knowledge_file(directory), owners={}, product_teams={ "global_control": "SOE - Global & Dedicated", "premium": "SOE - Essential & Premium & TiOperator", }, ) engine.classify( {"request_url": "https://api.example.invalid/v1beta2/tidbs"}, scenario_step_allowed=True, ) _, user_prompt = acp.calls[0] self.assertIn("global_control", user_prompt) self.assertIn("SOE - Global & Dedicated", user_prompt) self.assertIn("请求 URL", user_prompt) def test_rejects_product_team_not_present_in_current_api_catalog(self) -> None: invalid = dict(VALID_AI_RESULT, oncall_team="invented-team") with tempfile.TemporaryDirectory() as directory: engine = TriageEngine( StubACP(json.dumps(invalid)), self.knowledge_file(directory), owners={}, product_teams={ "global_control": "SOE - Global & Dedicated", }, ) result = engine.classify({}, scenario_step_allowed=True) self.assertEqual("fallback_invalid_output", result["analysis_status"]) def test_accepts_missing_oncall_team_as_partial_ai_result(self) -> None: partial = { key: value for key, value in VALID_AI_RESULT.items() if key != "oncall_team" } with tempfile.TemporaryDirectory() as directory: engine = TriageEngine( StubACP(json.dumps(partial)), self.knowledge_file(directory), owners={"case": "case-owner"}, product_teams={ "global_control": "SOE - Global & Dedicated", }, ) result = engine.classify({}, scenario_step_allowed=True) self.assertEqual("product_error", result["category"]) self.assertEqual(VALID_AI_RESULT["summary"], result["summary"]) self.assertEqual(["oncall_team"], result["missing_fields"]) self.assertEqual("ai_partial", result["analysis_status"]) self.assertTrue(result["ai_used"]) self.assertEqual("unknown", result["oncall_team"]) self.assertEqual(("case", "case-owner"), (result["owner_type"], result["owner"])) def test_empty_ai_object_uses_invalid_output_fallback(self) -> None: with tempfile.TemporaryDirectory() as directory: engine = TriageEngine( StubACP("{}"), self.knowledge_file(directory), owners={"case": "case-owner"}, ) result = engine.classify({}, scenario_step_allowed=True) self.assertEqual("case_error", result["category"]) self.assertEqual("fallback_invalid_output", result["analysis_status"]) self.assertFalse(result["ai_used"]) self.assertNotIn("missing_fields", result) def test_api_path_mapping_overrides_ai_team_choice(self) -> None: ai_result = dict(VALID_AI_RESULT, oncall_team="invented-team") router = StubApiTeamRouter("premium") with tempfile.TemporaryDirectory() as directory: acp = StubACP(json.dumps(ai_result)) engine = TriageEngine( acp, self.knowledge_file(directory), owners={}, product_teams={ "premium": "SOE - Essential & Premium & TiOperator", }, api_team_router=router, ) result = engine.classify( { "request_url": ( "https://example.invalid/api/v2025/cloud/v1beta2/" "tidbs/123:resetRootPassword?orgId=456" ) }, scenario_step_allowed=True, ) self.assertEqual("premium", result["oncall_team"]) self.assertEqual("ai_validated", result["analysis_status"]) self.assertEqual(1, len(router.calls)) self.assertIn("确定性 API 路径映射已命中：premium", acp.calls[0][1]) def test_unmatched_api_path_keeps_ai_team_choice(self) -> None: router = StubApiTeamRouter(None) with tempfile.TemporaryDirectory() as directory: engine = TriageEngine( StubACP(json.dumps(VALID_AI_RESULT)), self.knowledge_file(directory), owners={}, product_teams={ "global_control": "SOE - Global & Dedicated", "premium": "SOE - Essential & Premium & TiOperator", }, api_team_router=router, ) result = engine.classify( {"request_url": "https://example.invalid/unmapped"}, scenario_step_allowed=True, ) self.assertEqual("global_control", result["oncall_team"]) def test_rejects_ai_output_with_english_user_facing_text(self) -> None: english_result = { **VALID_AI_RESULT, "summary": "The API returned HTTP 500", "evidence": ["POST /v1/clusters returned 500"], "root_cause": "The create path regressed", "recommendations": ["Inspect the server trace"], } with tempfile.TemporaryDirectory() as directory: engine = TriageEngine( StubACP(json.dumps(english_result)), self.knowledge_file(directory), owners={}, ) result = engine.classify({}, scenario_step_allowed=True) self.assertEqual("fallback_invalid_output", result["analysis_status"]) self.assertEqual("AI 分析结果不可用，已按保守策略归类。", result["summary"]) def test_non_scenario_gate_cannot_be_overridden_by_ai(self) -> None: with tempfile.TemporaryDirectory() as directory: engine = TriageEngine( StubACP(json.dumps(VALID_AI_RESULT)), self.knowledge_file(directory), owners={"infrastructure": "infra-owner"}, ) result = engine.classify({"failed_step": "Checkout"}, scenario_step_allowed=False) self.assertEqual("infrastructure_error", result["category"]) self.assertEqual("infrastructure", result["owner_type"]) self.assertEqual("infra-owner", result["owner"]) self.assertEqual("gha-runner", result["component"]) self.assertEqual("checkout", result["problem_key"]) self.assertEqual(1.0, result["confidence"]) self.assertEqual("失败发生在允许的场景执行步骤之外。", result["summary"]) self.assertIn("确定性规则", result["root_cause"]) self.assertTrue(all("检查" in item for item in result["recommendations"])) def test_invalid_ai_json_uses_safe_low_confidence_fallback(self) -> None: with tempfile.TemporaryDirectory() as directory: engine = TriageEngine( StubACP("not json"), self.knowledge_file(directory), owners={"case": "case-owner", "infrastructure": "infra-owner"}, ) result = engine.classify({}, scenario_step_allowed=True) self.assertEqual("case_error", result["category"]) self.assertLessEqual(result["confidence"], 0.25) self.assertEqual(("case", "case-owner"), (result["owner_type"], result["owner"])) self.assertIn("fallback", result["analysis_status"]) def test_scenario_ai_cannot_claim_infrastructure_without_runner_evidence( self, ) -> None: ai_result = dict( VALID_AI_RESULT, category="infrastructure_error", component="gha-runner", oncall_team="unknown", ) with tempfile.TemporaryDirectory() as directory: result = TriageEngine( StubACP(json.dumps(ai_result)), self.knowledge_file(directory), owners={"case": "case-owner", "infrastructure": "infra-owner"}, ).classify( { "failed_step": "Run scenario cases", "error": "现有证据不足，无法区分产品与用例问题", }, scenario_step_allowed=True, ) self.assertEqual("case_error", result["category"]) self.assertEqual(("case", "case-owner"), (result["owner_type"], result["owner"])) def test_confirmed_runner_failure_overrides_scenario_ai_category(self) -> None: ai_result = dict( VALID_AI_RESULT, category="case_error", component="api-testgen", oncall_team="unknown", ) with tempfile.TemporaryDirectory() as directory: result = TriageEngine( StubACP(json.dumps(ai_result)), self.knowledge_file(directory), owners={"case": "case-owner", "infrastructure": "infra-owner"}, ).classify( { "failed_step": "Run scenario cases", "error": ( "The self-hosted runner lost communication with the " "GitHub Actions service." ), }, scenario_step_allowed=True, ) self.assertEqual("infrastructure_error", result["category"]) self.assertEqual( ("infrastructure", "infra-owner"), (result["owner_type"], result["owner"]), ) self.assertEqual("unknown", result["oncall_team"]) def test_schema_rejects_extra_fields_and_wrong_types(self) -> None: invalid = dict(VALID_AI_RESULT, confidence="high", unexpected=True) with tempfile.TemporaryDirectory() as directory: engine = TriageEngine(StubACP(json.dumps(invalid)), self.knowledge_file(directory), owners={}) result = engine.classify({}, scenario_step_allowed=True) self.assertEqual("case_error", result["category"]) self.assertEqual("fallback_invalid_output", result["analysis_status"]) def test_routes_case_and_product_owners(self) -> None: with tempfile.TemporaryDirectory() as directory: knowledge = self.knowledge_file(directory) owners = {"product": "component-a", "case": "case-a"} product = TriageEngine(StubACP(json.dumps(VALID_AI_RESULT)), knowledge, owners).classify({}, True) case_json = dict( VALID_AI_RESULT, category="case_error", oncall_team="unknown", ) case = TriageEngine(StubACP(json.dumps(case_json)), knowledge, owners).classify({}, True) self.assertEqual(("component", "component-a"), (product["owner_type"], product["owner"])) self.assertEqual(("case", "case-a"), (case["owner_type"], case["owner"])) def test_unknown_product_component_falls_back_to_case_owner(self) -> None: for component in ("unknown", "unavailable", "unassigned"): with self.subTest(component=component), tempfile.TemporaryDirectory() as directory: ai_result = dict( VALID_AI_RESULT, component=component, oncall_team="unknown", ) result = TriageEngine( StubACP(json.dumps(ai_result)), self.knowledge_file(directory), owners={"product": "component-owner", "case": "case-owner"}, ).classify({}, scenario_step_allowed=True) self.assertEqual("product_error", result["category"]) self.assertEqual( ("case", "case-owner"), (result["owner_type"], result["owner"]), ) def test_unknown_product_component_without_config_stays_case_owner_type( self, ) -> None: ai_result = dict( VALID_AI_RESULT, component="unknown", oncall_team="unknown", ) with tempfile.TemporaryDirectory() as directory: result = TriageEngine( StubACP(json.dumps(ai_result)), self.knowledge_file(directory), owners={}, ).classify({}, scenario_step_allowed=True) self.assertEqual(("case", "未配置"), (result["owner_type"], result["owner"])) def test_product_without_resolvable_owner_falls_back_to_case_owner(self) -> None: with tempfile.TemporaryDirectory() as directory: result = TriageEngine( StubACP(json.dumps(VALID_AI_RESULT)), self.knowledge_file(directory), owners={"case": "case-owner"}, ).classify({}, scenario_step_allowed=True) self.assertEqual(("case", "case-owner"), (result["owner_type"], result["owner"])) def test_production_knowledge_defaults_insufficient_scenario_evidence_to_case(self) -> None: knowledge = ( Path(__file__).parents[1] / "prompts" / "triage_knowledge.md" ).read_text(encoding="utf-8") normalized = " ".join(knowledge.split()) self.assertIn( "Insufficient evidence inside the scenario execution boundary defaults to " "`case_error`", normalized, ) self.assertIn( "Reserve `infrastructure_error` for confirmed GitHub Actions runner or " "GitHub Actions platform failures", normalized, ) self.assertNotIn( "evidence is insufficient to safely distinguish product from case", normalized, ) def test_redacts_validated_ai_text_before_returning_it(self) -> None: ai_result = dict( VALID_AI_RESULT, summary="用户 user@example.com 的请求失败", root_cause="鉴权头包含 Authorization: Bearer should-not-leak", ) with tempfile.TemporaryDirectory() as directory: engine = TriageEngine( StubACP(json.dumps(ai_result)), self.knowledge_file(directory), owners={} ) result = engine.classify({}, scenario_step_allowed=True) serialized = json.dumps(result) self.assertNotIn("user@example.com", serialized) self.assertNotIn("should-not-leak", serialized) def test_limits_combined_knowledge_and_evidence_prompt_size(self) -> None: with tempfile.TemporaryDirectory() as directory: knowledge = Path(directory) / "knowledge.md" knowledge.write_text("k" * 2_000, encoding="utf-8") acp = StubACP(json.dumps(VALID_AI_RESULT)) engine = TriageEngine( acp, knowledge, owners={}, max_prompt_chars=600, ) engine.classify( {"response_body": "r" * 2_000, "expected": "e" * 2_000}, scenario_step_allowed=True, ) base_prompt, user_prompt = acp.calls[0] self.assertLessEqual(len(base_prompt) + len(user_prompt), 600) if __name__ == "__main__": unittest.main()