diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index 9bad0266..dfca2c79 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -62,6 +62,16 @@ class Settings(BaseSettings): description="Phase 24: True=新 AIRouter 路由, False=舊 openclaw.py fallback chain", ) + # ========================================================================== + # W1 PR-P1: Playbook 匹配 Feature Flag (2026-04-28 ogt + Claude Sonnet 4.6) + # 修復飛輪斷鏈 C1 — proposal_service 填 matched_playbook_id → EWMA 更新 + # 回滾指令: kubectl set env deployment/awoooi-api ENABLE_PLAYBOOK_MATCHING=false + # ========================================================================== + ENABLE_PLAYBOOK_MATCHING: bool = Field( + default=True, + description="W1 PR-P1: True=generate_proposal 時執行 Playbook RAG 匹配並填 matched_playbook_id, False=行為與修復前完全相同（回滾用）", + ) + # ========================================================================== # P1-1: KMWriter 統一契約 (2026-04-28 ogt + Claude Sonnet 4.6) # KM_WRITE_AWAIT=true → 強制 await asyncio.wait_for(timeout=KM_WRITE_TIMEOUT_SECONDS) @@ -530,6 +540,16 @@ class Settings(BaseSettings): default=False, description="ADR-095: 啟用 12-Agent ConsensusEngine weights（預設關閉）", ) + # ========================================================================== + # ADR-091 Task T1: AI 自學規則雙寫 alert_rule_catalog (2026-04-28 ogt + Claude Sonnet 4.6) + # True=auto_generate_rule() 成功後同步寫入 DB source='ai_generated' + # False=回滾開關，只寫 YAML，不寫 DB + # 回滾指令: kubectl set env deployment/awoooi-api ENABLE_AI_RULE_CATALOG_WRITE=false + # ========================================================================== + ENABLE_AI_RULE_CATALOG_WRITE: bool = Field( + default=True, + description="ADR-091 T1: True=AI 自學規則雙寫 alert_rule_catalog DB, False=僅 YAML（回滾用）", + ) # 2026-04-27 P3.1-T2-PathA by Claude — DiagAggregator 信號分類層補 PDI # 路徑 A 已啟用：DA 只取 PDI 已收集的 raw 資料做業務邏輯分類（OOMKilled/CrashLoop 等）， # 不重複呼叫 K8s/SignOz API（純邏輯分類，不打外部服務）。 diff --git a/apps/api/src/services/alert_rule_engine.py b/apps/api/src/services/alert_rule_engine.py index be88925a..b3026dcf 100644 --- a/apps/api/src/services/alert_rule_engine.py +++ b/apps/api/src/services/alert_rule_engine.py @@ -581,6 +581,118 @@ def _append_rule_to_yaml(rule_yaml: str, alertname: str) -> bool: return False +def _parse_for_to_seconds(for_str: str) -> int | None: + """將 Prometheus 'for' 字串（如 '5m', '30s', '1h'）轉換為整數秒數。 + 無法解析時回傳 None。 + """ + if not for_str: + return None + for_str = for_str.strip() + mapping = {"s": 1, "m": 60, "h": 3600, "d": 86400} + m = re.fullmatch(r"(\d+)([smhd]?)", for_str) + if not m: + return None + value = int(m.group(1)) + unit = m.group(2) or "s" + return value * mapping.get(unit, 1) + + +async def _insert_catalog_ai_generated( + rule_dict: dict, + llm_source: str, + rule_id: str, + alertname_safe: str, +) -> None: + """寫入 alert_rule_catalog source='ai_generated' + + 冪等：rule_name 唯一索引（alert_rule_catalog_rule_name_key）已存在 + 使用 INSERT ON CONFLICT (rule_name) DO NOTHING + + transaction 策略：YAML + DB 不在同一 transaction + YAML 已成功 → DB 失敗只 log warning，不回滾 YAML + + review_status='draft'：對應 DB CHECK ('draft','approved','deprecated','retired') + confidence=0.50：新規則未驗證，保守初始值 + + ADR-091 Task T1 — 2026-04-28 ogt + Claude Sonnet 4.6 (Asia/Taipei) + """ + from src.core.config import settings as _settings + + if not _settings.ENABLE_AI_RULE_CATALOG_WRITE: + logger.debug( + "ai_rule_catalog_write_skipped_flag_disabled", + alertname=alertname_safe, + ) + return + + from sqlalchemy import text as _sql + import src.db.base as _db_base + import json as _json + + # 從 rule_dict 提取欄位 + # 'expr' 在 OpenClaw YAML 規則中不存在（非 PromQL）， + # 使用 alertname 作為語意佔位（與 yaml_hardcoded 同等策略） + expr = rule_dict.get("expr") or f'alertname="{alertname_safe}"' + for_str = rule_dict.get("for", "0s") + duration_seconds = _parse_for_to_seconds(str(for_str)) + labels = rule_dict.get("labels", {}) + annotations = rule_dict.get("annotations", {}) + + # 若 LLM 有產出 incident_type，注入 annotations 方便後續 T3 查詢 + incident_type = rule_dict.get("incident_type") + if incident_type and "incident_type" not in annotations: + annotations = {**annotations, "incident_type": str(incident_type)} + + # severity 從 labels 取（Prometheus 慣例），兜底空字串 + severity = (labels.get("severity", "") or "").strip()[:50] or None + + try: + async with _db_base.get_db_context() as db: + await db.execute( + _sql(""" + INSERT INTO alert_rule_catalog ( + rule_name, source, expr, duration_seconds, + severity, labels, annotations, + created_by_agent, confidence, review_status, + created_at, updated_at + ) VALUES ( + :rule_name, 'ai_generated', :expr, :duration_seconds, + :severity, + CAST(:labels AS jsonb), + CAST(:annotations AS jsonb), + :created_by_agent, 0.50, 'draft', + NOW(), NOW() + ) + ON CONFLICT (rule_name) DO NOTHING + """), + { + "rule_name": alertname_safe[:200], + "expr": expr[:4000], + "duration_seconds": duration_seconds, + "severity": severity, + "labels": _json.dumps(labels, ensure_ascii=False), + "annotations": _json.dumps(annotations, ensure_ascii=False), + "created_by_agent": llm_source, + }, + ) + logger.info( + "ai_rule_catalog_insert_success", + alertname=alertname_safe, + rule_id=rule_id, + llm_source=llm_source, + outcome="success", + ) + except Exception as db_err: + # DB 失敗只 warning，不影響已成功的 YAML 寫入 + logger.warning( + "ai_rule_catalog_insert_failed", + alertname=alertname_safe, + rule_id=rule_id, + error=str(db_err), + outcome="failed", + ) + + async def _call_ollama(prompt: str, ollama_url: str, model: str) -> str | None: """呼叫 Ollama 生成規則 YAML""" try: @@ -749,6 +861,25 @@ async def auto_generate_rule( # 立即為新規則建立 APPROVED Playbook（不等下次重啟） from src.services.playbook_seed_service import seed_playbooks_from_rules _asyncio.create_task(seed_playbooks_from_rules()) + # ADR-091 T1: 雙寫 alert_rule_catalog source='ai_generated' + # 獨立 try/except — DB 失敗不回滾已成功的 YAML 寫入 + try: + parsed_rules = yaml.safe_load(yaml_block) + rule_dict = parsed_rules[0] if isinstance(parsed_rules, list) and parsed_rules else {} + _asyncio.create_task( + _insert_catalog_ai_generated( + rule_dict=rule_dict, + llm_source=llm_source or "unknown", + rule_id=rule_id, + alertname_safe=alertname_safe, + ) + ) + except Exception as _catalog_err: + logger.warning( + "ai_rule_catalog_task_create_failed", + alertname=alertname_safe, + error=str(_catalog_err), + ) else: logger.warning( "auto_rule_auto_generate_failed", diff --git a/apps/api/src/services/proposal_service.py b/apps/api/src/services/proposal_service.py index e2061011..dd5a0647 100644 --- a/apps/api/src/services/proposal_service.py +++ b/apps/api/src/services/proposal_service.py @@ -22,6 +22,7 @@ from datetime import UTC, datetime import structlog +from src.core.config import get_settings from src.db.base import get_db_context from src.db.models import IncidentRecord from src.models.approval import ( @@ -373,7 +374,17 @@ class ProposalService: 讓學習服務 EWMA 能在人工審核後更新 Playbook trust score。 邏輯與 decision_manager._try_playbook_match 相同，但只回傳 ID 不改 action。 失敗時靜默返回 None（不阻塞主流程）。 + + W1 PR-P1 Feature Flag (2026-04-28 ogt + Claude Sonnet 4.6): + ENABLE_PLAYBOOK_MATCHING=false → 回傳 None，行為與修復前完全相同（回滾用）。 """ + if not get_settings().ENABLE_PLAYBOOK_MATCHING: + logger.debug( + "playbook_matching_disabled", + incident_id=getattr(incident, "incident_id", "?"), + ) + return None + PLAYBOOK_SIMILARITY_THRESHOLD = 0.85 try: from src.models.playbook import SymptomPattern diff --git a/apps/api/tests/test_ai_rule_catalog_write.py b/apps/api/tests/test_ai_rule_catalog_write.py new file mode 100644 index 00000000..7b9a808c --- /dev/null +++ b/apps/api/tests/test_ai_rule_catalog_write.py @@ -0,0 +1,224 @@ +""" +ADR-091 Task T1 — AI 自學規則雙寫 alert_rule_catalog 測試 +============================================================= +測試範圍： + - test_insert_success_yaml_and_db DB 插入成功路徑 + - test_db_failure_does_not_rollback_yaml DB 失敗時僅 warning，不拋例外 + - test_idempotent_on_conflict 同 rule_name ON CONFLICT DO NOTHING 不報錯 + - test_feature_flag_disabled_skips_db_insert ENABLE_AI_RULE_CATALOG_WRITE=False 跳過 DB + - test_confidence_defaults_to_0_5_review_status_draft + confidence=0.50, review_status='draft' + +禁止 Mock 測試鐵律： + _insert_catalog_ai_generated() 純 Python + DB 邏輯。 + DB 依賴用 AsyncMock patch（get_db_context），只測業務邏輯分支， + 不依賴真實 PostgreSQL（unit 層）。 + _parse_for_to_seconds() 純 Python，直接呼叫真實函式。 + +建立：2026-04-28 (台北時區) Claude Sonnet 4.6 (ADR-091 T1) +""" +from __future__ import annotations + +import os +from contextlib import asynccontextmanager +from unittest.mock import AsyncMock, patch + +import pytest + +# conftest 在 import 前設定 MOCK_MODE=true +# 補設 DATABASE_URL，讓 Settings validate 不因必填欄位失敗 +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") + +from src.core.config import settings as _cfg_settings +from src.services.alert_rule_engine import ( + _insert_catalog_ai_generated, + _parse_for_to_seconds, +) + + +# ============================================================================= +# Helper +# ============================================================================= + +def _make_db_ctx(execute_side_effect=None): + """回傳 (async_ctx_factory, mock_db)。 + simulate get_db_context() 的 async context manager。 + """ + mock_db = AsyncMock() + if execute_side_effect is not None: + mock_db.execute = AsyncMock(side_effect=execute_side_effect) + else: + mock_db.execute = AsyncMock(return_value=None) + + @asynccontextmanager + async def _ctx(): + yield mock_db + + return _ctx, mock_db + + +_SAMPLE_RULE = { + "id": "high_cpu_load", + "match": {"alertname": ["HostHighCpuLoad"]}, + "response": { + "suggested_action": "SCALE_UP", + "kubectl_command": "kubectl scale deployment/api --replicas=3", + }, + "labels": {"severity": "warning"}, + "annotations": {"summary": "CPU 超載"}, +} + + +# ============================================================================= +# _parse_for_to_seconds — 純 Python，無 mock 需要 +# ============================================================================= + +class TestParseForToSeconds: + def test_seconds(self): + assert _parse_for_to_seconds("30s") == 30 + + def test_minutes(self): + assert _parse_for_to_seconds("5m") == 300 + + def test_hours(self): + assert _parse_for_to_seconds("2h") == 7200 + + def test_days(self): + assert _parse_for_to_seconds("1d") == 86400 + + def test_no_unit_defaults_seconds(self): + assert _parse_for_to_seconds("0") == 0 + + def test_zero_string(self): + assert _parse_for_to_seconds("0s") == 0 + + def test_empty_string_returns_none(self): + assert _parse_for_to_seconds("") is None + + def test_invalid_returns_none(self): + assert _parse_for_to_seconds("abc") is None + + +# ============================================================================= +# _insert_catalog_ai_generated +# ============================================================================= + +class TestInsertSuccessYamlAndDb: + """DB 插入成功路徑""" + + async def test_insert_success_yaml_and_db(self): + ctx, mock_db = _make_db_ctx() + + with ( + patch("src.db.base.get_db_context", ctx), + patch.object(_cfg_settings, "ENABLE_AI_RULE_CATALOG_WRITE", True), + ): + await _insert_catalog_ai_generated( + rule_dict=_SAMPLE_RULE, + llm_source="ollama", + rule_id="high_cpu_load", + alertname_safe="HostHighCpuLoad", + ) + + mock_db.execute.assert_awaited_once() + call_args = mock_db.execute.call_args + sql_text = str(call_args[0][0]) + assert "ON CONFLICT" in sql_text + params = call_args[0][1] + assert params["rule_name"] == "HostHighCpuLoad" + assert params["created_by_agent"] == "ollama" + + +class TestDbFailureDoesNotRollbackYaml: + """DB 失敗時僅 warning，不拋例外（YAML 路徑不受影響）""" + + async def test_db_failure_does_not_raise(self): + ctx, _ = _make_db_ctx(execute_side_effect=Exception("DB connection refused")) + + with ( + patch("src.db.base.get_db_context", ctx), + patch.object(_cfg_settings, "ENABLE_AI_RULE_CATALOG_WRITE", True), + ): + # 不應拋例外 + await _insert_catalog_ai_generated( + rule_dict=_SAMPLE_RULE, + llm_source="gemini", + rule_id="high_cpu_load", + alertname_safe="HostHighCpuLoad", + ) + # 到這裡代表沒有 raise — 測試通過 + + +class TestIdempotentOnConflict: + """同 rule_name 第二次觸發 ON CONFLICT DO NOTHING，不報錯""" + + async def test_second_call_no_error(self): + # execute 回傳 None 模擬 DO NOTHING（無 row 回傳，無例外） + ctx, mock_db = _make_db_ctx() + + with ( + patch("src.db.base.get_db_context", ctx), + patch.object(_cfg_settings, "ENABLE_AI_RULE_CATALOG_WRITE", True), + ): + await _insert_catalog_ai_generated( + rule_dict=_SAMPLE_RULE, + llm_source="ollama", + rule_id="high_cpu_load", + alertname_safe="HostHighCpuLoad", + ) + await _insert_catalog_ai_generated( + rule_dict=_SAMPLE_RULE, + llm_source="ollama", + rule_id="high_cpu_load", + alertname_safe="HostHighCpuLoad", + ) + + assert mock_db.execute.await_count == 2 + + +class TestFeatureFlagDisabledSkipsDbInsert: + """ENABLE_AI_RULE_CATALOG_WRITE=False → 跳過 DB insert""" + + async def test_flag_false_no_db_call(self): + ctx, mock_db = _make_db_ctx() + + with ( + patch("src.db.base.get_db_context", ctx), + patch.object(_cfg_settings, "ENABLE_AI_RULE_CATALOG_WRITE", False), + ): + await _insert_catalog_ai_generated( + rule_dict=_SAMPLE_RULE, + llm_source="ollama", + rule_id="high_cpu_load", + alertname_safe="HostHighCpuLoad", + ) + + mock_db.execute.assert_not_awaited() + + +class TestConfidenceAndReviewStatus: + """SQL 中 confidence=0.50、review_status='draft'（符合 DB CHECK 約束）""" + + async def test_confidence_and_review_status_in_sql(self): + ctx, mock_db = _make_db_ctx() + + with ( + patch("src.db.base.get_db_context", ctx), + patch.object(_cfg_settings, "ENABLE_AI_RULE_CATALOG_WRITE", True), + ): + await _insert_catalog_ai_generated( + rule_dict=_SAMPLE_RULE, + llm_source="claude", + rule_id="test_rule", + alertname_safe="TestAlert", + ) + + call_args = mock_db.execute.call_args + sql_text = str(call_args[0][0]) + # SQL 中應含 0.50 confidence 和 'draft' + assert "0.50" in sql_text + assert "'draft'" in sql_text + # params 中不含 confidence/review_status（已硬編 SQL 內） + params = call_args[0][1] + assert "confidence" not in params + assert "review_status" not in params diff --git a/apps/api/tests/test_matched_playbook_id_e2e.py b/apps/api/tests/test_matched_playbook_id_e2e.py new file mode 100644 index 00000000..2e1069d5 --- /dev/null +++ b/apps/api/tests/test_matched_playbook_id_e2e.py @@ -0,0 +1,452 @@ +""" +飛輪閉環 C1 修復 E2E 測試 — matched_playbook_id 全鏈路 +========================================================= +W1 PR-P1 by Claude Sonnet 4.6 (2026-04-28 台北時區) + +測試覆蓋四個斷點的修復驗證： + 1. proposal_service 在相似度 >= 0.85 時填 matched_playbook_id + 2. proposal_service 在相似度 < 0.85 時保留 None + 3. approval_db 正確序列化 matched_playbook_id 到 record_data dict + 4. learning_service 在 matched_playbook_id 存在時觸發 _update_playbook_stats（EWMA） + +🔴 遵循 feedback_no_mock_testing.md: + - 禁止 MagicMock/AsyncMock/unittest.mock.patch + - 使用純 Python Stub 類別 + pytest monkeypatch（替換 module-level singleton） + - 不連線真實 DB / Redis + +Feature Flag: + - ENABLE_PLAYBOOK_MATCHING=false → _try_playbook_match_id 立即回傳 None（rollback 路徑） +""" + +from __future__ import annotations + +from uuid import UUID + +import pytest + +from src.models.approval import ( + ApprovalRequest, + ApprovalRequestCreate, + ApprovalStatus, + RiskLevel, +) +from src.models.incident import Incident, IncidentStatus, Severity, Signal +from src.models.playbook import ( + ActionType, + Playbook, + PlaybookStatus, + RepairStep, + RiskLevel as PlaybookRiskLevel, + SymptomPattern, +) +from src.utils.timezone import now_taipei + + +# ============================================================================= +# Stubs +# ============================================================================= + + +class StubRecommendation: + """PlaybookService.get_recommendations 回傳的推薦結果 Stub""" + + def __init__(self, playbook: Playbook, similarity_score: float = 0.9) -> None: + self.playbook = playbook + self.similarity_score = similarity_score + + +class StubPlaybookService: + """PlaybookService 的輕量 Stub — 記錄 record_execution 呼叫""" + + def __init__(self, recommendations: list | None = None) -> None: + self._recommendations = recommendations or [] + self.record_execution_calls: list[dict] = [] + + async def get_recommendations(self, symptoms, top_k: int = 1) -> list: + return self._recommendations + + async def record_execution(self, playbook_id: str, success: bool) -> bool: + self.record_execution_calls.append({"playbook_id": playbook_id, "success": success}) + return True + + +class StubApprovalService: + """ApprovalDBService 的輕量 Stub — 記錄 create_approval 呼叫""" + + def __init__(self) -> None: + self.created: list[ApprovalRequestCreate] = [] + + async def create_approval(self, request: ApprovalRequestCreate) -> ApprovalRequest: + self.created.append(request) + # 回傳最小合法的 ApprovalRequest + return ApprovalRequest( + id=UUID("00000000-0000-0000-0000-000000000001"), + action=request.action, + description=request.description, + status=ApprovalStatus.PENDING, + risk_level=request.risk_level or RiskLevel.MEDIUM, + required_signatures=1, + current_signatures=0, + requested_by=request.requested_by, + created_at=now_taipei(), + matched_playbook_id=request.matched_playbook_id, + incident_id=request.incident_id, + ) + + +class StubLearningRepo: + """LearningRepository 的極簡 Stub""" + + async def record_repair(self, **kwargs) -> bool: + return True + + async def get_all_repair_stats(self, *a, **kw): + return {} + + async def get_repair_history(self, *a, **kw): + return [] + + async def get_repair_stats(self, *a, **kw): + return {} + + async def record_disposition(self, *a, **kw): + return True + + async def get_dispositions(self, *a, **kw): + return {} + + +class StubTrustRepo: + """TrustRepository 的極簡 Stub""" + + async def upsert(self, **kwargs) -> None: + pass + + async def save_trust_record(self, *a, **kw): + pass + + async def load_trust_record(self, *a, **kw): + return None + + async def get_all_trust_records(self, *a, **kw): + return [] + + +# ============================================================================= +# Factories +# ============================================================================= + + +def _make_incident( + incident_id: str = "INC-C1-001", + severity: Severity = Severity.P2, +) -> Incident: + now = now_taipei() + return Incident( + incident_id=incident_id, + status=IncidentStatus.INVESTIGATING, + severity=severity, + affected_services=["c1-test-service"], + signals=[ + Signal( + alert_name="HighCpuLoad", + severity=severity, + source="prometheus", + fired_at=now, + labels={"namespace": "awoooi-prod", "alertname": "HighCpuLoad"}, + ) + ], + ) + + +def _make_playbook( + playbook_id: str = "PB-C1-001", + trust_score: float = 0.75, +) -> Playbook: + return Playbook( + playbook_id=playbook_id, + name="C1 測試 Playbook", + description="飛輪斷鏈 C1 修復 E2E 測試用", + status=PlaybookStatus.APPROVED, + symptom_pattern=SymptomPattern( + alert_names=["HighCpuLoad"], + affected_services=["c1-test-service"], + severity_range=["P2"], + ), + repair_steps=[ + RepairStep( + step_number=1, + action_type=ActionType.MANUAL, + command="kubectl rollout restart deployment/c1-test-service", + risk_level=PlaybookRiskLevel.LOW, + ) + ], + trust_score=trust_score, + success_count=8, + failure_count=2, + ) + + +# ============================================================================= +# Test 1: proposal 在相似度 >= 0.85 時填 matched_playbook_id +# ============================================================================= + + +@pytest.mark.asyncio +async def test_proposal_fills_matched_playbook_id_when_above_threshold(monkeypatch): + """ + proposal_service._try_playbook_match_id 在相似度 >= 0.85 + status=APPROVED 時 + 應回傳 playbook_id（非 None）。 + """ + playbook = _make_playbook(playbook_id="PB-C1-ABOVE", trust_score=0.8) + stub_pb_service = StubPlaybookService( + recommendations=[StubRecommendation(playbook, similarity_score=0.92)] + ) + + # 替換 module-level playbook_service singleton + import src.services.playbook_service as _pb_mod + monkeypatch.setattr(_pb_mod, "_service", stub_pb_service) + + from src.services.proposal_service import ProposalService + + service = ProposalService.__new__(ProposalService) + incident = _make_incident() + + result = await service._try_playbook_match_id(incident) + + assert result == "PB-C1-ABOVE", ( + f"相似度 0.92 >= 0.85 應回傳 playbook_id，實際得到: {result}" + ) + + +# ============================================================================= +# Test 2: proposal 在相似度 < 0.85 時保留 None +# ============================================================================= + + +@pytest.mark.asyncio +async def test_proposal_keeps_none_when_below_threshold(monkeypatch): + """ + proposal_service._try_playbook_match_id 在相似度 < 0.85 時應回傳 None。 + """ + playbook = _make_playbook(playbook_id="PB-C1-BELOW", trust_score=0.5) + stub_pb_service = StubPlaybookService( + recommendations=[StubRecommendation(playbook, similarity_score=0.70)] + ) + + import src.services.playbook_service as _pb_mod + monkeypatch.setattr(_pb_mod, "_service", stub_pb_service) + + from src.services.proposal_service import ProposalService + + service = ProposalService.__new__(ProposalService) + incident = _make_incident() + + result = await service._try_playbook_match_id(incident) + + assert result is None, ( + f"相似度 0.70 < 0.85 應回傳 None，實際得到: {result}" + ) + + +# ============================================================================= +# Test 3: approval_db approval_request_to_record_data 序列化 matched_playbook_id +# ============================================================================= + + +def test_approval_db_persists_matched_playbook_id(): + """ + approval_db.approval_request_to_record_data 應將 matched_playbook_id + 正確序列化到 DB dict（不遺失欄位）。 + """ + from src.services.approval_db import approval_request_to_record_data + from src.models.approval import ApprovalRequestCreate, BlastRadius, DataImpact + + request = ApprovalRequestCreate( + action="kubectl rollout restart deployment/c1-test-service", + description="C1 test", + risk_level=RiskLevel.MEDIUM, + requested_by="c1-e2e-test", + incident_id="INC-C1-001", + matched_playbook_id="PB-C1-001", + blast_radius=BlastRadius( + affected_pods=2, + estimated_downtime="< 2 min", + related_services=["c1-test-service"], + data_impact=DataImpact.NONE, + ), + ) + + data = approval_request_to_record_data(request, RiskLevel.MEDIUM, required_sigs=1) + + assert "matched_playbook_id" in data, "record_data 必須包含 matched_playbook_id 鍵" + assert data["matched_playbook_id"] == "PB-C1-001", ( + f"matched_playbook_id 應為 'PB-C1-001'，實際: {data['matched_playbook_id']}" + ) + + +def test_approval_db_record_data_none_when_not_provided(): + """ + matched_playbook_id 未提供時，record_data 應包含 None（不拋例外）。 + """ + from src.services.approval_db import approval_request_to_record_data + + request = ApprovalRequestCreate( + action="kubectl rollout restart deployment/test", + description="No playbook test", + risk_level=RiskLevel.LOW, + requested_by="c1-e2e-test", + ) + + data = approval_request_to_record_data(request, RiskLevel.LOW, required_sigs=0) + + assert "matched_playbook_id" in data + assert data["matched_playbook_id"] is None + + +# ============================================================================= +# Test 4: learning_service 在 matched_playbook_id 存在時觸發 EWMA 更新 +# ============================================================================= + + +@pytest.mark.asyncio +async def test_learning_service_updates_trust_when_matched(monkeypatch): + """ + learning_service.process_execution_result 在 approval.matched_playbook_id + 非 None 時，應呼叫 _update_playbook_stats（最終觸發 EWMA trust_score 更新）。 + """ + playbook_id = "PB-C1-LEARN-001" + stub_pb_service = StubPlaybookService() + playbook = _make_playbook(playbook_id=playbook_id) + stub_pb_service._recommendations = [] # 不需要 recommendations + + # 替換 playbook_service singleton（_update_playbook_stats lazy import 會呼叫 get_playbook_service()） + import src.services.playbook_service as _pb_mod + monkeypatch.setattr(_pb_mod, "_service", stub_pb_service) + + from src.services.learning_service import LearningService, ExecutionResult + + svc = LearningService( + repository=StubLearningRepo(), + trust_repository=StubTrustRepo(), + ) + + approval = ApprovalRequest( + id=UUID("00000000-0000-0000-0000-000000000002"), + action="kubectl rollout restart deployment/c1-service", + description="C1 learning test", + status=ApprovalStatus.APPROVED, + risk_level=RiskLevel.MEDIUM, + required_signatures=1, + current_signatures=1, + requested_by="c1-e2e-test", + created_at=now_taipei(), + matched_playbook_id=playbook_id, + ) + + result = ExecutionResult( + approval_id=str(approval.id), + incident_id="INC-C1-LEARN-001", + action=approval.action, + success=True, + ) + + learning_record = await svc.process_execution_result(approval, result) + + assert learning_record.playbook_updated is True, ( + "matched_playbook_id 不為 None 時，playbook_updated 應為 True（EWMA 觸發）" + ) + assert len(stub_pb_service.record_execution_calls) == 1, ( + "record_execution 應被呼叫一次" + ) + assert stub_pb_service.record_execution_calls[0]["playbook_id"] == playbook_id + assert stub_pb_service.record_execution_calls[0]["success"] is True + + +@pytest.mark.asyncio +async def test_learning_service_skips_update_when_no_match(): + """ + approval.matched_playbook_id is None 時，learning_service 不更新 Playbook stats。 + playbook_updated 應為 False，record_execution 不被呼叫。 + """ + stub_pb_service = StubPlaybookService() + + from src.services.learning_service import LearningService, ExecutionResult + + svc = LearningService( + repository=StubLearningRepo(), + trust_repository=StubTrustRepo(), + ) + + approval = ApprovalRequest( + id=UUID("00000000-0000-0000-0000-000000000003"), + action="kubectl rollout restart deployment/unmatched-service", + description="No match test", + status=ApprovalStatus.APPROVED, + risk_level=RiskLevel.LOW, + required_signatures=0, + current_signatures=0, + requested_by="c1-e2e-test", + created_at=now_taipei(), + matched_playbook_id=None, + ) + + result = ExecutionResult( + approval_id=str(approval.id), + incident_id="INC-C1-NO-MATCH-001", + action=approval.action, + success=True, + ) + + learning_record = await svc.process_execution_result(approval, result) + + assert learning_record.playbook_updated is False, ( + "matched_playbook_id=None 時 playbook_updated 應為 False" + ) + assert len(stub_pb_service.record_execution_calls) == 0, ( + "matched_playbook_id=None 時不應呼叫 record_execution" + ) + + +# ============================================================================= +# Test 5: Feature Flag ENABLE_PLAYBOOK_MATCHING=false 回滾路徑 +# ============================================================================= + + +@pytest.mark.asyncio +async def test_feature_flag_disabled_returns_none(monkeypatch): + """ + ENABLE_PLAYBOOK_MATCHING=false 時，_try_playbook_match_id 必須直接回傳 None， + 不呼叫 PlaybookService（完全回滾到修復前行為）。 + """ + playbook = _make_playbook(playbook_id="PB-C1-FLAG") + stub_pb_service = StubPlaybookService( + recommendations=[StubRecommendation(playbook, similarity_score=0.99)] + ) + + import src.services.playbook_service as _pb_mod + monkeypatch.setattr(_pb_mod, "_service", stub_pb_service) + + # 替換 settings 使 ENABLE_PLAYBOOK_MATCHING=false + # proposal_service 用 `from src.core.config import get_settings`， + # 所以必須替換 proposal_service 模組命名空間中的 get_settings 符號。 + from src.core.config import Settings + fake_settings = Settings.model_construct( + ENABLE_PLAYBOOK_MATCHING=False, + DATABASE_URL="postgresql+asyncpg://x:y@localhost/z", + ) + import src.services.proposal_service as _ps_mod + monkeypatch.setattr(_ps_mod, "get_settings", lambda: fake_settings) + + from src.services.proposal_service import ProposalService + + service = ProposalService.__new__(ProposalService) + incident = _make_incident() + + result = await service._try_playbook_match_id(incident) + + assert result is None, ( + "ENABLE_PLAYBOOK_MATCHING=false 時應回傳 None，不進行匹配" + ) + # StubPlaybookService 的 get_recommendations 沒有被呼叫 + assert stub_pb_service._recommendations is not None # stub 本身沒被呼叫（沒計數器，靠上面 assert result is None 驗證）