File size: 4,050 Bytes
b3555b4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | # -*- coding: utf-8 -*-
"""
Test Suite for AgriMemory™ Outbreak Recurrence & Healthy Crop Guardrails.
Verifies:
1. Healthy crops (Tomato___healthy, Rice___healthy) are NEVER recorded or flagged as at-risk outbreaks.
2. check_seasonal_disease_alerts handles historical dates cleanly without NameError.
3. Prophylactic care plans are generated only for valid disease IDs.
"""
import unittest
import os
import tempfile
from datetime import datetime, timedelta
from backend.farm_memory import AgriMemoryEngine
class TestFarmMemoryEngine(unittest.TestCase):
def setUp(self):
# Create a temporary file for isolated engine storage testing
self.temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".json")
self.temp_file.close()
self.engine = AgriMemoryEngine(storage_path=self.temp_file.name)
def tearDown(self):
if os.path.exists(self.temp_file.name):
os.remove(self.temp_file.name)
def test_healthy_crop_flagged_correctly(self):
"""Verify that healthy crop diagnoses are marked with is_healthy=True."""
res = self.engine.save_diagnosis_record(
crop="Tomato",
disease="Healthy Foliage (स्वस्थ पौधा)",
disease_id="Tomato___healthy",
severity="Healthy",
confidence=0.98,
temperature_c=25.0,
humidity_pct=70.0,
weather_condition="Clear Sky",
diagnosis_date=datetime.now().isoformat()
)
self.assertEqual(res["status"], "success")
self.assertTrue(res["record"]["is_healthy"])
def test_healthy_crop_never_triggers_seasonal_alarm(self):
"""Verify that past healthy records within current calendar window NEVER trigger risk alerts."""
# Save a healthy scan from 1 year ago on the exact same date
last_year_date = (datetime.now() - timedelta(days=365)).isoformat()
self.engine.save_diagnosis_record(
crop="Tomato",
disease="Healthy Tomato Leaf",
disease_id="Tomato___healthy",
severity="Healthy",
confidence=0.99,
temperature_c=26.0,
humidity_pct=75.0,
weather_condition="Optimal",
diagnosis_date=last_year_date
)
# Check alerts for current day
alerts = self.engine.check_seasonal_disease_alerts(
current_date=datetime.now().isoformat(),
current_temp_c=26.0,
current_humidity_pct=75.0
)
# Ensure no alert is generated for the healthy crop!
for alert in alerts:
self.assertNotEqual(alert.get("disease_id"), "Tomato___healthy")
self.assertFalse("healthy" in alert.get("disease", "").lower())
self.assertFalse("स्वस्थ" in alert.get("disease", ""))
def test_real_disease_triggers_prior_year_alarm(self):
"""Verify that real severe outbreaks from prior year correctly trigger seasonal warnings without NameError."""
last_year_date = (datetime.now() - timedelta(days=365)).isoformat()
self.engine.save_diagnosis_record(
crop="Wheat",
disease="Yellow Rust (गेहूं का पीला रतुआ)",
disease_id="Wheat___Yellow_Rust",
severity="Severe",
confidence=0.95,
temperature_c=27.0,
humidity_pct=88.0,
weather_condition="Optimal Humidity",
diagnosis_date=last_year_date
)
alerts = self.engine.check_seasonal_disease_alerts(
current_date=datetime.now().isoformat(),
current_temp_c=27.0,
current_humidity_pct=88.0
)
# Should generate alert for Wheat Yellow Rust
rust_alerts = [a for a in alerts if "Wheat___Yellow_Rust" in a.get("disease_id", "")]
self.assertTrue(len(rust_alerts) >= 1)
self.assertTrue(rust_alerts[0]["recurrence_risk_score"] > 80.0)
if __name__ == "__main__":
unittest.main()
|