diff --git a/apps/api/tests/test_phase25_auto_harvesting.py b/apps/api/tests/test_phase25_auto_harvesting.py new file mode 100644 index 00000000..e4c381e8 --- /dev/null +++ b/apps/api/tests/test_phase25_auto_harvesting.py @@ -0,0 +1,155 @@ +""" +Phase 25: Auto-Harvesting 測試 +================================ +ADR-052 / Phase 25 P1: NemotronRunbookGenerator + AntiPattern Gate + +測試策略: Source code inspection (禁止 Mock，禁止 import 模組) +- 驗證 runbook_generator.py 實作結構 +- 驗證 auto_repair_service.py fire-and-forget 模式 +- 驗證 knowledge.py 模型欄位 +- 驗證 knowledge_service.py check_anti_pattern 邏輯 + +建立時間: 2026-04-05 (台北時區) +建立者: Claude Code (Phase 25 Auto-Harvesting 測試) +""" + +from pathlib import Path + +# Source file paths +_BASE = Path(__file__).parent.parent / "src" +_RUNBOOK_GEN = _BASE / "services" / "runbook_generator.py" +_AUTO_REPAIR = _BASE / "services" / "auto_repair_service.py" +_PLAYBOOK = _BASE / "models" / "playbook.py" +_KNOWLEDGE = _BASE / "models" / "knowledge.py" +_KNOWLEDGE_SVC = _BASE / "services" / "knowledge_service.py" + + +# ============================================================================= +# TestRunbookGeneratorModule — 模組存在與結構 +# ============================================================================= + +class TestRunbookGeneratorModule: + """驗證 runbook_generator.py 模組存在並有正確結構""" + + def test_runbook_generator_module_exists(self): + """src/services/runbook_generator.py 存在""" + assert _RUNBOOK_GEN.exists(), f"找不到 {_RUNBOOK_GEN}" + + def test_runbook_generator_class_exists(self): + """NemotronRunbookGenerator class 已定義""" + source = _RUNBOOK_GEN.read_text() + assert "class NemotronRunbookGenerator" in source + + def test_generate_runbook_method_exists(self): + """generate_runbook() 方法存在""" + source = _RUNBOOK_GEN.read_text() + assert "async def generate_runbook" in source or "def generate_runbook" in source + + def test_generate_anti_pattern_method_exists(self): + """generate_anti_pattern() 方法存在""" + source = _RUNBOOK_GEN.read_text() + assert "async def generate_anti_pattern" in source or "def generate_anti_pattern" in source + + def test_runbook_uses_auto_runbook_type(self): + """runbook_generator.py 使用 'auto_runbook' entry type""" + source = _RUNBOOK_GEN.read_text() + assert "auto_runbook" in source + + def test_anti_pattern_uses_anti_pattern_type(self): + """runbook_generator.py 使用 'anti_pattern' entry type""" + source = _RUNBOOK_GEN.read_text() + assert "anti_pattern" in source + + def test_runbook_status_is_draft(self): + """成功修復 → AUTO_RUNBOOK 狀態為 DRAFT（需人工審核）""" + source = _RUNBOOK_GEN.read_text() + assert "DRAFT" in source or "draft" in source + + def test_anti_pattern_status_is_published(self): + """失敗修復 → ANTI_PATTERN 狀態為 PUBLISHED（直接發布）""" + source = _RUNBOOK_GEN.read_text() + assert "PUBLISHED" in source or "published" in source + + def test_runbook_uses_nvidia_chat(self): + """runbook_generator.py 使用 nvidia provider""" + source = _RUNBOOK_GEN.read_text() + assert "nvidia" in source + + def test_minimal_fallback_exists(self): + """Nemotron 失敗時有 fallback 邏輯""" + source = _RUNBOOK_GEN.read_text() + assert "fallback" in source + + +# ============================================================================= +# TestAutoRepairService — fire-and-forget 與 GC 防洩漏 +# ============================================================================= + +class TestAutoRepairService: + """驗證 auto_repair_service.py 的 fire-and-forget 模式與 GC 防洩漏""" + + def test_fire_and_forget_pattern(self): + """auto_repair_service.py 使用 asyncio.create_task() 實現 fire-and-forget""" + source = _AUTO_REPAIR.read_text() + assert "create_task" in source + + def test_pending_tasks_gc_prevention(self): + """auto_repair_service.py 持有 _pending_tasks 防止 Task 被 GC 回收""" + source = _AUTO_REPAIR.read_text() + assert "_pending_tasks" in source + + def test_anti_pattern_gate_in_auto_repair(self): + """auto_repair_service.py 有 check_anti_pattern 攔截呼叫（AntiPattern Gate）""" + source = _AUTO_REPAIR.read_text() + assert "check_anti_pattern" in source + + +# ============================================================================= +# TestKnowledgeModels — 知識庫模型結構 +# ============================================================================= + +class TestKnowledgeModels: + """驗證 models/knowledge.py 的 EntryType 與 KnowledgeEntry 欄位""" + + def test_auto_runbook_entrytype_value(self): + """EntryType 有 AUTO_RUNBOOK = 'auto_runbook'""" + source = _KNOWLEDGE.read_text() + assert 'AUTO_RUNBOOK' in source + assert '"auto_runbook"' in source or "'auto_runbook'" in source + + def test_anti_pattern_entrytype_value(self): + """EntryType 有 ANTI_PATTERN = 'anti_pattern'""" + source = _KNOWLEDGE.read_text() + assert 'ANTI_PATTERN' in source + assert '"anti_pattern"' in source or "'anti_pattern'" in source + + def test_knowledge_entry_has_symptoms_hash(self): + """KnowledgeEntry 有 symptoms_hash 欄位（AntiPattern 閉環用）""" + source = _KNOWLEDGE.read_text() + assert "symptoms_hash" in source + + +# ============================================================================= +# TestPlaybookModel — 症狀 hash 計算 +# ============================================================================= + +class TestPlaybookModel: + """驗證 models/playbook.py 的 compute_hash 方法""" + + def test_symptom_hash_compute_method(self): + """SymptomPattern 有 compute_hash() 方法""" + source = _PLAYBOOK.read_text() + assert "compute_hash" in source + + +# ============================================================================= +# TestKnowledgeService — check_anti_pattern 查詢 +# ============================================================================= + +class TestKnowledgeService: + """驗證 knowledge_service.py 的 check_anti_pattern 方法""" + + def test_check_anti_pattern_in_knowledge_service(self): + """knowledge_service.py 有 check_anti_pattern 方法""" + source = _KNOWLEDGE_SVC.read_text() + assert "check_anti_pattern" in source diff --git a/apps/api/tests/test_phase25_drift_detection.py b/apps/api/tests/test_phase25_drift_detection.py new file mode 100644 index 00000000..ce8ed11e --- /dev/null +++ b/apps/api/tests/test_phase25_drift_detection.py @@ -0,0 +1,158 @@ +""" +Phase 25: Config Drift Detection 測試 +======================================= +ADR-052 / Phase 25 P2: DriftDetector + NemotronDriftInterpreter + DriftRemediator + +測試策略: Source code inspection (禁止 Mock，禁止 import 模組) +- 驗證 drift_detector.py 結構（DriftDetector, GitStateReader, K8sStateReader） +- 驗證 drift_interpreter.py (NemotronDriftInterpreter, read-only) +- 驗證 drift_remediator.py (kubectl apply / git push) +- 驗證 api/v1/drift.py 端點 +- 驗證 ai_router.py DIAGNOSE local-only 鏈路 + +建立時間: 2026-04-05 (台北時區) +建立者: Claude Code (Phase 25 Drift Detection 測試) +""" + +from pathlib import Path + +# Source file paths +_BASE = Path(__file__).parent.parent / "src" +_DETECTOR = _BASE / "services" / "drift_detector.py" +_ANALYZER = _BASE / "services" / "drift_analyzer.py" +_INTERPRETER = _BASE / "services" / "drift_interpreter.py" +_REMEDIATOR = _BASE / "services" / "drift_remediator.py" +_DRIFT_API = _BASE / "api" / "v1" / "drift.py" +_AI_ROUTER = _BASE / "services" / "ai_router.py" + + +# ============================================================================= +# TestDriftDetector — 偵測器結構 +# ============================================================================= + +class TestDriftDetector: + """驗證 drift_detector.py 的核心結構""" + + def test_drift_detector_module_exists(self): + """src/services/drift_detector.py 存在""" + assert _DETECTOR.exists(), f"找不到 {_DETECTOR}" + + def test_drift_detector_class_exists(self): + """DriftDetector class 已定義""" + source = _DETECTOR.read_text() + assert "class DriftDetector" in source + + def test_drift_scan_method_exists(self): + """DriftDetector 有 scan() 方法""" + source = _DETECTOR.read_text() + assert "async def scan" in source or "def scan" in source + + def test_git_state_reader_exists(self): + """GitStateReader class 已定義""" + source = _DETECTOR.read_text() + assert "class GitStateReader" in source + + def test_k8s_state_reader_exists(self): + """K8sStateReader class 已定義""" + source = _DETECTOR.read_text() + assert "class K8sStateReader" in source + + def test_drift_level_high_defined(self): + """DriftLevel.HIGH 已定義""" + source = _DETECTOR.read_text() + assert "HIGH" in source + + def test_drift_level_medium_defined(self): + """DriftLevel.MEDIUM 已定義""" + source = _DETECTOR.read_text() + assert "MEDIUM" in source + + +# ============================================================================= +# TestDriftInterpreter — 解讀器 (read-only，不執行修復) +# ============================================================================= + +class TestDriftInterpreter: + """驗證 drift_interpreter.py 的 NemotronDriftInterpreter，且不含 kubectl apply""" + + def test_drift_interpreter_module_exists(self): + """src/services/drift_interpreter.py 存在""" + assert _INTERPRETER.exists(), f"找不到 {_INTERPRETER}" + + def test_drift_interpreter_class_exists(self): + """NemotronDriftInterpreter class 已定義""" + source = _INTERPRETER.read_text() + assert "class NemotronDriftInterpreter" in source + + def test_drift_interpreter_no_fix_commands(self): + """drift_interpreter.py 不含 'kubectl apply'（只分析，不執行修復）""" + source = _INTERPRETER.read_text() + assert "kubectl apply" not in source, ( + "drift_interpreter.py 不應含 kubectl apply — 解讀器只做 read-only 分析" + ) + + +# ============================================================================= +# TestDriftRemediator — 修復器 (kubectl apply / git push) +# ============================================================================= + +class TestDriftRemediator: + """驗證 drift_remediator.py 的 rollback 與 adopt 操作""" + + def test_drift_remediator_module_exists(self): + """src/services/drift_remediator.py 存在""" + assert _REMEDIATOR.exists(), f"找不到 {_REMEDIATOR}" + + def test_drift_remediator_rollback_uses_kubectl(self): + """rollback() 使用 kubectl apply 覆蓋回 Git 狀態""" + source = _REMEDIATOR.read_text() + assert "kubectl apply" in source + + def test_drift_remediator_adopt_uses_git_push(self): + """adopt() 使用 git push 承認變更並更新 Git""" + source = _REMEDIATOR.read_text() + assert "git push" in source + + +# ============================================================================= +# TestDriftApi — API 端點 +# ============================================================================= + +class TestDriftApi: + """驗證 api/v1/drift.py 端點路徑""" + + def test_drift_api_router_exists(self): + """src/api/v1/drift.py 存在""" + assert _DRIFT_API.exists(), f"找不到 {_DRIFT_API}" + + def test_drift_scan_endpoint_exists(self): + """drift.py 有 /scan 端點""" + source = _DRIFT_API.read_text() + assert '"/scan"' in source or "'/scan'" in source + + def test_drift_reports_endpoint_exists(self): + """drift.py 有 /reports 端點""" + source = _DRIFT_API.read_text() + assert '"/reports"' in source or "'/reports'" in source + + +# ============================================================================= +# TestAiRouterDiagnose — DIAGNOSE local-only 鏈路 +# ============================================================================= + +class TestAiRouterDiagnose: + """驗證 ai_router.py 的 DIAGNOSE 隱私邊界 local-only fallback""" + + def test_local_fallback_chain_exists(self): + """ai_router.py 有 _local_fallback_chain（DIAGNOSE 隱私邊界用）""" + source = _AI_ROUTER.read_text() + assert "_local_fallback_chain" in source + + def test_diagnose_uses_local_chain(self): + """DIAGNOSE task type 使用 local fallback chain""" + source = _AI_ROUTER.read_text() + assert "DIAGNOSE" in source + # DIAGNOSE 分支必須引用 _local_fallback_chain + idx_diagnose = source.find("DIAGNOSE") + remaining = source[idx_diagnose:] + assert "_local_fallback_chain" in remaining