| """ |
| 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)): |
| 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): |
| |
| 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() |
|
|
