84 lines
4.0 KiB
Python
84 lines
4.0 KiB
Python
|
from django.test import SimpleTestCase
|
||
|
|
|
||
|
|
from farm_alerts.alerts_tracker import build_farm_alerts_tracker
|
||
|
|
|
||
|
|
|
||
|
|
class FarmAlertsTrackerCanonicalTests(SimpleTestCase):
|
||
|
|
def test_whole_farm_alert_uses_aggregated_metrics(self):
|
||
|
|
ai_snapshot = {
|
||
|
|
"farm_metrics": {"resolved_metrics": {"soil_moisture": 16.0}},
|
||
|
|
"block_metrics": [{"block_code": "block-1", "resolved_metrics": {"soil_moisture": 14.0}}],
|
||
|
|
"sub_block_metrics": [
|
||
|
|
{"block_code": "block-1", "sub_block_code": "cluster-a", "resolved_metrics": {"soil_moisture": 12.0}}
|
||
|
|
],
|
||
|
|
"weather": {"forecast": {"temperature_min": 8.0, "humidity_mean": 60.0}, "source_metadata": {}},
|
||
|
|
"source_metadata": {"farm_metrics": {"canonical_source": "farmer_block_aggregated_snapshot"}},
|
||
|
|
"aggregation_policy": {"default_block_policy": "1_main_block + 1_default_sub_block_when_missing"},
|
||
|
|
}
|
||
|
|
|
||
|
|
tracker = build_farm_alerts_tracker(
|
||
|
|
sensor_id="farm-1",
|
||
|
|
context={"forecasts": []},
|
||
|
|
ai_bundle={"ai_snapshot": ai_snapshot},
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertEqual(tracker["totalAlerts"], 2)
|
||
|
|
self.assertEqual(tracker["alerts"][0]["metadata"]["evaluation_level"], "farm")
|
||
|
|
self.assertEqual(tracker["alerts"][0]["metadata"]["source"], "farm_metrics")
|
||
|
|
self.assertEqual(tracker["source_metadata"]["farm_metrics"]["canonical_source"], "farmer_block_aggregated_snapshot")
|
||
|
|
|
||
|
|
def test_block_specific_alert_includes_affected_block(self):
|
||
|
|
ai_snapshot = {
|
||
|
|
"farm_metrics": {"resolved_metrics": {"soil_moisture": 30.0}},
|
||
|
|
"block_metrics": [
|
||
|
|
{"block_code": "block-1", "resolved_metrics": {"soil_moisture": 31.0}},
|
||
|
|
{"block_code": "block-2", "resolved_metrics": {"soil_moisture": 17.0}},
|
||
|
|
],
|
||
|
|
"sub_block_metrics": [
|
||
|
|
{"block_code": "block-2", "sub_block_code": "cluster-b", "resolved_metrics": {"soil_moisture": 15.0}}
|
||
|
|
],
|
||
|
|
"weather": {"forecast": {}, "source_metadata": {}},
|
||
|
|
"source_metadata": {"farm_metrics": {}},
|
||
|
|
"aggregation_policy": {"default_block_policy": "1_main_block + 1_default_sub_block_when_missing"},
|
||
|
|
}
|
||
|
|
|
||
|
|
tracker = build_farm_alerts_tracker(
|
||
|
|
sensor_id="farm-1",
|
||
|
|
context={"forecasts": []},
|
||
|
|
ai_bundle={"ai_snapshot": ai_snapshot},
|
||
|
|
)
|
||
|
|
|
||
|
|
block_alerts = [alert for alert in tracker["alerts"] if alert.get("zone_id") == "block-2"]
|
||
|
|
self.assertEqual(len(block_alerts), 1)
|
||
|
|
self.assertEqual(block_alerts[0]["metadata"]["evaluation_level"], "block")
|
||
|
|
self.assertEqual(block_alerts[0]["metadata"]["affected_blocks"], ["block-2"])
|
||
|
|
|
||
|
|
def test_default_block_sub_block_policy_is_reported(self):
|
||
|
|
ai_snapshot = {
|
||
|
|
"farm_metrics": {"resolved_metrics": {"soil_moisture": 22.0}},
|
||
|
|
"block_metrics": [{"block_code": "default-block", "resolved_metrics": {"soil_moisture": 22.0}}],
|
||
|
|
"sub_block_metrics": [
|
||
|
|
{"block_code": "default-block", "sub_block_code": "default-sub-block", "resolved_metrics": {"soil_moisture": 22.0}}
|
||
|
|
],
|
||
|
|
"weather": {"forecast": {}, "source_metadata": {}},
|
||
|
|
"source_metadata": {"farm_metrics": {}},
|
||
|
|
"aggregation_policy": {"default_block_policy": "1_main_block + 1_default_sub_block_when_missing"},
|
||
|
|
}
|
||
|
|
|
||
|
|
tracker = build_farm_alerts_tracker(
|
||
|
|
sensor_id="farm-1",
|
||
|
|
context={"forecasts": []},
|
||
|
|
ai_bundle={"ai_snapshot": ai_snapshot},
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertEqual(
|
||
|
|
tracker["source_metadata"]["default_block_policy"],
|
||
|
|
"1_main_block + 1_default_sub_block_when_missing",
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_missing_snapshot_uses_explicit_fallback_metadata(self):
|
||
|
|
tracker = build_farm_alerts_tracker(sensor_id="farm-1", context={"forecasts": []}, ai_bundle={})
|
||
|
|
|
||
|
|
self.assertEqual(tracker["totalAlerts"], 0)
|
||
|
|
self.assertEqual(tracker["source_metadata"]["fallback"], "no_ai_snapshot")
|