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