fb130c9a28
Some checks failed
CD Pipeline / build-and-deploy (push) Failing after 2m16s
Wave 8 P3.1-T2 後續補測 + 配套： 新增測試: - test_diagnosis_aggregator_stub.py (238 行) — 15 tests · stub fixture 驗證 _collect_diagnosis_aggregator 接線 · feature flag default off 不呼叫 · timeout 邊界 / exception fail-soft 修改: - core/metrics.py +23 — 新增 DiagnosisAggregator 相關 Prometheus 指標 - sanitization_service.py +24 — 補強 prompt sanitize 邊界（vuln #4 配套） - RUNBOOK-AGENT-STEP-LATENCY.md / agent_step_latency_rules.yaml — 微調 Tests: 15 passed Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
239 lines
11 KiB
Python
239 lines
11 KiB
Python
"""
|
||
DiagnosisAggregator Conservative 整合測試
|
||
==========================================
|
||
P3.1-T2 by Claude 2026-04-27 — Tier-2 三服務感知強化
|
||
|
||
Conservative 策略：ENABLE_DIAGNOSIS_AGGREGATOR=False（預設關閉）
|
||
驗證：
|
||
1. ENABLE_DIAGNOSIS_AGGREGATOR=False 時 aggregator 不被呼叫
|
||
2. ENABLE_DIAGNOSIS_AGGREGATOR=True 時 _collect_diagnosis_aggregator 被呼叫
|
||
3. aggregator 呼叫失敗時不影響主路徑（exception 隔離）
|
||
4. EvidenceSnapshot.extra_diagnosis 欄位存在
|
||
5. build_summary() 包含 extra_diagnosis 時正確輸出
|
||
6. DiagnosisAggregator.collect_pod_diagnosis 介面正確
|
||
|
||
注意：不依賴真實 K8s/SignOz — 全 mock 測試
|
||
|
||
重疊分析報告（見 P7-COMPLETION 章節）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Test: EvidenceSnapshot.extra_diagnosis 欄位
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestEvidenceSnapshotExtraDiagnosis:
|
||
def test_extra_diagnosis_field_exists(self):
|
||
"""EvidenceSnapshot 應有 extra_diagnosis 欄位，預設 None"""
|
||
from src.services.evidence_snapshot import EvidenceSnapshot
|
||
snap = EvidenceSnapshot(incident_id="INC-001")
|
||
assert hasattr(snap, "extra_diagnosis")
|
||
assert snap.extra_diagnosis is None
|
||
|
||
def test_build_summary_includes_extra_diagnosis(self):
|
||
"""extra_diagnosis 不為 None 時 build_summary 應包含 Pod深診斷"""
|
||
from src.services.evidence_snapshot import EvidenceSnapshot
|
||
snap = EvidenceSnapshot(incident_id="INC-001")
|
||
snap.extra_diagnosis = "## 診斷目標\n- Target: api-pod\n- Namespace: awoooi-prod"
|
||
summary = snap.build_summary()
|
||
assert "Pod深診斷" in summary
|
||
assert "api-pod" in summary
|
||
|
||
def test_build_summary_no_extra_diagnosis_no_section(self):
|
||
"""extra_diagnosis=None 時 build_summary 不應包含 Pod深診斷"""
|
||
from src.services.evidence_snapshot import EvidenceSnapshot
|
||
snap = EvidenceSnapshot(incident_id="INC-001")
|
||
snap.extra_diagnosis = None
|
||
summary = snap.build_summary()
|
||
assert "Pod深診斷" not in summary
|
||
|
||
def test_extra_diagnosis_not_persisted_to_db_record(self):
|
||
"""extra_diagnosis 是 in-memory only，save() 不應包含此欄位到 DB model"""
|
||
from src.services.evidence_snapshot import EvidenceSnapshot
|
||
from src.db.models import IncidentEvidence
|
||
# 確認 DB model 沒有 extra_diagnosis 欄位（in-memory only 設計）
|
||
assert not hasattr(IncidentEvidence, "extra_diagnosis"), \
|
||
"extra_diagnosis 應為 in-memory only，不應存在於 DB model"
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Test: ENABLE_DIAGNOSIS_AGGREGATOR feature flag
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestDiagnosisAggregatorFeatureFlag:
|
||
def test_feature_flag_exists_in_settings(self):
|
||
"""config.py 應有 ENABLE_DIAGNOSIS_AGGREGATOR 欄位，預設 False"""
|
||
from src.core.config import settings
|
||
assert hasattr(settings, "ENABLE_DIAGNOSIS_AGGREGATOR")
|
||
# 預設應為 False（conservative）
|
||
assert settings.ENABLE_DIAGNOSIS_AGGREGATOR is False
|
||
|
||
def test_feature_flag_default_false(self):
|
||
"""直接從 Settings class 確認預設值"""
|
||
from src.core.config import Settings
|
||
import inspect
|
||
source = inspect.getsource(Settings)
|
||
# 確認 ENABLE_DIAGNOSIS_AGGREGATOR 存在且預設 False
|
||
assert "ENABLE_DIAGNOSIS_AGGREGATOR" in source
|
||
assert "default=False" in source or "False" in source
|
||
|
||
def test_aggregator_guarded_by_flag_in_investigate(self):
|
||
"""investigate() 4.6 區塊有 ENABLE_DIAGNOSIS_AGGREGATOR flag 守門（source inspection）"""
|
||
import inspect
|
||
from src.services.pre_decision_investigator import PreDecisionInvestigator
|
||
source = inspect.getsource(PreDecisionInvestigator.investigate)
|
||
assert "ENABLE_DIAGNOSIS_AGGREGATOR" in source, \
|
||
"investigate() 應有 ENABLE_DIAGNOSIS_AGGREGATOR feature flag 守門"
|
||
assert "_collect_diagnosis_aggregator" in source, \
|
||
"investigate() 應呼叫 _collect_diagnosis_aggregator"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_collect_diagnosis_aggregator_skips_no_pod(self):
|
||
"""沒有 pod label 時 _collect_diagnosis_aggregator 應 early return"""
|
||
from src.services.pre_decision_investigator import PreDecisionInvestigator
|
||
from src.services.evidence_snapshot import EvidenceSnapshot
|
||
|
||
investigator = PreDecisionInvestigator()
|
||
|
||
class _Signal:
|
||
labels = {"alertname": "HostDown", "namespace": "awoooi-prod", "severity": "critical"}
|
||
alert_name = "HostDown"
|
||
annotations = {}
|
||
source = "prometheus"
|
||
|
||
class _Inc:
|
||
incident_id = "INC-TEST-002"
|
||
signals = [_Signal()]
|
||
|
||
snap = EvidenceSnapshot(incident_id="INC-TEST-002")
|
||
mock_aggregator = AsyncMock()
|
||
mock_aggregator.collect_pod_diagnosis = AsyncMock()
|
||
|
||
with patch("src.services.diagnosis_aggregator.get_diagnosis_aggregator", return_value=mock_aggregator):
|
||
await investigator._collect_diagnosis_aggregator(snap, _Inc())
|
||
|
||
# 無 pod_name → early return，aggregator 不被呼叫
|
||
mock_aggregator.collect_pod_diagnosis.assert_not_called()
|
||
assert snap.extra_diagnosis is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_collect_diagnosis_aggregator_fills_extra_diagnosis(self):
|
||
"""有 pod label + aggregator 成功時 extra_diagnosis 應被填入"""
|
||
from src.services.pre_decision_investigator import PreDecisionInvestigator
|
||
from src.services.evidence_snapshot import EvidenceSnapshot
|
||
|
||
investigator = PreDecisionInvestigator()
|
||
|
||
class _Signal:
|
||
labels = {"alertname": "KubePodCrashLooping", "namespace": "awoooi-prod", "pod": "api-xyz-abc", "severity": "critical"}
|
||
alert_name = "KubePodCrashLooping"
|
||
annotations = {}
|
||
source = "prometheus"
|
||
|
||
class _Inc:
|
||
incident_id = "INC-TEST-003"
|
||
signals = [_Signal()]
|
||
|
||
snap = EvidenceSnapshot(incident_id="INC-TEST-003")
|
||
|
||
# mock DiagnosisContext
|
||
mock_ctx = MagicMock()
|
||
mock_ctx.signals = []
|
||
mock_ctx.highest_severity = MagicMock()
|
||
mock_ctx.highest_severity.value = "info"
|
||
mock_ctx.get_llm_prompt_context = MagicMock(return_value="## 診斷目標\n- Target: api-xyz-abc")
|
||
|
||
mock_aggregator = MagicMock()
|
||
mock_aggregator.collect_pod_diagnosis = AsyncMock(return_value=mock_ctx)
|
||
|
||
with patch("src.services.diagnosis_aggregator.get_diagnosis_aggregator", return_value=mock_aggregator):
|
||
await investigator._collect_diagnosis_aggregator(snap, _Inc())
|
||
|
||
assert snap.extra_diagnosis is not None
|
||
assert "api-xyz-abc" in snap.extra_diagnosis
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_collect_diagnosis_aggregator_exception_isolated(self):
|
||
"""aggregator 拋出 exception 時不影響主路徑，snap.extra_diagnosis 維持 None"""
|
||
from src.services.pre_decision_investigator import PreDecisionInvestigator
|
||
from src.services.evidence_snapshot import EvidenceSnapshot
|
||
|
||
investigator = PreDecisionInvestigator()
|
||
|
||
class _Signal:
|
||
labels = {"alertname": "KubePodCrashLooping", "namespace": "awoooi-prod", "pod": "api-fail", "severity": "critical"}
|
||
alert_name = "KubePodCrashLooping"
|
||
annotations = {}
|
||
source = "prometheus"
|
||
|
||
class _Inc:
|
||
incident_id = "INC-TEST-004"
|
||
signals = [_Signal()]
|
||
|
||
snap = EvidenceSnapshot(incident_id="INC-TEST-004")
|
||
|
||
mock_aggregator = MagicMock()
|
||
mock_aggregator.collect_pod_diagnosis = AsyncMock(
|
||
side_effect=Exception("K8s API timeout")
|
||
)
|
||
|
||
# _collect_diagnosis_aggregator 本身不 catch，外層 investigate() 有 try/except
|
||
# 測試：exception 傳出 (由外層 catch)
|
||
with patch("src.services.diagnosis_aggregator.get_diagnosis_aggregator", return_value=mock_aggregator):
|
||
try:
|
||
await investigator._collect_diagnosis_aggregator(snap, _Inc())
|
||
except Exception:
|
||
pass # 外層 investigate() 會 catch
|
||
|
||
# snap 不受影響
|
||
assert snap.extra_diagnosis is None
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Test: DiagnosisAggregator 基本結構
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestDiagnosisAggregatorInterface:
|
||
def test_get_diagnosis_aggregator_singleton(self):
|
||
"""get_diagnosis_aggregator() singleton 可正常取得"""
|
||
import src.services.diagnosis_aggregator as m
|
||
original = m._aggregator
|
||
m._aggregator = None
|
||
try:
|
||
a = m.get_diagnosis_aggregator()
|
||
b = m.get_diagnosis_aggregator()
|
||
assert a is b
|
||
finally:
|
||
m._aggregator = original
|
||
|
||
def test_collect_pod_diagnosis_method_exists(self):
|
||
"""DiagnosisAggregator 必須有 collect_pod_diagnosis 方法"""
|
||
from src.services.diagnosis_aggregator import DiagnosisAggregator
|
||
agg = DiagnosisAggregator.__new__(DiagnosisAggregator)
|
||
assert hasattr(agg, "collect_pod_diagnosis")
|
||
assert callable(agg.collect_pod_diagnosis)
|
||
|
||
def test_collect_service_diagnosis_method_exists(self):
|
||
"""DiagnosisAggregator 必須有 collect_service_diagnosis 方法"""
|
||
from src.services.diagnosis_aggregator import DiagnosisAggregator
|
||
assert hasattr(DiagnosisAggregator, "collect_service_diagnosis")
|
||
|
||
def test_diagnosis_context_get_llm_prompt_context(self):
|
||
"""DiagnosisContext.get_llm_prompt_context() 回傳非空字串"""
|
||
from src.services.diagnosis_aggregator import DiagnosisContext
|
||
ctx = DiagnosisContext(target="test-pod", namespace="awoooi-prod")
|
||
result = ctx.get_llm_prompt_context()
|
||
assert isinstance(result, str)
|
||
assert "test-pod" in result
|
||
|
||
def test_pre_decision_investigator_has_collect_diagnosis_method(self):
|
||
"""PreDecisionInvestigator 必須有 _collect_diagnosis_aggregator 方法"""
|
||
from src.services.pre_decision_investigator import PreDecisionInvestigator
|
||
assert hasattr(PreDecisionInvestigator, "_collect_diagnosis_aggregator")
|
||
assert callable(PreDecisionInvestigator._collect_diagnosis_aggregator)
|