Spaces:
Running on Zero
Running on Zero
| """Phase 7 tests — risk fusion engine, safety intelligence, boundary cases. | |
| Unit tests cover the fusion math, threshold boundaries, and missing/invalid | |
| input handling without touching any model artifact. Integration tests load | |
| the real Phase 4 near-miss pipeline and Phase 6 calibrated detector (marked | |
| `slow`) and are skipped when the artifacts are absent. | |
| """ | |
| import json | |
| import math | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| import pytest | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) | |
| import config | |
| from risk_engine import ( | |
| RiskInputError, | |
| assess_risk, | |
| compute_risk_level, | |
| recency_factor, | |
| ) | |
| from safety_intelligence import SafetyIntelligence, to_json | |
| PROJECT_ROOT = Path(__file__).resolve().parents[1] | |
| NEAR_MISS_ARTIFACT = PROJECT_ROOT / "models" / "near_miss" / "random_forest_A_calibrated.joblib" | |
| ACCIDENT_CHECKPOINT = PROJECT_ROOT / "models" / "accident" / "calibrated_accident_detector.pth" | |
| VALIDATION_SPLIT = PROJECT_ROOT / "data" / "processed" / "near_miss" / "splits" / "validation.parquet" | |
| def near_miss_row() -> dict: | |
| if not NEAR_MISS_ARTIFACT.exists() or not VALIDATION_SPLIT.exists(): | |
| pytest.skip("Phase 4 artifacts not built yet.") | |
| return pd.read_parquet(VALIDATION_SPLIT).iloc[0].to_dict() | |
| def safety() -> SafetyIntelligence: | |
| if not NEAR_MISS_ARTIFACT.exists() or not ACCIDENT_CHECKPOINT.exists(): | |
| pytest.skip("Phase 4/6 artifacts not built yet.") | |
| return SafetyIntelligence() | |
| # --------------------------------------------------------------------------- | |
| # Fusion math | |
| # --------------------------------------------------------------------------- | |
| def test_full_assessment_contributions_sum_to_score(): | |
| out = assess_risk( | |
| near_miss_probability=0.8, | |
| accident_probability=0.6, | |
| severity=0.5, | |
| event_age_hours=24.0, | |
| ) | |
| assert abs(sum(out["contributions"].values()) - out["risk_score"]) < 1e-3 | |
| # Weights are renormalized over the 4 present inputs: w_i / total. | |
| assert abs(sum(out["weights_applied"].values()) - 1.0) < 1e-6 | |
| expected = 100.0 * (0.35 * 0.8 + 0.40 * 0.6 + 0.15 * 0.5 + 0.10 * 0.5) | |
| assert out["risk_score"] == pytest.approx(expected, abs=1e-3) | |
| def test_single_input_score_is_scaled_value(): | |
| assert assess_risk(accident_probability=1.0)["risk_score"] == pytest.approx(100.0) | |
| assert assess_risk(accident_probability=0.3)["risk_score"] == pytest.approx(30.0) | |
| assert assess_risk(severity=0.7)["risk_score"] == pytest.approx(70.0) | |
| def test_each_input_is_monotone_in_its_own_value(): | |
| base = {"near_miss_probability": 0.2, "accident_probability": 0.2, | |
| "severity": 0.2, "event_age_hours": 1.0} | |
| baseline = assess_risk(**base) | |
| for bump in ( | |
| {"near_miss_probability": 0.9}, | |
| {"accident_probability": 0.9}, | |
| {"severity": 0.9}, | |
| {"event_age_hours": 0.0}, # more recent -> higher recency factor | |
| ): | |
| out = assess_risk(**{**base, **bump}) | |
| assert out["risk_score"] > baseline["risk_score"] | |
| def test_recency_factor_decay(): | |
| assert recency_factor(0.0, 24.0) == pytest.approx(1.0) | |
| assert recency_factor(24.0, 24.0) == pytest.approx(0.5) | |
| assert recency_factor(48.0, 24.0) == pytest.approx(0.25) | |
| # Far enough past the half-life that the exponent underflows to 0.0. | |
| assert recency_factor(100_000.0, 24.0) == 0.0 | |
| assert 0.0 < recency_factor(10_000.0, 24.0) < 1e-100 | |
| # --------------------------------------------------------------------------- | |
| # Risk levels and threshold boundaries (thresholds are inclusive lower bounds) | |
| # --------------------------------------------------------------------------- | |
| def test_level_boundaries_default_config(): | |
| defaults = config.RISK_LEVEL_THRESHOLDS | |
| assert compute_risk_level(defaults["medium"] - 1e-9, defaults) == "LOW" | |
| assert compute_risk_level(defaults["medium"], defaults) == "MEDIUM" | |
| assert compute_risk_level(defaults["high"], defaults) == "HIGH" | |
| assert compute_risk_level(defaults["critical"], defaults) == "CRITICAL" | |
| def test_score_exactly_at_thresholds_goes_up_a_level(): | |
| # With default weights, a lone accident probability p maps to score 100*p. | |
| assert assess_risk(accident_probability=0.2499)["risk_level"] == "LOW" | |
| assert assess_risk(accident_probability=0.25)["risk_level"] == "MEDIUM" | |
| assert assess_risk(accident_probability=0.4999)["risk_level"] == "MEDIUM" | |
| assert assess_risk(accident_probability=0.50)["risk_level"] == "HIGH" | |
| assert assess_risk(accident_probability=0.7499)["risk_level"] == "HIGH" | |
| assert assess_risk(accident_probability=0.75)["risk_level"] == "CRITICAL" | |
| def test_config_override_changes_levels(): | |
| cfg = {"risk_level_thresholds": {"medium": 10.0, "high": 20.0, "critical": 30.0}} | |
| assert assess_risk(accident_probability=0.29, config=cfg)["risk_level"] == "HIGH" | |
| assert assess_risk(accident_probability=0.31, config=cfg)["risk_level"] == "CRITICAL" | |
| def test_config_override_weights_change_fusion(): | |
| cfg = {"risk_weights": {"near_miss": 1.0, "accident": 1.0, "severity": 1.0, "recency": 1.0}} | |
| out = assess_risk(near_miss_probability=0.0, accident_probability=1.0, | |
| severity=0.0, event_age_hours=0.0, config=cfg) | |
| assert out["risk_score"] == pytest.approx(50.0) | |
| def test_invalid_config_overrides_raise(): | |
| with pytest.raises(RiskInputError): | |
| assess_risk(accident_probability=0.5, config={"risk_weights": {"near_miss": 0.0}}) | |
| with pytest.raises(RiskInputError): | |
| assess_risk(accident_probability=0.5, | |
| config={"risk_level_thresholds": {"medium": 60.0, "high": 30.0, "critical": 90.0}}) | |
| with pytest.raises(RiskInputError): | |
| assess_risk(accident_probability=0.5, config={"risk_score_scale": 0.0}) | |
| with pytest.raises(RiskInputError): | |
| assess_risk(accident_probability=0.5, config={"not_a_key": 1}) | |
| # --------------------------------------------------------------------------- | |
| # Missing-input handling | |
| # --------------------------------------------------------------------------- | |
| def test_all_inputs_missing_raises(): | |
| with pytest.raises(RiskInputError, match="No risk inputs"): | |
| assess_risk() | |
| def test_each_single_input_alone_produces_a_level(): | |
| assert assess_risk(near_miss_probability=0.5)["risk_level"] in ("LOW", "MEDIUM", "HIGH", "CRITICAL") | |
| assert assess_risk(accident_probability=0.5)["risk_level"] in ("LOW", "MEDIUM", "HIGH", "CRITICAL") | |
| assert assess_risk(severity=0.5)["risk_level"] in ("LOW", "MEDIUM", "HIGH", "CRITICAL") | |
| assert assess_risk(event_age_hours=12.0)["risk_level"] in ("LOW", "MEDIUM", "HIGH", "CRITICAL") | |
| def test_missing_inputs_are_reported_and_excluded(): | |
| out = assess_risk(accident_probability=0.5, severity=0.5) | |
| assert out["missing_inputs"] == ["near_miss", "recency"] | |
| assert sorted(out["present_inputs"]) == ["accident", "severity"] | |
| # Renormalization: weights_applied covers only present inputs. | |
| assert set(out["weights_applied"]) == {"accident", "severity"} | |
| assert any("renormalized" in line for line in out["explanation"]) | |
| def test_missing_accident_and_near_miss_are_null_not_zero(): | |
| out = assess_risk(severity=0.5) | |
| assert out["near_miss_probability"] is None | |
| assert out["accident_probability"] is None | |
| # --------------------------------------------------------------------------- | |
| # Invalid-input handling (fail fast, never guess) | |
| # --------------------------------------------------------------------------- | |
| def test_invalid_inputs_raise(kwargs): | |
| with pytest.raises(RiskInputError): | |
| assess_risk(**kwargs) | |
| # --------------------------------------------------------------------------- | |
| # Output contract | |
| # --------------------------------------------------------------------------- | |
| def test_output_is_json_serializable(): | |
| out = assess_risk(near_miss_probability=0.8, accident_probability=0.6, | |
| severity=0.5, event_age_hours=24.0) | |
| parsed = json.loads(json.dumps(out)) | |
| assert parsed["risk_level"] == out["risk_level"] | |
| assert math.isfinite(parsed["risk_score"]) | |
| def test_example_schema_keys_present(): | |
| out = assess_risk(near_miss_probability=0.8, accident_probability=0.6, | |
| severity=0.5, event_age_hours=24.0) | |
| for key in ("near_miss_probability", "accident_probability", "risk_score", | |
| "risk_level", "severity", "explanation"): | |
| assert key in out | |
| def test_semantic_separation_is_enforced_and_documented(): | |
| out = assess_risk(near_miss_probability=0.9, accident_probability=0.1, severity=0.0) | |
| # The two probabilities are echoed separately and unchanged. | |
| assert out["near_miss_probability"] == 0.9 | |
| assert out["accident_probability"] == 0.1 | |
| # The score is explicitly not a probability. | |
| assert out["is_probability"] is False | |
| explanation = " ".join(out["explanation"]) | |
| assert "NOT a probability" in explanation | |
| assert "kept semantically separate" in explanation | |
| assert "raw detector confidence is never treated as a probability" in explanation | |
| assert "unconfirmed" in explanation # near-miss label semantics caveat | |
| def test_explanation_discloses_missing_inputs(): | |
| out = assess_risk(accident_probability=0.5) | |
| explanation = " ".join(out["explanation"]) | |
| assert "near_miss_probability is missing" in explanation | |
| assert "recency factor" in explanation | |
| # --------------------------------------------------------------------------- | |
| # Integration — real Phase 4 near-miss model (no retraining) | |
| # --------------------------------------------------------------------------- | |
| def test_near_miss_probability_from_real_model(near_miss_row): | |
| si = SafetyIntelligence() | |
| out = si.assess_event(near_miss_features=near_miss_row, severity=0.4, event_age_hours=6.0) | |
| assert 0.0 <= out["near_miss_probability"] <= 1.0 | |
| assert out["risk_level"] in ("LOW", "MEDIUM", "HIGH", "CRITICAL") | |
| assert "near_miss" in out["present_inputs"] | |
| assert out["model_provenance"]["retrained"] is False | |
| json.loads(to_json(out)) | |
| def test_near_miss_missing_feature_column_raises_clear_error(near_miss_row): | |
| si = SafetyIntelligence() | |
| required_feature = si._ensure_near_miss().features[0] | |
| broken = dict(near_miss_row) | |
| broken.pop(required_feature) | |
| with pytest.raises(ValueError, match="missing required.*columns"): | |
| si.assess_event(near_miss_features=broken) | |
| def test_near_miss_lazy_load_and_model_version(near_miss_row): | |
| si = SafetyIntelligence() | |
| assert not si.near_miss_model_loaded | |
| si.assess_event(near_miss_features=near_miss_row) | |
| assert si.near_miss_model_loaded | |
| # --------------------------------------------------------------------------- | |
| # Integration — Phase 6 accident detector output handling (no retraining) | |
| # --------------------------------------------------------------------------- | |
| def test_detector_output_dict_supplies_calibrated_probability(): | |
| si = SafetyIntelligence() | |
| out = si.assess_event( | |
| accident_detector_output={ | |
| "calibrated_accident_probability": 0.9, | |
| "accident_detection_confidence": 0.99, | |
| }, | |
| severity=0.5, | |
| event_age_hours=0.0, | |
| ) | |
| assert out["accident_probability"] == 0.9 | |
| # The raw confidence is carried separately, never as the probability. | |
| assert out["detector_raw_signal"]["accident_detection_confidence"] == 0.99 | |
| assert "NOT a probability" in out["detector_raw_signal"]["note"] | |
| json.loads(to_json(out)) | |
| def test_detector_output_without_calibrated_probability_is_refused(): | |
| si = SafetyIntelligence() | |
| with pytest.raises(ValueError, match="calibrated_accident_probability"): | |
| si.assess_event(accident_detector_output={"accident_detection_confidence": 0.99}) | |
| def test_explicit_probability_wins_over_detector_output(): | |
| si = SafetyIntelligence() | |
| out = si.assess_event( | |
| accident_detector_output={"calibrated_accident_probability": 0.9}, | |
| accident_probability=0.2, | |
| ) | |
| assert out["accident_probability"] == 0.2 | |
| def test_full_image_assessment_with_real_detector(safety): | |
| rng = np.random.RandomState(0) | |
| image = (rng.rand(320, 320, 3) * 255).astype(np.uint8) | |
| out = safety.assess_image(image, severity=0.5, event_age_hours=0.0) | |
| assert 0.0 <= out["accident_probability"] <= 1.0 | |
| assert 0.0 <= out["risk_score"] <= 100.0 | |
| assert out["risk_level"] in ("LOW", "MEDIUM", "HIGH", "CRITICAL") | |
| assert out["detector_output"]["detector_version"] == "accident_calibrated_detector_v1" | |
| assert "accident" in out["present_inputs"] | |
| json.loads(to_json(out)) | |
| def test_image_assessment_with_both_real_models(safety, near_miss_row): | |
| rng = np.random.RandomState(1) | |
| image = (rng.rand(320, 320, 3) * 255).astype(np.uint8) | |
| out = safety.assess_image(image, near_miss_features=near_miss_row, severity=0.2) | |
| assert 0.0 <= out["near_miss_probability"] <= 1.0 | |
| assert sorted(out["present_inputs"]) == ["accident", "near_miss", "severity"] | |
| parsed = json.loads(to_json(out)) | |
| assert parsed["risk_level"] == out["risk_level"] |