awoooi/apps/api/tests/test_classify_alert_early.py

"""
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_info_only(self):
        # severity=info → severity 規則先命中，TYPE-1
        ac, nt = classify_alert_early("BackupJobComplete", "info", {})
        assert nt == "TYPE-1"

    def test_backup_keyword_warning_not_type1(self):
        # BackupJobFailed severity=warning → 繼續走 prefix 規則，不應是 TYPE-1
        ac, nt = classify_alert_early("BackupJobFailed", "warning", {})
        assert nt == "TYPE-3"

    def test_watchdog_heartbeat(self):
        # Watchdog (Alertmanager 心跳) severity=none → severity 規則先命中，TYPE-1
        ac, nt = classify_alert_early("Watchdog", "none", {})
        assert nt == "TYPE-1"

    def test_backup_critical_not_type1(self):
        # critical backup 告警應走各自 prefix，不是純資訊
        ac, nt = classify_alert_early("BACKUP_MISSING", "critical", {})
        assert nt == "TYPE-3"


# --------------------------------------------------------------------------- #
# 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_failed_is_kubernetes(self):
        # VeleroBackupFailed severity=critical → backup 規則不命中，走 Velero prefix → kubernetes TYPE-3
        ac, nt = classify_alert_early("VeleroBackupFailed", "critical", {})
        assert nt == "TYPE-3"
        assert ac == "kubernetes"

    def test_velero_backup_success_info_is_type1(self):
        # VeleroBackupSuccess severity=info → TYPE-1
        ac, nt = classify_alert_early("VeleroBackupSuccess", "info", {})
        assert nt == "TYPE-1"


# --------------------------------------------------------------------------- #
# 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"