""" tests/test_crop_risk_scorer.py ================================ Validation tests for crop_risk_scorer.compute_risk_score. These tests ensure: - numerical stability - monotonic risk behavior - correct alert threshold transitions - proper use of forecast signals """ import numpy as np import pytest from crop_risk_scorer import compute_risk_score, RiskWeights from zone_observation import ( AlertLevel, ForecastConfig, make_synthetic_zone_obs, make_synthetic_forecast_result, ) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def base_inputs(): obs = make_synthetic_zone_obs("z", seed=0) fc = make_synthetic_forecast_result("z", horizon_days=14, valid_time=obs.valid_time) return obs, fc # --------------------------------------------------------------------------- # Numerical sanity # --------------------------------------------------------------------------- class TestNumericalSanity: def test_outputs_are_finite(self, base_inputs): obs, fc = base_inputs rs = compute_risk_score(obs, fc) values = [ rs.supply_shortfall_prob, rs.drought_risk, rs.flood_risk, rs.fungi_contamination_prob, rs.quality_risk_composite, rs.confidence, ] for v in values: assert np.isfinite(v), f"Non-finite value detected: {v}" def test_outputs_in_unit_interval(self, base_inputs): obs, fc = base_inputs rs = compute_risk_score(obs, fc) for field in [ rs.supply_shortfall_prob, rs.drought_risk, rs.flood_risk, rs.fungi_contamination_prob, rs.quality_risk_composite, rs.confidence, ]: assert 0.0 <= field <= 1.0 # --------------------------------------------------------------------------- # Monotonicity (critical for RL learning) # --------------------------------------------------------------------------- class TestMonotonicity: def test_drought_increase_raises_risk(self, base_inputs): obs, fc = base_inputs obs_low = obs obs_high = make_synthetic_zone_obs("z", drought=True, seed=1) rs_low = compute_risk_score(obs_low, fc) rs_high = compute_risk_score(obs_high, fc) assert rs_high.drought_risk >= rs_low.drought_risk def test_flood_increase_raises_risk(self, base_inputs): obs, fc = base_inputs obs_low = obs obs_high = make_synthetic_zone_obs("z", flood=True, seed=2) rs_low = compute_risk_score(obs_low, fc) rs_high = compute_risk_score(obs_high, fc) assert rs_high.flood_risk >= rs_low.flood_risk def test_combined_risk_raises_supply(self, base_inputs): obs, fc = base_inputs obs_low = obs obs_high = make_synthetic_zone_obs("z", flood=True, drought=True, seed=3) rs_low = compute_risk_score(obs_low, fc) rs_high = compute_risk_score(obs_high, fc) assert rs_high.supply_shortfall_prob >= rs_low.supply_shortfall_prob # --------------------------------------------------------------------------- # Forecast influence # --------------------------------------------------------------------------- class TestForecastInfluence: def test_heavy_rain_forecast_increases_flood_risk(self, base_inputs): obs, fc = base_inputs fc_heavy = make_synthetic_forecast_result( "z", horizon_days=14, valid_time=obs.valid_time, flood=True ) rs_base = compute_risk_score(obs, fc) rs_heavy = compute_risk_score(obs, fc_heavy) assert rs_heavy.flood_risk >= rs_base.flood_risk def test_drought_forecast_increases_drought_risk(self, base_inputs): obs, fc = base_inputs fc_dry = make_synthetic_forecast_result( "z", horizon_days=14, valid_time=obs.valid_time, drought=True ) rs_base = compute_risk_score(obs, fc) rs_dry = compute_risk_score(obs, fc_dry) assert rs_dry.drought_risk >= rs_base.drought_risk # --------------------------------------------------------------------------- # Alert thresholds (VERY important) # --------------------------------------------------------------------------- class TestAlertThresholds: def test_critical_threshold(self, base_inputs): obs, fc = base_inputs from zone_observation import ForecastResult obs_extreme = make_synthetic_zone_obs("z", drought=True, flood=True, seed=10) obs_extreme.precip_anomaly_idx = -4.0 # maxes drought_signal()'s precip term obs_extreme.soil_moisture_anom = -4.0 # maxes drought_signal()'s soil term obs_extreme.flood_extent_pct = 100.0 # maxes flood_signal()'s extent term obs_extreme.drainage_risk_idx = 1.0 # maxes flood_signal()'s drainage term fc_extreme = ForecastResult( zone_id=fc.zone_id, forecast_time=fc.forecast_time, horizon_days=fc.horizon_days, precip_mm=fc.precip_mm, precip_p10=fc.precip_p10, precip_p90=fc.precip_p90, temp_mean_c=fc.temp_mean_c, temp_p10=fc.temp_p10, temp_p90=fc.temp_p90, rh_mean_pct=fc.rh_mean_pct, prob_heavy_rain=tuple([1.0] * fc.horizon_days), prob_drought_day=tuple([1.0] * fc.horizon_days), prob_high_humidity=fc.prob_high_humidity, source=fc.source, ) rs = compute_risk_score(obs_extreme, fc_extreme) assert rs.alert_level in [ AlertLevel.WARNING, AlertLevel.CRITICAL, ] def test_low_risk_produces_none_or_watch(self, base_inputs): obs, fc = base_inputs rs = compute_risk_score(obs, fc) assert rs.alert_level in [ AlertLevel.NONE, AlertLevel.WATCH, AlertLevel.ADVISORY, ] # --------------------------------------------------------------------------- # Confidence model # --------------------------------------------------------------------------- class TestConfidence: def test_confidence_in_unit_interval(self, base_inputs): obs, fc = base_inputs rs = compute_risk_score(obs, fc) assert 0 <= rs.confidence <= 1 def test_observational_data_has_higher_confidence(self): obs_obs = make_synthetic_zone_obs("z", seed=0) fc_obs = make_synthetic_forecast_result("z", horizon_days=14, valid_time=obs_obs.valid_time) obs_lowq = make_synthetic_zone_obs("z", seed=1) obs_lowq.quality_flag = 3 rs_high = compute_risk_score(obs_obs, fc_obs) rs_low = compute_risk_score(obs_lowq, fc_obs) assert rs_high.confidence >= rs_low.confidence # --------------------------------------------------------------------------- # Stability / repeatability # --------------------------------------------------------------------------- class TestStability: def test_same_inputs_same_output(self, base_inputs): obs, fc = base_inputs rs1 = compute_risk_score(obs, fc) rs2 = compute_risk_score(obs, fc) assert rs1.supply_shortfall_prob == pytest.approx(rs2.supply_shortfall_prob) def test_no_nan_under_extreme_inputs(self): obs = make_synthetic_zone_obs("z", seed=99) fc_base = make_synthetic_forecast_result("z", horizon_days=30, valid_time=obs.valid_time) from zone_observation import ForecastResult fc = ForecastResult( zone_id=fc_base.zone_id, forecast_time=fc_base.forecast_time, horizon_days=30, precip_mm=tuple([500.0] * 30), precip_p10=tuple([400.0] * 30), precip_p90=tuple([500.0] * 30), temp_mean_c=fc_base.temp_mean_c, temp_p10=fc_base.temp_p10, temp_p90=fc_base.temp_p90, rh_mean_pct=fc_base.rh_mean_pct, prob_heavy_rain=tuple([1.0] * 30), prob_drought_day=tuple([0.0] * 30), prob_high_humidity=fc_base.prob_high_humidity, source=fc_base.source, ) rs = compute_risk_score(obs, fc) assert np.isfinite(rs.supply_shortfall_prob)