Spaces:

bapi143
/

CropAI

Running on Zero

CropAI / backend /tests /test_phase2_real_pipelines.py

Deploy CropAI Deep Learning backend, YOLO11 pest model and Audio-CNN

60ed0cd 7 days ago

7.68 kB

# -*- coding: utf-8 -*-
"""
Tests for Phase 2 Real Data Pipelines:
1. Kisan Jan-Mat SQLite Persistence & Deduplication
2. Government Schemes Directory Endpoint
3. Dynamic SporeCast Outbreak Clusters
4. Voice Co-Pilot Multimodal Audio & Dialect Transcription
5. Precision Dosage Calculator CIBRC Binding
"""

import unittest
import os
import uuid
from fastapi.testclient import TestClient

# Ensure test environment key
os.environ["GEMINI_API_KEY"] = "AIzaSyDummyKeyForTestingPhase2_123456"

from backend.main import app
from backend.database import db_manager
from backend.schemes_and_polls import janmat_engine, GOVERNMENT_SCHEMES_DB
from backend.voice_copilot import detect_audio_mime_type, transcribe_audio_bytes, transcribe_and_advise


class TestPhase2RealPipelines(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.client = TestClient(app)

    def test_kisan_poll_questions_api(self):
        """Verify GET /api/v1/kisan-poll/questions returns live options with computed percentages."""
        resp = self.client.get("/api/v1/kisan-poll/questions")
        self.assertEqual(resp.status_code, 200)
        data = resp.json()
        self.assertEqual(data["status"], "success")
        self.assertGreater(data["total_options"], 0)
        
        # Verify percentages sum approximately to 100%
        options = data["options"]
        total_pct = sum(opt["vote_percentage"] for opt in options)
        self.assertGreaterEqual(total_pct, 95.0)
        self.assertLessEqual(total_pct, 105.0)

    def test_kisan_poll_vote_persistence_and_deduplication(self):
        """Verify voting persists into SQLite and duplicate votes within 24h are handled gracefully."""
        unique_device = f"DEV-TEST-{uuid.uuid4().hex[:8]}"
        payload = {
            "problem_id": "spurious_pesticides",
            "district": "Surguja",
            "state": "Chhattisgarh",
            "crop_affected": "Paddy (Rice)",
            "farmer_comment": "दुकानदार से ली गई दवा काम नहीं की।",
            "device_id": unique_device
        }

        # 1. First vote should succeed
        resp1 = self.client.post("/api/v1/kisan-poll/vote", json=payload)
        self.assertEqual(resp1.status_code, 200)
        data1 = resp1.json()
        self.assertEqual(data1["status"], "success")
        self.assertTrue(data1["ballot_id"].startswith("VOTE-"))

        # 2. Second vote with same device and problem should trigger deduplication
        resp2 = self.client.post("/api/v1/kisan-poll/vote", json=payload)
        self.assertEqual(resp2.status_code, 200)
        data2 = resp2.json()
        self.assertEqual(data2["status"], "already_voted")
        self.assertIn("पहले ही", data2["message_hi"])

    def test_government_schemes_api(self):
        """Verify GET /api/v1/schemes returns official government schemes with portal links."""
        resp = self.client.get("/api/v1/schemes")
        self.assertEqual(resp.status_code, 200)
        data = resp.json()
        self.assertEqual(data["total_schemes"], len(GOVERNMENT_SCHEMES_DB))
        self.assertGreaterEqual(data["total_schemes"], 5)
        self.assertIn("schemes", data)
        
        # Check PM-KISAN schema
        pm_kisan = next(s for s in data["schemes"] if s["id"] == "pm_kisan")
        self.assertEqual(pm_kisan["official_portal"], "https://pmkisan.gov.in/")

    def test_sporecast_dynamic_heatmap(self):
        """Verify GET /api/v1/sporecast-heatmap combines anchors with dynamic clusters."""
        # Log a mock diagnosis into diagnoses_log to simulate a live outbreak
        db_manager.log_diagnosis({
            "image_id": f"IMG-{uuid.uuid4().hex[:6]}",
            "crop": "Tomato",
            "disease": "Early Blight",
            "severity": "High",
            "confidence": 0.95,
            "is_healthy": False,
            "timestamp": "2026-10-19T20:00:00Z"
        })

        resp = self.client.get("/api/v1/sporecast-heatmap")
        self.assertEqual(resp.status_code, 200)
        data = resp.json()
        self.assertGreaterEqual(data["active_epidemics"], 3)
        self.assertIn("regional_heatmap", data)

    def test_voice_copilot_mime_detection(self):
        """Verify audio header MIME type detection."""
        fake_wav = b"RIFF\x24\x00\x00\x00WAVEfmt "
        self.assertEqual(detect_audio_mime_type(fake_wav), "audio/wav")

        fake_mp3 = b"ID3\x03\x00\x00\x00"
        self.assertEqual(detect_audio_mime_type(fake_mp3), "audio/mp3")

        fake_ogg = b"OggS\x00\x02\x00"
        self.assertEqual(detect_audio_mime_type(fake_ogg), "audio/ogg")

    def test_voice_copilot_transcribe_and_advise(self):
        """Verify audio byte transcription pipeline and dialect advisory output."""
        dummy_audio = b"RIFF" + b"\x00" * 200
        adv = transcribe_and_advise(audio_bytes=dummy_audio, dialect="ch-IN")
        self.assertEqual(adv["status"], "success")
        self.assertEqual(adv["dialect"], "ch-IN")
        self.assertIn("जय जोहार", adv["advisory_text"])
        self.assertGreater(adv["audio_stream_received_bytes"], 0)
        self.assertIn("transcribed_text", adv)

    def test_dosage_calculator_across_crops(self):
        """Verify chemical and organic dosage calculation for Wheat, Tomato, and Cotton."""
        # 1. Tomato Early Blight
        res_tomato = self.client.post("/api/v1/dosage-calculator", json={
            "crop": "Tomato",
            "disease_id": "Tomato___Early_blight",
            "farm_area_acres": 2.0,
            "knapsack_tank_liters": 16.0,
            "treatment_type": "chemical"
        })
        self.assertEqual(res_tomato.status_code, 200)
        d_tom = res_tomato.json()
        self.assertEqual(d_tom["crop"], "Tomato")
        self.assertGreater(d_tom["total_water_liters"], 300)
        self.assertGreater(d_tom["total_knapsack_tanks"], 10)

        # 2. Wheat Yellow Rust
        res_wheat = self.client.post("/api/v1/dosage-calculator", json={
            "crop": "Wheat",
            "disease_id": "Wheat___Yellow_Rust",
            "farm_area_acres": 1.0,
            "knapsack_tank_liters": 16.0,
            "treatment_type": "organic"
        })
        self.assertEqual(res_wheat.status_code, 200)
        d_wheat = res_wheat.json()
        self.assertTrue(len(d_wheat["recommended_chemical_or_bio"]) > 5)

    def test_soil_health_card_ocr_and_expanded_crops(self):
        """Verify Soil Health Card OCR endpoint and 25+ expanded ICAR crop nutrient targets."""
        from backend.soil_health import CROP_NUTRIENT_TARGETS
        # Check expanded crops
        self.assertIn("Cotton", CROP_NUTRIENT_TARGETS)
        self.assertIn("Sugarcane", CROP_NUTRIENT_TARGETS)
        self.assertIn("Mustard", CROP_NUTRIENT_TARGETS)
        self.assertIn("Potato", CROP_NUTRIENT_TARGETS)
        self.assertIn("Chickpea", CROP_NUTRIENT_TARGETS)
        self.assertIn("Banana", CROP_NUTRIENT_TARGETS)
        self.assertGreaterEqual(len(CROP_NUTRIENT_TARGETS), 20)

        # Test POST /api/v1/soil/ocr-card with dummy image bytes
        dummy_card = b"\xff\xd8\xff\xe0\x00\x10JFIF" + b"\x00" * 300
        files = {"card_photo": ("shc_card.jpg", dummy_card, "image/jpeg")}
        resp = self.client.post("/api/v1/soil/ocr-card?crop=Cotton&area_acres=2.5", files=files)
        self.assertEqual(resp.status_code, 200)
        data = resp.json()
        self.assertEqual(data["status"], "success")
        self.assertEqual(data["area_acres"], 2.5)
        self.assertIn("total_commercial_fertilizer", data)
        self.assertIn("split_schedule", data)


if __name__ == "__main__":
    unittest.main()

	# -- coding: utf-8 --
	"""
	Tests for Phase 2 Real Data Pipelines:
	1. Kisan Jan-Mat SQLite Persistence & Deduplication
	2. Government Schemes Directory Endpoint
	3. Dynamic SporeCast Outbreak Clusters
	4. Voice Co-Pilot Multimodal Audio & Dialect Transcription
	5. Precision Dosage Calculator CIBRC Binding
	"""

	import unittest
	import os
	import uuid
	from fastapi.testclient import TestClient

	# Ensure test environment key
	os.environ["GEMINI_API_KEY"] = "AIzaSyDummyKeyForTestingPhase2_123456"

	from backend.main import app
	from backend.database import db_manager
	from backend.schemes_and_polls import janmat_engine, GOVERNMENT_SCHEMES_DB
	from backend.voice_copilot import detect_audio_mime_type, transcribe_audio_bytes, transcribe_and_advise


	class TestPhase2RealPipelines(unittest.TestCase):
	@classmethod
	def setUpClass(cls):
	cls.client = TestClient(app)

	def test_kisan_poll_questions_api(self):
	"""Verify GET /api/v1/kisan-poll/questions returns live options with computed percentages."""
	resp = self.client.get("/api/v1/kisan-poll/questions")
	self.assertEqual(resp.status_code, 200)
	data = resp.json()
	self.assertEqual(data["status"], "success")
	self.assertGreater(data["total_options"], 0)

	# Verify percentages sum approximately to 100%
	options = data["options"]
	total_pct = sum(opt["vote_percentage"] for opt in options)
	self.assertGreaterEqual(total_pct, 95.0)
	self.assertLessEqual(total_pct, 105.0)

	def test_kisan_poll_vote_persistence_and_deduplication(self):
	"""Verify voting persists into SQLite and duplicate votes within 24h are handled gracefully."""
	unique_device = f"DEV-TEST-{uuid.uuid4().hex[:8]}"
	payload = {
	"problem_id": "spurious_pesticides",
	"district": "Surguja",
	"state": "Chhattisgarh",
	"crop_affected": "Paddy (Rice)",
	"farmer_comment": "दुकानदार से ली गई दवा काम नहीं की।",
	"device_id": unique_device
	}

	# 1. First vote should succeed
	resp1 = self.client.post("/api/v1/kisan-poll/vote", json=payload)
	self.assertEqual(resp1.status_code, 200)
	data1 = resp1.json()
	self.assertEqual(data1["status"], "success")
	self.assertTrue(data1["ballot_id"].startswith("VOTE-"))

	# 2. Second vote with same device and problem should trigger deduplication
	resp2 = self.client.post("/api/v1/kisan-poll/vote", json=payload)
	self.assertEqual(resp2.status_code, 200)
	data2 = resp2.json()
	self.assertEqual(data2["status"], "already_voted")
	self.assertIn("पहले ही", data2["message_hi"])

	def test_government_schemes_api(self):
	"""Verify GET /api/v1/schemes returns official government schemes with portal links."""
	resp = self.client.get("/api/v1/schemes")
	self.assertEqual(resp.status_code, 200)
	data = resp.json()
	self.assertEqual(data["total_schemes"], len(GOVERNMENT_SCHEMES_DB))
	self.assertGreaterEqual(data["total_schemes"], 5)
	self.assertIn("schemes", data)

	# Check PM-KISAN schema
	pm_kisan = next(s for s in data["schemes"] if s["id"] == "pm_kisan")
	self.assertEqual(pm_kisan["official_portal"], "https://pmkisan.gov.in/")

	def test_sporecast_dynamic_heatmap(self):
	"""Verify GET /api/v1/sporecast-heatmap combines anchors with dynamic clusters."""
	# Log a mock diagnosis into diagnoses_log to simulate a live outbreak
	db_manager.log_diagnosis({
	"image_id": f"IMG-{uuid.uuid4().hex[:6]}",
	"crop": "Tomato",
	"disease": "Early Blight",
	"severity": "High",
	"confidence": 0.95,
	"is_healthy": False,
	"timestamp": "2026-10-19T20:00:00Z"
	})

	resp = self.client.get("/api/v1/sporecast-heatmap")
	self.assertEqual(resp.status_code, 200)
	data = resp.json()
	self.assertGreaterEqual(data["active_epidemics"], 3)
	self.assertIn("regional_heatmap", data)

	def test_voice_copilot_mime_detection(self):
	"""Verify audio header MIME type detection."""
	fake_wav = b"RIFF\x24\x00\x00\x00WAVEfmt "
	self.assertEqual(detect_audio_mime_type(fake_wav), "audio/wav")

	fake_mp3 = b"ID3\x03\x00\x00\x00"
	self.assertEqual(detect_audio_mime_type(fake_mp3), "audio/mp3")

	fake_ogg = b"OggS\x00\x02\x00"
	self.assertEqual(detect_audio_mime_type(fake_ogg), "audio/ogg")

	def test_voice_copilot_transcribe_and_advise(self):
	"""Verify audio byte transcription pipeline and dialect advisory output."""
	dummy_audio = b"RIFF" + b"\x00" * 200
	adv = transcribe_and_advise(audio_bytes=dummy_audio, dialect="ch-IN")
	self.assertEqual(adv["status"], "success")
	self.assertEqual(adv["dialect"], "ch-IN")
	self.assertIn("जय जोहार", adv["advisory_text"])
	self.assertGreater(adv["audio_stream_received_bytes"], 0)
	self.assertIn("transcribed_text", adv)

	def test_dosage_calculator_across_crops(self):
	"""Verify chemical and organic dosage calculation for Wheat, Tomato, and Cotton."""
	# 1. Tomato Early Blight
	res_tomato = self.client.post("/api/v1/dosage-calculator", json={
	"crop": "Tomato",
	"disease_id": "Tomato___Early_blight",
	"farm_area_acres": 2.0,
	"knapsack_tank_liters": 16.0,
	"treatment_type": "chemical"
	})
	self.assertEqual(res_tomato.status_code, 200)
	d_tom = res_tomato.json()
	self.assertEqual(d_tom["crop"], "Tomato")
	self.assertGreater(d_tom["total_water_liters"], 300)
	self.assertGreater(d_tom["total_knapsack_tanks"], 10)

	# 2. Wheat Yellow Rust
	res_wheat = self.client.post("/api/v1/dosage-calculator", json={
	"crop": "Wheat",
	"disease_id": "Wheat___Yellow_Rust",
	"farm_area_acres": 1.0,
	"knapsack_tank_liters": 16.0,
	"treatment_type": "organic"
	})
	self.assertEqual(res_wheat.status_code, 200)
	d_wheat = res_wheat.json()
	self.assertTrue(len(d_wheat["recommended_chemical_or_bio"]) > 5)

	def test_soil_health_card_ocr_and_expanded_crops(self):
	"""Verify Soil Health Card OCR endpoint and 25+ expanded ICAR crop nutrient targets."""
	from backend.soil_health import CROP_NUTRIENT_TARGETS
	# Check expanded crops
	self.assertIn("Cotton", CROP_NUTRIENT_TARGETS)
	self.assertIn("Sugarcane", CROP_NUTRIENT_TARGETS)
	self.assertIn("Mustard", CROP_NUTRIENT_TARGETS)
	self.assertIn("Potato", CROP_NUTRIENT_TARGETS)
	self.assertIn("Chickpea", CROP_NUTRIENT_TARGETS)
	self.assertIn("Banana", CROP_NUTRIENT_TARGETS)
	self.assertGreaterEqual(len(CROP_NUTRIENT_TARGETS), 20)

	# Test POST /api/v1/soil/ocr-card with dummy image bytes
	dummy_card = b"\xff\xd8\xff\xe0\x00\x10JFIF" + b"\x00" * 300
	files = {"card_photo": ("shc_card.jpg", dummy_card, "image/jpeg")}
	resp = self.client.post("/api/v1/soil/ocr-card?crop=Cotton&area_acres=2.5", files=files)
	self.assertEqual(resp.status_code, 200)
	data = resp.json()
	self.assertEqual(data["status"], "success")
	self.assertEqual(data["area_acres"], 2.5)
	self.assertIn("total_commercial_fertilizer", data)
	self.assertIn("split_schedule", data)


	if __name__ == "__main__":
	unittest.main()