# -*- 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()