tests/test_core_drift.py

"""
Real correctness tests for core.drift — drift_score weighting and the
severity thresholds. These pin the behaviour before Stage 1 makes the
weights/thresholds config-driven.
"""
import math

from core.drift import DriftIssue, drift_score, severity_from_score

# --- drift_score ----------------------------------------------------------

HEALTHY = {"delta": 0.0, "psi": 1.0, "xi": 0.0, "gamma": 0.0, "kappa": 1.0}
DEGRADED = {"delta": 1.0, "psi": 0.0, "xi": 1.0, "gamma": 1.0, "kappa": 0.0}


def test_drift_score_is_zero_for_a_perfectly_healthy_entity():
    assert drift_score(HEALTHY) == 0.0


def test_drift_score_is_one_for_a_fully_degraded_entity():
    assert math.isclose(drift_score(DEGRADED), 1.0, rel_tol=1e-9)


def test_drift_score_uses_documented_weights():
    # 0.25*delta + 0.20*(1-psi) + 0.25*xi + 0.15*gamma + 0.15*(1-kappa)
    only_delta = drift_score({**HEALTHY, "delta": 1.0})
    only_xi = drift_score({**HEALTHY, "xi": 1.0})
    only_gamma = drift_score({**HEALTHY, "gamma": 1.0})
    assert math.isclose(only_delta, 0.25, rel_tol=1e-9)
    assert math.isclose(only_xi, 0.25, rel_tol=1e-9)
    assert math.isclose(only_gamma, 0.15, rel_tol=1e-9)
    # delta and xi are the heaviest contributors, gamma/kappa the lightest
    assert only_delta > only_gamma


def test_drift_score_handles_missing_signals_with_safe_defaults():
    # missing psi/kappa default to 1.0 (healthy), missing delta/xi/gamma to 0.0
    assert drift_score({}) == 0.0


def test_drift_score_rises_monotonically_with_anomaly():
    low = drift_score({**HEALTHY, "xi": 0.1})
    high = drift_score({**HEALTHY, "xi": 0.9})
    assert high > low


# --- severity_from_score --------------------------------------------------

def test_severity_thresholds_at_boundaries():
    assert severity_from_score(0.75) == "critical"
    assert severity_from_score(0.749) == "high"
    assert severity_from_score(0.55) == "high"
    assert severity_from_score(0.549) == "medium"
    assert severity_from_score(0.35) == "medium"
    assert severity_from_score(0.349) == "low"
    assert severity_from_score(0.0) == "low"


def test_severity_is_ordered():
    order = ["low", "medium", "high", "critical"]
    seen = [severity_from_score(s) for s in (0.1, 0.4, 0.6, 0.9)]
    assert seen == order


# --- DriftIssue dataclass -------------------------------------------------

def test_drift_issue_constructs_with_defaults():
    from datetime import date

    issue = DriftIssue(
        issue_id="i1",
        entity_id="warehouse_north",
        entity_type="warehouse",
        detected_at=date(2026, 5, 15),
        score=0.62,
        severity="high",
        title="warehouse_north operational drift detected",
        explanation="backlog rising",
    )
    assert issue.evidence == []
    assert issue.severity == severity_from_score(issue.score)