from __future__ import annotations import tempfile import unittest from pathlib import Path from ai_triage.config import ConfigError, enabled_targets, load_config from ai_triage.models import FailureCategory VALID_CONFIG = """ [github] owner = "tidbcloud" repo = "tidb-management-service" token_env = "GITHUB_TOKEN" [agent] command = ["opencode", "acp"] prompt_path = "prompts/triage_knowledge.md" knowledge_paths = [ "docs/knowledge/gha-triage-evidence.md", "docs/knowledge/scenario-framework/triage-context.md", ] timeout_seconds = 60 [monitor] poll_interval_seconds = 60 lookback_seconds = 86400 max_pages_per_target = 10 state_path = "ai-triage-state/state.sqlite3" lease_seconds = 600 retry_base_seconds = 30 retry_max_seconds = 1800 max_attempts = 10 retention_days = 30 [feishu] webhook_env = "FEISHU_WEBHOOK" app_id_env = "LARK_APP_ID" app_secret_env = "LARK_APP_SECRET" [oncall] api_key_env = "ONCALL_API_KEY" base_url = "https://oncall.pingcap.net" group_name = "ONCALL" timeout_seconds = 10 cache_ttl_seconds = 60 [routing] api_team_map_path = "routing/api-team-map.toml" [owners] product = "" case = "" infrastructure = "" [output] directory = "artifacts/ai-triage" [limits] max_log_bytes = 2000000 max_evidence_chars = 50000 max_agent_output_chars = 20000 max_acp_protocol_bytes = 1000000 [[targets]] id = "dedicated-gcp" enabled = true workflow = "e2e-openapi-dedicated-scenario.yaml" branch = "main" allowed_scenario_steps = ["Run scenario cases"] [[targets]] id = "disabled" enabled = false workflow = "disabled.yaml" """ class ConfigTests(unittest.TestCase): def _load(self, content: str = VALID_CONFIG): with tempfile.TemporaryDirectory() as directory: path = Path(directory, "config.toml") path.write_text(content, encoding="utf-8") return load_config(path) def test_loads_monitor_and_targets(self) -> None: config = self._load() self.assertEqual("tidbcloud", config.github.owner) self.assertEqual("2026-03-10", config.github.api_version) self.assertEqual(60, config.monitor.poll_interval_seconds) self.assertEqual(86_400, config.monitor.lookback_seconds) self.assertEqual(10, config.monitor.max_pages_per_target) self.assertEqual( ( "docs/knowledge/gha-triage-evidence.md", "docs/knowledge/scenario-framework/triage-context.md", ), config.agent.knowledge_paths, ) self.assertEqual("ai-triage-state/state.sqlite3", config.monitor.state_path) self.assertEqual("dedicated-gcp", config.targets[0].id) self.assertEqual("main", config.targets[0].branch) self.assertEqual(("Run scenario cases",), config.targets[0].allowed_scenario_steps) self.assertEqual("LARK_APP_ID", config.feishu.app_id_env) self.assertEqual("LARK_APP_SECRET", config.feishu.app_secret_env) self.assertEqual("ONCALL_API_KEY", config.oncall.api_key_env) self.assertEqual("https://oncall.pingcap.net", config.oncall.base_url) self.assertEqual("ONCALL", config.oncall.group_name) self.assertEqual( "routing/api-team-map.toml", config.routing.api_team_map_path, ) self.assertEqual(2_000_000, config.limits.max_log_bytes) self.assertEqual(1_000_000, config.limits.max_acp_protocol_bytes) self.assertEqual(FailureCategory.PRODUCT_ERROR.value, "product_error") def test_committed_example_config_includes_historical_triage_rules(self) -> None: project_root = Path(__file__).parents[1] rules_relative = "docs/knowledge/triage-experience-rules.md" config = load_config(project_root / "config.example.toml") self.assertIn(rules_relative, config.agent.knowledge_paths) rules_path = project_root / rules_relative self.assertTrue(rules_path.is_file()) rules = rules_path.read_text(encoding="utf-8") self.assertIn("signal: interrupt", rules) self.assertIn("infrastructure_error", rules) self.assertIn("其他任务", rules) self.assertIn("橙色", rules) def test_selects_only_enabled_targets(self) -> None: self.assertEqual( [target.id for target in enabled_targets(self._load())], ["dedicated-gcp"], ) def test_rejects_unknown_configuration_keys(self) -> None: content = VALID_CONFIG.replace( 'repo = "tidb-management-service"', 'repo = "tidb-management-service"\nunexpected = true', ) with self.assertRaisesRegex(ConfigError, "unexpected"): self._load(content) def test_rejects_duplicate_target_ids(self) -> None: content = VALID_CONFIG.replace('id = "disabled"', 'id = "dedicated-gcp"') with self.assertRaisesRegex(ConfigError, "duplicate target id"): self._load(content) def test_rejects_non_integer_monitor_counts(self) -> None: for field in ("max_pages_per_target", "max_attempts", "retention_days"): with self.subTest(field=field): content = VALID_CONFIG.replace(f"{field} = 10" if field != "retention_days" else f"{field} = 30", f"{field} = 1.5") with self.assertRaisesRegex(ConfigError, field): self._load(content) def test_rejects_retry_max_below_retry_base(self) -> None: content = VALID_CONFIG.replace("retry_max_seconds = 1800", "retry_max_seconds = 10") with self.assertRaisesRegex(ConfigError, "retry_max_seconds"): self._load(content) def test_rejects_retention_shorter_than_scan_lookback(self) -> None: content = VALID_CONFIG.replace("lookback_seconds = 86400", "lookback_seconds = 2678400") with self.assertRaisesRegex(ConfigError, "retention_days"): self._load(content) def test_rejects_lease_shorter_than_agent_timeout(self) -> None: content = VALID_CONFIG.replace("lease_seconds = 600", "lease_seconds = 60") with self.assertRaisesRegex(ConfigError, "lease_seconds"): self._load(content) def test_rejects_removed_initial_log_tail_bytes(self) -> None: content = VALID_CONFIG.replace( "max_log_bytes = 2000000", "max_log_bytes = 2000000\ninitial_log_tail_bytes = 100000", ) with self.assertRaisesRegex(ConfigError, "initial_log_tail_bytes"): self._load(content) def test_rejects_unsafe_target_id(self) -> None: content = VALID_CONFIG.replace('id = "dedicated-gcp"', 'id = "../dedicated"') with self.assertRaisesRegex(ConfigError, r"targets\[0\].id"): self._load(content) def test_rejects_non_string_owner(self) -> None: content = VALID_CONFIG.replace('product = ""', "product = true") with self.assertRaisesRegex(ConfigError, "owners.product"): self._load(content) def test_rejects_literal_secrets_where_environment_names_are_required(self) -> None: replacements = ( ("GITHUB_TOKEN", "github_" + "pat_not-an-environment-name"), ("ONCALL_API_KEY", "oc_v2_literal-secret"), ( "FEISHU_WEBHOOK", "https://open.feishu.cn/open-apis/bot/v2/" + "hook/secret", ), ) for expected, literal in replacements: with self.subTest(expected=expected): content = VALID_CONFIG.replace(f'"{expected}"', f'"{literal}"') with self.assertRaisesRegex(ConfigError, "environment variable name"): self._load(content) def test_rejects_empty_target_list(self) -> None: content = VALID_CONFIG.split("[[targets]]", 1)[0] with self.assertRaisesRegex(ConfigError, "targets"): self._load(content) if __name__ == "__main__": unittest.main()