fix(cr): Code Review P2 — 測試覆蓋 + CronJob 腳本重構
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled
P2-1: CronJob inline Python 抽成 scripts/cron_km_vectorize.py
Dockerfile 加入 COPY scripts/,CronJob YAML 改用腳本路徑
P2-2: 新增 test_classify_alert_early.py — 23 tests 覆蓋 7 條分類規則
含邊界情況:VeleroBackupFailed(backup優先於k8s)、優先順序驗證
595 unit tests passed
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -60,6 +60,8 @@ COPY k8s/ ./k8s/
|
||||
# 2026-04-10 Claude Sonnet 4.6: RAG 知識庫索引來源 (ADR-067 Phase 33)
|
||||
COPY docs/ ./docs/
|
||||
COPY .agents/skills/ ./.agents/skills/
|
||||
# 2026-04-12 ogt (ADR-073 P2-1): CronJob 腳本 — 獨立腳本取代 inline Python
|
||||
COPY scripts/ ./scripts/
|
||||
|
||||
# Install openssh-client + curl — SSH_COMMAND Playbook + healthcheck
|
||||
# Install kubectl — drift_detector 需要 kubectl 讀取 K8s 實際狀態
|
||||
|
||||
155
apps/api/tests/test_classify_alert_early.py
Normal file
155
apps/api/tests/test_classify_alert_early.py
Normal file
@@ -0,0 +1,155 @@
|
||||
"""
|
||||
classify_alert_early() 分類函數單元測試 — ADR-073 Phase 2-2
|
||||
|
||||
覆蓋 7 條分類規則的正常路徑與邊界情況:
|
||||
config_drift, info, backup, infrastructure, kubernetes, database, general
|
||||
|
||||
2026-04-12 ogt (ADR-073 P2-2 測試補強)
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from src.services.incident_service import classify_alert_early
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# TYPE-4D: Config Drift
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
class TestConfigDrift:
|
||||
def test_configuration_drift(self):
|
||||
ac, nt = classify_alert_early("ConfigurationDrift", "critical", {})
|
||||
assert nt == "TYPE-4D"
|
||||
assert ac == "config_drift"
|
||||
|
||||
def test_kube_config_drift(self):
|
||||
ac, nt = classify_alert_early("KubeConfigDrift", "warning", {})
|
||||
assert nt == "TYPE-4D"
|
||||
assert ac == "config_drift"
|
||||
|
||||
def test_config_drift_case_sensitive(self):
|
||||
# 不在白名單裡的變體 → 不應命中 config_drift
|
||||
ac, nt = classify_alert_early("configurationdrift", "critical", {})
|
||||
assert ac != "config_drift"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# TYPE-1: Info / Heartbeat
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
class TestInfoAlerts:
|
||||
def test_severity_info(self):
|
||||
ac, nt = classify_alert_early("SomeAlert", "info", {})
|
||||
assert nt == "TYPE-1"
|
||||
assert ac == "info"
|
||||
|
||||
def test_severity_none(self):
|
||||
ac, nt = classify_alert_early("SomeAlert", "none", {})
|
||||
assert nt == "TYPE-1"
|
||||
assert ac == "info"
|
||||
|
||||
def test_backup_keyword(self):
|
||||
ac, nt = classify_alert_early("BackupJobFailed", "warning", {})
|
||||
assert nt == "TYPE-1"
|
||||
assert ac == "backup"
|
||||
|
||||
def test_heartbeat_keyword(self):
|
||||
ac, nt = classify_alert_early("WatchdogHeartbeat", "warning", {})
|
||||
assert nt == "TYPE-1"
|
||||
assert ac == "backup"
|
||||
|
||||
def test_backup_case_insensitive(self):
|
||||
ac, nt = classify_alert_early("BACKUP_MISSING", "critical", {})
|
||||
assert nt == "TYPE-1"
|
||||
assert ac == "backup"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# TYPE-3: Infrastructure (Docker / Host)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
class TestInfrastructure:
|
||||
def test_docker_prefix(self):
|
||||
ac, nt = classify_alert_early("DockerContainerOOM", "critical", {})
|
||||
assert nt == "TYPE-3"
|
||||
assert ac == "infrastructure"
|
||||
|
||||
def test_host_prefix(self):
|
||||
ac, nt = classify_alert_early("HostHighCpuLoad", "warning", {})
|
||||
assert nt == "TYPE-3"
|
||||
assert ac == "infrastructure"
|
||||
|
||||
def test_docker_restart(self):
|
||||
ac, nt = classify_alert_early("DockerContainerRestarting", "warning", {})
|
||||
assert ac == "infrastructure"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# TYPE-3: Kubernetes
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
class TestKubernetes:
|
||||
@pytest.mark.parametrize("alertname", [
|
||||
"KubePodCrashLooping",
|
||||
"PodHighMemory",
|
||||
"DeploymentReplicasMismatch",
|
||||
"NodeNotReady",
|
||||
"ArgoCDSyncFailed",
|
||||
])
|
||||
def test_k8s_prefixes(self, alertname):
|
||||
ac, nt = classify_alert_early(alertname, "critical", {})
|
||||
assert nt == "TYPE-3"
|
||||
assert ac == "kubernetes"
|
||||
|
||||
def test_velero_backup_keyword_wins(self):
|
||||
# VeleroBackupFailed 含 "backup" → backup 規則優先於 kubernetes prefix
|
||||
ac, nt = classify_alert_early("VeleroBackupFailed", "critical", {})
|
||||
assert nt == "TYPE-1"
|
||||
assert ac == "backup"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# TYPE-3: Database
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
class TestDatabase:
|
||||
def test_postgres(self):
|
||||
ac, nt = classify_alert_early("PostgresDown", "critical", {})
|
||||
assert nt == "TYPE-3"
|
||||
assert ac == "database"
|
||||
|
||||
def test_redis(self):
|
||||
ac, nt = classify_alert_early("RedisMemoryHigh", "warning", {})
|
||||
assert nt == "TYPE-3"
|
||||
assert ac == "database"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# TYPE-3: General (fallback)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
class TestGeneral:
|
||||
def test_unknown_alert(self):
|
||||
ac, nt = classify_alert_early("SomeUnknownAlert", "warning", {})
|
||||
assert nt == "TYPE-3"
|
||||
assert ac == "general"
|
||||
|
||||
def test_empty_alertname(self):
|
||||
ac, nt = classify_alert_early("", "warning", {})
|
||||
assert nt == "TYPE-3"
|
||||
assert ac == "general"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 優先順序驗證 — config_drift 和 info 應優先於 prefix 規則
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
class TestPriority:
|
||||
def test_config_drift_beats_severity(self):
|
||||
# ConfigurationDrift 即使 severity=warning 也應命中 config_drift
|
||||
ac, nt = classify_alert_early("ConfigurationDrift", "warning", {})
|
||||
assert ac == "config_drift"
|
||||
|
||||
def test_info_severity_beats_docker_prefix(self):
|
||||
# Docker 前綴 + severity=info → info 規則先命中
|
||||
ac, nt = classify_alert_early("DockerContainerOOM", "info", {})
|
||||
assert ac == "info"
|
||||
Reference in New Issue
Block a user