7b9b0c490b
## Phase 19 Omni-Terminal (Wave 0-6 全部完成) ### 核心功能 - SSE 狀態機 (7-State 設計，10/10 分) - GenUI 動態渲染 (6 張卡片 + Zod Schema 驗證) - 核鑰 UX (長按授權 + 風險分級) - Terminal Telemetry (Sentry 整合) ### P0-P2 修復 - P0: Singleton → FastAPI Depends 依賴注入 - P1: Zod Schema 升級 (7 個驗證 Schema) - P1: 錯誤分類碼聚合 (Sentry fingerprint) - P2: Slow Query 監控 (5s 警告 / 10s 嚴重) ### 測試 - test_terminal_service.py: 54 項測試全通過 - 意圖分類: 42 個測試案例 (9 種 IntentType) ### 文檔 - ADR-031: SSE 架構實作紀錄 - ADR-032: GenUI 渲染實作紀錄 - Skills: v1.9 (後端 Terminal 章節) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
286 lines
10 KiB
Python
286 lines
10 KiB
Python
"""
|
|
Auto Repair Service Tests - #8 自動升級決策
|
|
==========================================
|
|
測試自動修復服務層功能
|
|
|
|
版本: v1.0
|
|
建立: 2026-03-26 (台北時區)
|
|
建立者: Claude Code (#8 自動升級決策)
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from src.models.incident import Incident, IncidentStatus, Severity, Signal
|
|
from src.models.playbook import (
|
|
ActionType,
|
|
Playbook,
|
|
PlaybookStatus,
|
|
RepairStep,
|
|
RiskLevel,
|
|
SymptomPattern,
|
|
)
|
|
from src.services.auto_repair_service import AutoRepairService
|
|
from src.utils.timezone import now_taipei
|
|
|
|
|
|
class MockPlaybookService:
|
|
"""Mock playbook service for testing"""
|
|
|
|
def __init__(self):
|
|
self._playbooks: dict[str, Playbook] = {}
|
|
self._recommendations: list = []
|
|
|
|
def add_playbook(self, playbook: Playbook):
|
|
self._playbooks[playbook.playbook_id] = playbook
|
|
|
|
def set_recommendations(self, recommendations: list):
|
|
self._recommendations = recommendations
|
|
|
|
async def get_recommendations(self, symptoms, top_k=3):
|
|
return self._recommendations
|
|
|
|
async def get_by_id(self, playbook_id: str):
|
|
return self._playbooks.get(playbook_id)
|
|
|
|
async def record_execution(self, playbook_id: str, success: bool):
|
|
playbook = self._playbooks.get(playbook_id)
|
|
if playbook:
|
|
if success:
|
|
playbook.success_count += 1
|
|
else:
|
|
playbook.failure_count += 1
|
|
return playbook is not None
|
|
|
|
|
|
def create_test_incident(
|
|
incident_id: str = "INC-TEST-001",
|
|
severity: Severity = Severity.P2,
|
|
) -> Incident:
|
|
"""Create a test incident"""
|
|
now = now_taipei()
|
|
return Incident(
|
|
incident_id=incident_id,
|
|
status=IncidentStatus.INVESTIGATING,
|
|
severity=severity,
|
|
affected_services=["test-service"],
|
|
signals=[
|
|
Signal(
|
|
alert_name="HighCPU",
|
|
severity=severity,
|
|
source="prometheus",
|
|
fired_at=now,
|
|
labels={"namespace": "prod"},
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
def create_high_quality_playbook(
|
|
playbook_id: str = "PB-TEST-001",
|
|
risk_level: RiskLevel = RiskLevel.MEDIUM,
|
|
) -> Playbook:
|
|
"""Create a high quality playbook (success_rate >= 95%, count >= 10)"""
|
|
return Playbook(
|
|
playbook_id=playbook_id,
|
|
name="HighCPU - test-service 修復劇本",
|
|
description="High quality playbook for auto repair",
|
|
status=PlaybookStatus.APPROVED,
|
|
symptom_pattern=SymptomPattern(
|
|
alert_names=["HighCPU"],
|
|
affected_services=["test-service"],
|
|
severity_range=["P2"],
|
|
),
|
|
repair_steps=[
|
|
RepairStep(
|
|
step_number=1,
|
|
action_type=ActionType.KUBECTL,
|
|
command="kubectl rollout restart deployment/{target}",
|
|
risk_level=risk_level,
|
|
),
|
|
],
|
|
success_count=20, # >= 10
|
|
failure_count=1, # success_rate = 95.2%
|
|
ai_confidence=0.9,
|
|
)
|
|
|
|
|
|
class MockPlaybookRecommendation:
|
|
"""Mock recommendation for testing"""
|
|
|
|
def __init__(self, playbook: Playbook, similarity_score: float):
|
|
self.playbook = playbook
|
|
self.similarity_score = similarity_score
|
|
|
|
|
|
class TestAutoRepairService:
|
|
"""Auto Repair Service unit tests"""
|
|
|
|
@pytest.fixture
|
|
def mock_playbook_service(self):
|
|
return MockPlaybookService()
|
|
|
|
@pytest.fixture
|
|
def service(self, mock_playbook_service):
|
|
return AutoRepairService(playbook_service=mock_playbook_service)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_blocks_p1_severity(self, service):
|
|
"""Test that P1 severity incidents are blocked"""
|
|
incident = create_test_incident(severity=Severity.P1)
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is False
|
|
assert decision.blocked_by == "HIGH_SEVERITY"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_blocks_p0_severity(self, service):
|
|
"""Test that P0 severity incidents are blocked"""
|
|
incident = create_test_incident(severity=Severity.P0)
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is False
|
|
assert decision.blocked_by == "HIGH_SEVERITY"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_no_playbook_match(self, service, mock_playbook_service):
|
|
"""Test when no playbook matches"""
|
|
mock_playbook_service.set_recommendations([])
|
|
incident = create_test_incident(severity=Severity.P2)
|
|
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is False
|
|
assert decision.blocked_by == "NO_MATCH"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_low_similarity(self, service, mock_playbook_service):
|
|
"""Test when similarity is too low"""
|
|
playbook = create_high_quality_playbook()
|
|
mock_playbook_service.add_playbook(playbook)
|
|
mock_playbook_service.set_recommendations([
|
|
MockPlaybookRecommendation(playbook, similarity_score=0.5) # Below 0.7
|
|
])
|
|
|
|
incident = create_test_incident(severity=Severity.P2)
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is False
|
|
assert decision.blocked_by == "LOW_SIMILARITY"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_not_high_quality(self, service, mock_playbook_service):
|
|
"""Test when playbook is not high quality"""
|
|
playbook = Playbook(
|
|
playbook_id="PB-LOW-QUALITY",
|
|
name="Low quality playbook",
|
|
description="Not enough executions",
|
|
status=PlaybookStatus.APPROVED,
|
|
symptom_pattern=SymptomPattern(
|
|
alert_names=["HighCPU"],
|
|
affected_services=["test-service"],
|
|
),
|
|
repair_steps=[],
|
|
success_count=5, # < 10
|
|
failure_count=0,
|
|
)
|
|
mock_playbook_service.add_playbook(playbook)
|
|
mock_playbook_service.set_recommendations([
|
|
MockPlaybookRecommendation(playbook, similarity_score=0.9)
|
|
])
|
|
|
|
incident = create_test_incident(severity=Severity.P2)
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is False
|
|
assert decision.blocked_by == "NOT_HIGH_QUALITY"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_high_risk_blocked(self, service, mock_playbook_service):
|
|
"""Test when playbook contains HIGH risk actions"""
|
|
playbook = create_high_quality_playbook(risk_level=RiskLevel.HIGH)
|
|
mock_playbook_service.add_playbook(playbook)
|
|
mock_playbook_service.set_recommendations([
|
|
MockPlaybookRecommendation(playbook, similarity_score=0.9)
|
|
])
|
|
|
|
incident = create_test_incident(severity=Severity.P2)
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is False
|
|
assert decision.blocked_by == "HIGH_RISK"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_critical_risk_blocked(self, service, mock_playbook_service):
|
|
"""Test when playbook contains CRITICAL risk actions"""
|
|
playbook = create_high_quality_playbook(risk_level=RiskLevel.CRITICAL)
|
|
mock_playbook_service.add_playbook(playbook)
|
|
mock_playbook_service.set_recommendations([
|
|
MockPlaybookRecommendation(playbook, similarity_score=0.9)
|
|
])
|
|
|
|
incident = create_test_incident(severity=Severity.P2)
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is False
|
|
assert decision.blocked_by == "HIGH_RISK"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_success(self, service, mock_playbook_service):
|
|
"""Test successful auto repair evaluation"""
|
|
playbook = create_high_quality_playbook(risk_level=RiskLevel.MEDIUM)
|
|
mock_playbook_service.add_playbook(playbook)
|
|
mock_playbook_service.set_recommendations([
|
|
MockPlaybookRecommendation(playbook, similarity_score=0.85)
|
|
])
|
|
|
|
incident = create_test_incident(severity=Severity.P2)
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is True
|
|
assert decision.playbook is not None
|
|
assert decision.playbook.playbook_id == playbook.playbook_id
|
|
assert decision.blocked_by is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_low_risk_allowed(self, service, mock_playbook_service):
|
|
"""Test that LOW risk actions are allowed"""
|
|
playbook = create_high_quality_playbook(risk_level=RiskLevel.LOW)
|
|
mock_playbook_service.add_playbook(playbook)
|
|
mock_playbook_service.set_recommendations([
|
|
MockPlaybookRecommendation(playbook, similarity_score=0.85)
|
|
])
|
|
|
|
incident = create_test_incident(severity=Severity.P2)
|
|
decision = await service.evaluate_auto_repair(incident)
|
|
|
|
assert decision.can_auto_repair is True
|
|
assert decision.risk_level == RiskLevel.LOW
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_high_quality_calculation(self):
|
|
"""Test is_high_quality property"""
|
|
# High quality: APPROVED + 95%+ success rate + 10+ successes
|
|
playbook = create_high_quality_playbook()
|
|
assert playbook.is_high_quality is True
|
|
assert playbook.success_rate >= 0.95
|
|
assert playbook.success_count >= 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_not_high_quality_low_success_rate(self):
|
|
"""Test playbook with low success rate is not high quality"""
|
|
playbook = Playbook(
|
|
playbook_id="PB-LOW-RATE",
|
|
name="Low success rate",
|
|
description="Too many failures",
|
|
status=PlaybookStatus.APPROVED,
|
|
symptom_pattern=SymptomPattern(
|
|
alert_names=["Test"],
|
|
affected_services=["test"],
|
|
),
|
|
repair_steps=[],
|
|
success_count=15,
|
|
failure_count=5, # 75% success rate
|
|
)
|
|
assert playbook.is_high_quality is False
|
|
assert playbook.success_rate < 0.95
|