File size: 2,841 Bytes
d2d1903 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """
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)
|