"""Register lifecycle, coverage, risk, and the incubation prefilter rule.""" from __future__ import annotations import unittest from datetime import date, timedelta from types import SimpleNamespace from unittest.mock import MagicMock, patch from pipeline.stages.disease import select_prefilter_rule from pipeline.stages.incubations import ( CoveringTreatment, IncubationRecord, first_future_disease_date, forecast_has_risk, incubation_start_covered, parse_register, reconcile_register, ) from pipeline.stages.products import ProductIndex, ProductRow from pipeline.stages.treatments import TreatmentHistory _AS_OF = date(2026, 4, 16) _PAST = _AS_OF - timedelta(days=2) _TODAY = _AS_OF _FUTURE = _AS_OF + timedelta(days=3) _FAR = _AS_OF + timedelta(days=4) def _record( day: date, covered: bool, spray: CoveringTreatment | None = None, ) -> IncubationRecord: return IncubationRecord( start_date=day, infection_covered=covered, covering_treatment=spray, ) def _spray( day: date, *products: str, interval: int = 14, ) -> CoveringTreatment: names = products or ("LIETO",) return CoveringTreatment( spray_date=day, products=names, treatment_interval_min_days=interval, ) def _none(_start: date) -> None: return None def _forecasts(*days: date, label: str = "INCUBAZPRIMARIA2_80.7") -> dict[date, list[str] | None]: return {day: [label] for day in days} class ReconcileRegisterTests(unittest.TestCase): def test_covered_row_without_spray_stays_true_when_lookup_misses(self) -> None: previous = {"incubazioneprimaria2": _record(_FAR, True)} result = reconcile_register(previous, _forecasts(_FUTURE), _AS_OF, _none) record = result["incubazioneprimaria2"] self.assertEqual(record.start_date, _FUTURE) self.assertTrue(record.infection_covered) self.assertIsNone(record.covering_treatment) def test_covered_row_without_spray_stores_a_hit_and_stays_true(self) -> None: found = _spray(_FUTURE - timedelta(days=3), "LIETO", interval=10) previous = {"incubazioneprimaria2": _record(_FAR, True)} result = reconcile_register(previous, _forecasts(_FUTURE), _AS_OF, lambda _start: found) record = result["incubazioneprimaria2"] self.assertTrue(record.infection_covered) self.assertEqual(record.covering_treatment, found) def test_uncovered_lookup_stores_the_spray(self) -> None: found = _spray(_FUTURE - timedelta(days=2), interval=10) previous = {"incubazioneprimaria2": _record(_FUTURE, False)} result = reconcile_register(previous, _forecasts(_FAR), _AS_OF, lambda _start: found) record = result["incubazioneprimaria2"] self.assertEqual(record.start_date, _FAR) self.assertTrue(record.infection_covered) self.assertEqual(record.covering_treatment, found) def test_disappearance_deletes_the_entry(self) -> None: previous = {"incubazioneprimaria2": _record(_FUTURE, False)} self.assertEqual(reconcile_register(previous, {}, _AS_OF, _none), {}) def test_same_start_adopts_a_newer_spray(self) -> None: stored = _spray(_PAST - timedelta(days=10), "OLD", interval=30) newer = _spray(_PAST - timedelta(days=2), "LIETO", "TWINGO", interval=7) previous = {"incubazioneprimaria2": _record(_PAST, True, stored)} result = reconcile_register(previous, _forecasts(_PAST), _AS_OF, lambda _start: newer) record = result["incubazioneprimaria2"] self.assertTrue(record.infection_covered) self.assertEqual(record.covering_treatment, newer) self.assertEqual(record.covering_treatment.products, ("LIETO", "TWINGO")) def test_spray_on_the_start_date_does_not_replace_the_stored_spray(self) -> None: stored = _spray(_PAST - timedelta(days=6), "OLD", interval=20) on_start = _spray(_PAST, "LIETO", interval=10) previous = {"incubazioneprimaria2": _record(_PAST, True, stored)} result = reconcile_register(previous, _forecasts(_PAST), _AS_OF, lambda _start: on_start) record = result["incubazioneprimaria2"] self.assertTrue(record.infection_covered) self.assertEqual(record.covering_treatment, stored) def test_earlier_start_after_the_spray_keeps_the_stored_spray(self) -> None: stored = _spray(_AS_OF - timedelta(days=20), "OLD", interval=30) previous = {"incubazioneprimaria2": _record(_FAR, True, stored)} result = reconcile_register(previous, _forecasts(_FUTURE), _AS_OF, _none) record = result["incubazioneprimaria2"] self.assertEqual(record.start_date, _FUTURE) self.assertTrue(record.infection_covered) self.assertEqual(record.covering_treatment, stored) def test_earlier_start_on_the_spray_date_uses_the_lookup(self) -> None: stored = _spray(_FUTURE, "OLD", interval=30) previous = {"incubazioneprimaria2": _record(_FAR, True, stored)} result = reconcile_register(previous, _forecasts(_FUTURE), _AS_OF, _none) record = result["incubazioneprimaria2"] self.assertEqual(record.start_date, _FUTURE) self.assertFalse(record.infection_covered) self.assertIsNone(record.covering_treatment) def test_later_start_past_the_stored_interval_uses_the_lookup(self) -> None: stored = _spray(_FUTURE - timedelta(days=2), "OLD", interval=3) previous = {"incubazioneprimaria2": _record(_FUTURE, True, stored)} result = reconcile_register(previous, _forecasts(_FAR), _AS_OF, _none) record = result["incubazioneprimaria2"] self.assertEqual(record.start_date, _FAR) self.assertFalse(record.infection_covered) self.assertIsNone(record.covering_treatment) def test_later_date_check_reads_the_interval_from_the_register_json(self) -> None: loaded = parse_register({ "incubazioneprimaria2": { "start_date": "19-04-2026", "infection_covered": True, "covering_treatment": { "date": "17-04-2026", "products": ["LIETO"], "treatment_interval_min_days": 3, }, } }) result = reconcile_register(loaded, _forecasts(_FAR), _AS_OF, _none) record = result["incubazioneprimaria2"] self.assertEqual(record.start_date, _FAR) self.assertFalse(record.infection_covered) def test_later_landed_run_can_become_covered(self) -> None: found = _spray(_PAST - timedelta(days=4), interval=10) previous = {"incubazioneprimaria2": _record(_PAST, False)} result = reconcile_register(previous, _forecasts(_PAST), _AS_OF, lambda _start: found) record = result["incubazioneprimaria2"] self.assertEqual(record.start_date, _PAST) self.assertTrue(record.infection_covered) self.assertEqual(record.covering_treatment, found) class CoverageTests(unittest.TestCase): def _index(self) -> ProductIndex: return ProductIndex([ ProductRow( product_id="p1", product_name="LIETO", registration_number=None, is_organic_certified=False, systemicity="systemic", has_preventive_action=False, has_curative_action=True, has_eradicant_action=False, target_crops=["grapevine"], target_diseases=["downy mildew"], active_ingredients=[], frac_codes=[], ) ]) def _history(self, spray: date) -> TreatmentHistory: return TreatmentHistory(as_of=_AS_OF, by_day={spray: ["LIETO"]}) def test_spray_on_the_start_date_does_not_cover(self) -> None: with patch("pipeline.stages.incubations.fetch_all") as fetch: covered = incubation_start_covered( MagicMock(), self._index(), crop_english="grapevine", disease_english="downy mildew", start_date=_PAST, treatment_history=self._history(_PAST), ) self.assertIsNone(covered) fetch.assert_not_called() def test_spray_before_the_start_covers_when_interval_reaches(self) -> None: spray = _PAST - timedelta(days=4) uses = [SimpleNamespace(product_id="p1", treatment_interval_min_days=7)] with patch("pipeline.stages.incubations.fetch_all", return_value=uses): covered = incubation_start_covered( MagicMock(), self._index(), crop_english="grapevine", disease_english="downy mildew", start_date=_PAST, treatment_history=self._history(spray), ) self.assertIsNotNone(covered) assert covered is not None self.assertEqual(covered.spray_date, spray) self.assertEqual(covered.products, ("LIETO",)) self.assertEqual(covered.treatment_interval_min_days, 7) class RiskAndPrefilterTests(unittest.TestCase): def test_covered_only_is_not_risk(self) -> None: forecasts = {_PAST: ["INCUBAZPRIMARIA2_80.7"]} register = {"incubazioneprimaria2": _record(_PAST, True)} self.assertFalse(forecast_has_risk(forecasts, register)) def test_uncovered_incubation_is_risk(self) -> None: forecasts = {_FUTURE: ["INCUBAZPRIMARIA2_40"]} register = {"incubazioneprimaria2": _record(_FUTURE, False)} self.assertTrue(forecast_has_risk(forecasts, register)) def test_first_future_ignores_a_covered_incubation(self) -> None: forecasts = { _FUTURE: ["INCUBAZPRIMARIA2_40"], _FAR: ["FASE5"], } register = {"incubazioneprimaria2": _record(_FUTURE, True)} self.assertEqual(first_future_disease_date(forecasts, register, _AS_OF), _FAR) def test_uncovered_past_incubation_selects_incubazprimaria(self) -> None: forecasts = {_PAST: ["INCUBAZPRIMARIA2_80.7", "FASE5"]} rule = select_prefilter_rule(forecasts, {}, _AS_OF, {"incubazioneprimaria2": False}) self.assertEqual(rule, "incubazprimaria") def test_incubation_only_today_does_not_select_incubazprimaria(self) -> None: forecasts = {_TODAY: ["INCUBAZPRIMARIA2_80.7"]} rule = select_prefilter_rule(forecasts, {}, _AS_OF, {"incubazioneprimaria2": False}) self.assertIsNone(rule) def test_covered_past_incubation_falls_through_to_fase5_past(self) -> None: forecasts = {_PAST: ["INCUBAZPRIMARIA2_80.7", "FASE5"]} rule = select_prefilter_rule(forecasts, {}, _AS_OF, {"incubazioneprimaria2": True}) self.assertEqual(rule, "fase5_past") def test_fase5_only_today_selects_fase5_future(self) -> None: rule = select_prefilter_rule({_TODAY: ["FASE5"]}, {}, _AS_OF, {}) self.assertEqual(rule, "fase5_future") if __name__ == "__main__": unittest.main()