""" Unit & Integration Tests for CropAI Breakthrough Innovation Engines & APIs. """ import io import wave import struct import unittest import numpy as np try: import pytest except ImportError: pytest = None from PIL import Image from fastapi.testclient import TestClient from backend.main import app from backend.spectroscopy import spectroscopy_engine from backend.bio_acoustics import bio_acoustic_engine from backend.bio_twin import bio_twin_engine from backend.drone_vra import drone_vra_engine from backend.proof_of_farming import proof_of_farming_engine from backend.sporecast_engine import sporecast_engine client = TestClient(app) def create_synthetic_leaf_image(color=(34, 139, 34)): # Forest Green img = Image.new("RGB", (100, 100), color=color) buf = io.BytesIO() img.save(buf, format="JPEG") return buf.getvalue() def create_synthetic_audio_wav(frequency_hz=2500, duration_sec=1.0, framerate=44100): buf = io.BytesIO() with wave.open(buf, 'wb') as wav: wav.setnchannels(1) wav.setsampwidth(2) wav.setframerate(framerate) t = np.linspace(0, duration_sec, int(framerate * duration_sec), endpoint=False) signal = 0.8 * np.sin(2 * np.pi * frequency_hz * t) int_signal = (signal * 32767).astype(np.int16) wav.writeframes(int_signal.tobytes()) return buf.getvalue() class TestSpectroscopyEngine(unittest.TestCase): def test_healthy_leaf_spectroscopy(self): img_bytes = create_synthetic_leaf_image((30, 180, 50)) result = spectroscopy_engine.analyze_leaf_reflectance(img_bytes) assert result["is_leaf"] is True assert result["chlorophyll_retention_percentage"] > 50.0 assert "green_leaf_index_gli" in result assert "spectral_diagnosis" in result def test_spectroscopy_api_endpoint(self): img_bytes = create_synthetic_leaf_image() response = client.post( "/api/v1/spectroscopy/analyze", files={"file": ("leaf.jpg", img_bytes, "image/jpeg")} ) assert response.status_code == 200 data = response.json() assert data["is_leaf"] is True assert "chlorophyll_retention_percentage" in data class TestBioAcousticEngine(unittest.TestCase): def test_stem_borer_frequency_detection(self): # 2500 Hz falls directly in the 1.2 kHz - 4.5 kHz pest mastication band audio_bytes = create_synthetic_audio_wav(frequency_hz=2500) result = bio_acoustic_engine.analyze_audio_stream(audio_bytes) assert result["pest_detected"] is True assert "Yellow Stem Borer" in result["pest_identified"] or "Sub-Bark" in result["pest_identified"] assert result["dominant_frequency_hz"] > 1000 def test_bio_acoustics_api_endpoint(self): audio_bytes = create_synthetic_audio_wav(frequency_hz=2200) response = client.post( "/api/v1/bio-acoustics/listen", files={"file": ("stem_sound.wav", audio_bytes, "audio/wav")} ) assert response.status_code == 200 data = response.json() assert data["pest_detected"] is True assert "dominant_frequency_hz" in data class TestBioTwinEngine(unittest.TestCase): def test_paddy_blast_infection_risk(self): result = bio_twin_engine.simulate_microclimate_infection_risk( crop="Paddy (Rice)", temperature_c=26.0, humidity_pct=90.0, wind_speed_kmh=5.0, leaf_wetness_hours=8.0 ) assert result["spore_germination_index_pct"] > 50.0 assert "dew_point" in result["calculated_dew_point_c"].__repr__() or result["calculated_dew_point_c"] > 20.0 assert len(result["preventive_canopy_actions"]) > 0 def test_bio_twin_api_endpoint(self): response = client.get("/api/v1/bio-twin/forecast?crop=Rice&temp=27&humidity=88") assert response.status_code == 200 data = response.json() assert "spore_germination_index_pct" in data class TestDroneVRAEngine(unittest.TestCase): def test_flight_plan_generation(self): result = drone_vra_engine.generate_vra_flight_plan( field_id="FIELD-DEMO-01", center_lat=23.1188, center_lng=83.1979, field_area_acres=2.5, infected_clusters=[{"lat": 23.1189, "lng": 83.1980}] ) assert result["total_waypoints"] > 0 assert "geojson_flight_plan" in result assert result["qgroundcontrol_format_compatible"] is True def test_drone_vra_api_endpoint(self): payload = { "field_id": "FIELD-01", "center_lat": 23.1188, "center_lng": 83.1979, "field_area_acres": 1.5, "infected_clusters": [{"lat": 23.1188, "lng": 83.1979}] } response = client.post("/api/v1/drone/generate-vra-mission", json=payload) assert response.status_code == 200 data = response.json() assert data["total_waypoints"] > 0 assert "chemical_savings_percentage" in data class TestProofOfFarmingEngine(unittest.TestCase): def test_proof_of_farming_certificate(self): result = proof_of_farming_engine.generate_proof_of_farming_certificate( farmer_id="FARMER-999", farmer_name="Rameshwar Yadav", khasra_survey_number="142/2", district="Surguja", state="Chhattisgarh", crop="Basmati Rice", disease_identified="Paddy Blast", severity_percentage=45.0, affected_area_acres=2.0, gps_latitude=23.1188, gps_longitude=83.1979 ) assert result["digital_signature_sha256"] is not None assert len(result["digital_signature_sha256"]) == 64 assert result["fast_track_claim_payload"]["estimated_indemnity_payout_inr"] > 0 def test_proof_of_farming_api_endpoint(self): payload = { "farmer_id": "FARMER-999", "farmer_name": "Rameshwar Yadav", "khasra_survey_number=":"142/2", "khasra_survey_number": "142/2", "district": "Surguja", "state": "Chhattisgarh", "crop": "Basmati Rice", "disease_identified": "Paddy Blast", "severity_percentage": 45.0, "affected_area_acres": 2.0, "gps_latitude": 23.1188, "gps_longitude": 83.1979 } response = client.post("/api/v1/insurance/proof-of-farming", json=payload) assert response.status_code == 200 data = response.json() assert "digital_signature_sha256" in data class TestSporeCastAndDosageEngines(unittest.TestCase): def test_sporecast_trajectory_api(self): response = client.post("/api/v1/sporecast/simulate-trajectory?source_lat=23.1188&source_lng=83.1979&wind_speed=15&wind_direction_deg=45") assert response.status_code == 200 data = response.json() assert "trajectory_forecast_24h" in data assert "trajectory_forecast_48h" in data def test_dosage_calculator_api(self): payload = { "crop": "Paddy", "disease_id": "rice_blast", "farm_area_acres": 2.0, "knapsack_tank_liters": 16.0, "treatment_type": "organic" } response = client.post("/api/v1/dosage-calculator", json=payload) assert response.status_code == 200 data = response.json() assert data["total_water_liters"] == 400.0 assert data["total_knapsack_tanks"] > 0 if __name__ == "__main__": unittest.main()