test(phase25): Phase 25 P1/P2 source code inspection tests (36 tests)

- test_phase25_auto_harvesting.py: 18 tests for NemotronRunbookGenerator, AntiPattern gate, fire-and-forget pattern, symptoms_hash - test_phase25_drift_detection.py: 18 tests for DriftDetector, NemotronDriftInterpreter (read-only), DriftRemediator, local fallback chain for DIAGNOSE Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-05 00:14:50 +08:00
parent cd5547f5eb
commit 4bc4757fdc
2 changed files with 313 additions and 0 deletions
--- a/apps/api/tests/test_phase25_auto_harvesting.py
+++ 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
--- a/apps/api/tests/test_phase25_drift_detection.py
+++ 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