feat(flywheel): W1 PR-P1 + ADR-091 T1 — 飛輪 80→90 第一波
依 onboarder 端到端閉環審計挖出的 10 條斷鏈 + critic 鐵律違反全景， W1 第一波修復飛輪鐵證 1 + 2 的核心斷鏈 C1。 ## W1 PR-P1 — matched_playbook_id 四斷點守門 (C1 修復) fullstack 探勘發現 4 斷點之前 session 已修，本 PR 補： - ENABLE_PLAYBOOK_MATCHING feature flag (default=true) rollback: kubectl set env deployment/awoooi-api ENABLE_PLAYBOOK_MATCHING=false - proposal_service._try_playbook_match_id 入口加 flag check - 7 個 e2e 測試補上保護網（之前無測試覆蓋） 斷鏈 C1 證據鏈：proposal_service.generate_proposal() → matched_playbook_id → approval_db → approval_repository → learning_service._update_playbook_stats 24h 後 playbooks.trust_score 應有真實 EWMA 更新。 ## ADR-091 T1 — auto_generate_rule 雙寫 DB (鐵證 1 第一步) 飛輪鐵證 1：alert_rule_catalog.source='ai_generated' 全 codebase 0 筆。 auto_generate_rule() 寫 alert_rules.yaml 但不寫 DB → AI 自學成果與 catalog 雙軌脫鉤。 修法（依 ADR-091 §1 D1）： - 新增 _insert_catalog_ai_generated()：YAML 寫入成功後雙寫 source='ai_generated', confidence=0.5, review_status='draft', created_by_agent - 新增 _parse_for_to_seconds() helper（"30s"/"5m"/"2h" → seconds） - ON CONFLICT (rule_name) DO NOTHING 冪等保證 - transaction 策略：YAML + DB 不在同一 transaction（YAML 已成 SoT，DB 失敗只 log） - ENABLE_AI_RULE_CATALOG_WRITE feature flag (default=true) rollback: kubectl set env deployment/awoooi-api ENABLE_AI_RULE_CATALOG_WRITE=false 13 個測試覆蓋：parse helper 8 + 業務邏輯 5（success/db_fail/idempotent/flag/SQL_lit） ## 驗證 1572 unit tests 全綠（+20 新增：PR-P1 7 + ADR-091 T1 13） ## 期望影響 飛輪自主化評分：42 → 65（+23 = C1 +3 + 鐵證 1 +20） ## 已知債（critic PR review 揭示，下一個 commit 處理） - KMWriter 統一契約 3 條 caller 路徑被旁路（C1/M1/M2） - KMWriter 冪等聲明與實作不符（M3 缺 ON CONFLICT） - Alertmanager equal:[] 爆炸抑制 + 版本未驗（M4/M5） - drift checker regex 脆弱（M7 應改 AST） - governance health score skipped 失真（M6） Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Your Name
@@ -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（純邏輯分類，不打外部服務）。
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
224
apps/api/tests/test_ai_rule_catalog_write.py
Normal file
224
apps/api/tests/test_ai_rule_catalog_write.py
Normal file
@@ -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
|
||||
452
apps/api/tests/test_matched_playbook_id_e2e.py
Normal file
452
apps/api/tests/test_matched_playbook_id_e2e.py
Normal file
@@ -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 驗證）
|
||||
Reference in New Issue
Block a user