from __future__ import annotations from datetime import date, timedelta from typing import Any import uuid from django.test import TransactionTestCase from rest_framework.test import APIClient from farm_data.models import PlantCatalogSnapshot from location_data.models import NdviObservation, SoilLocation from weather.models import WeatherForecast UNSET = object() def square_boundary(lat: float, lon: float, delta: float = 0.01) -> dict[str, Any]: return { "type": "Polygon", "coordinates": [ [ [lon - delta, lat - delta], [lon + delta, lat - delta], [lon + delta, lat + delta], [lon - delta, lat + delta], [lon - delta, lat - delta], ] ], } class IntegrationAPITestCase(TransactionTestCase): reset_sequences = True databases = {"default"} primary_lat = 35.700000 primary_lon = 51.400000 forecast_start = date.today() def setUp(self) -> None: super().setUp() self.client = APIClient() self._next_backend_plant_id = 100 self.primary_boundary = square_boundary(self.primary_lat, self.primary_lon) self.primary_location = self.create_complete_location( lat=self.primary_lat, lon=self.primary_lon, boundary=self.primary_boundary, ) self.seed_weather_forecasts(self.primary_location, start=self.forecast_start, days=7) self.seed_ndvi_observation(self.primary_location) def create_complete_location( self, *, lat: float, lon: float, boundary: dict[str, Any] | None = None, **_ignored: Any, ) -> SoilLocation: location = SoilLocation.objects.create( latitude=f"{lat:.6f}", longitude=f"{lon:.6f}", farm_boundary=boundary or square_boundary(lat, lon), ) return location def seed_weather_forecasts( self, location: SoilLocation, *, start: date, days: int, temperature_base: float = 22.0, et0_base: float = 3.4, ) -> list[WeatherForecast]: forecasts: list[WeatherForecast] = [] for day_index in range(days): forecasts.append( WeatherForecast.objects.create( location=location, forecast_date=start + timedelta(days=day_index), temperature_min=12.0 + day_index, temperature_max=temperature_base + day_index, temperature_mean=17.0 + day_index, precipitation=1.2 if day_index % 3 == 0 else 0.0, precipitation_probability=35.0 + day_index, humidity_mean=48.0 + day_index, wind_speed_max=10.0 + day_index, et0=et0_base + (day_index * 0.2), weather_code=0 if day_index == 0 else 2, ) ) return forecasts def seed_ndvi_observation( self, location: SoilLocation, *, observation_date: date | None = None, mean_ndvi: float = 0.73, ) -> NdviObservation: return NdviObservation.objects.create( location=location, observation_date=observation_date or self.forecast_start, mean_ndvi=mean_ndvi, ndvi_map={"type": "FeatureCollection", "features": []}, vegetation_health_class="Healthy", satellite_source="sentinel-2", metadata={"suite": "integration"}, ) def create_irrigation_method_via_api(self, name: str, **overrides: Any) -> dict[str, Any]: payload = { "name": name, "category": "localized", "description": "Primary drip line for the farm", "water_efficiency_percent": 91.0, "water_pressure_required": "1.5 bar", "flow_rate": "4 l/h", "coverage_area": "row-based", "soil_type": "loam", "climate_suitability": "dry", } payload.update(overrides) response = self.client.post("/api/irrigation/", data=payload, format="json") self.assertEqual(response.status_code, 201, response.json()) return response.json()["data"] def create_plant_via_api(self, name: str, **overrides: Any) -> dict[str, Any]: backend_plant_id = int(overrides.pop("id", self._next_backend_plant_id)) self._next_backend_plant_id = max(self._next_backend_plant_id, backend_plant_id + 1) payload = { "id": backend_plant_id, "name": name, "icon": "leaf", "light": "full sun", "watering": "every 2 days", "soil": "loamy", "temperature": "20-28C", "growth_stage": "vegetative", "growth_stages": ["vegetative"], "planting_season": "spring", "harvest_time": "90 days", "spacing": "50 cm", "fertilizer": "balanced NPK", } payload.update(overrides) if "growth_stages" not in overrides: payload["growth_stages"] = [payload["growth_stage"]] if payload.get("growth_stage") else [] response = self.client.post("/api/farm-data/plants/sync/", data=[payload], format="json") self.assertEqual(response.status_code, 200, response.json()) snapshot = PlantCatalogSnapshot.objects.get(backend_plant_id=backend_plant_id) return { "id": snapshot.backend_plant_id, "backend_plant_id": snapshot.backend_plant_id, "name": snapshot.name, "icon": snapshot.icon, "light": snapshot.light, "watering": snapshot.watering, "soil": snapshot.soil, "temperature": snapshot.temperature, "growth_stage": snapshot.growth_stage, "growth_stages": list(snapshot.growth_stages or []), "planting_season": snapshot.planting_season, "harvest_time": snapshot.harvest_time, "spacing": snapshot.spacing, "fertilizer": snapshot.fertilizer, } def create_sensor_parameter_via_api(self, **overrides: Any) -> dict[str, Any]: payload = { "sensor_key": "sensor-7-1", "code": "soil_moisture", "name_fa": "soil moisture", "unit": "%", "data_type": "float", "metadata": {"min": 0, "max": 100}, } payload.update(overrides) response = self.client.post("/api/farm-data/parameters/", data=payload, format="json") self.assertEqual(response.status_code, 201, response.json()) return response.json()["data"] def upsert_farm_via_api( self, *, farm_uuid: uuid.UUID, plant_ids: list[int] | None = None, irrigation_method_id: int | None | object = UNSET, sensor_payload: dict[str, Any] | None = None, boundary: dict[str, Any] | None = None, ) -> dict[str, Any]: payload: dict[str, Any] = { "farm_uuid": str(farm_uuid), "farm_boundary": boundary or self.primary_boundary, } if plant_ids is not None: payload["plant_ids"] = plant_ids if irrigation_method_id is not UNSET: payload["irrigation_method_id"] = irrigation_method_id if sensor_payload is not None: payload["sensor_payload"] = sensor_payload response = self.client.post("/api/farm-data/", data=payload, format="json") self.assertIn(response.status_code, {200, 201}, response.json()) return response.json()["data"]