368 lines
14 KiB
Python
368 lines
14 KiB
Python
from unittest.mock import patch
|
|
|
|
from django.core.cache import cache
|
|
from django.test import TestCase, override_settings
|
|
from rest_framework.test import APIRequestFactory
|
|
|
|
from external_api_adapter.adapter import AdapterResponse
|
|
from farm_hub.models import FarmHub, FarmType
|
|
from account.models import User
|
|
|
|
from .views import SoilAnomalyDetectionView, SoilMoistureHeatmapView, SoilSummaryView
|
|
|
|
|
|
TEST_CACHES = {
|
|
"default": {
|
|
"BACKEND": "django.core.cache.backends.locmem.LocMemCache",
|
|
"LOCATION": "soil-tests",
|
|
}
|
|
}
|
|
|
|
TEST_SOIL_SUMMARY_CACHE_TTL = 14400
|
|
TEST_SOIL_ANOMALIES_CACHE_TTL = 14400
|
|
|
|
|
|
@override_settings(
|
|
CACHES=TEST_CACHES,
|
|
SOIL_ANOMALIES_CACHE_TTL=TEST_SOIL_ANOMALIES_CACHE_TTL,
|
|
)
|
|
class SoilAnomalyDetectionViewTests(TestCase):
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.factory = APIRequestFactory()
|
|
self.user = User.objects.create_user(
|
|
username="soil-user",
|
|
password="secret123",
|
|
email="soil@example.com",
|
|
phone_number="09120000100",
|
|
)
|
|
self.farm_type = FarmType.objects.create(name="Soil Farm Type")
|
|
self.farm = FarmHub.objects.create(
|
|
owner=self.user,
|
|
farm_type=self.farm_type,
|
|
name="Soil Farm",
|
|
)
|
|
|
|
@patch("soil.views.external_api_request")
|
|
def test_anomalies_proxy_to_soile_anomaly_detection(self, mock_external_api_request):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(self.farm.farm_uuid),
|
|
"summary": "summary",
|
|
"explanation": "explanation",
|
|
"likely_cause": "cause",
|
|
"recommended_action": "action",
|
|
"monitoring_priority": "high",
|
|
"confidence": 0.91,
|
|
"generated_at": "2026-04-26T10:00:00Z",
|
|
"anomalies": [],
|
|
"interpretation": {},
|
|
"knowledge_base": None,
|
|
"raw_response": None,
|
|
}
|
|
},
|
|
)
|
|
|
|
request = self.factory.get(f"/api/soil/anomalies/?farm_uuid={self.farm.farm_uuid}")
|
|
response = SoilAnomalyDetectionView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["code"], 200)
|
|
self.assertEqual(response.data["data"]["monitoring_priority"], "high")
|
|
mock_external_api_request.assert_called_once_with(
|
|
"ai",
|
|
"/api/soile/anomaly-detection/",
|
|
method="POST",
|
|
payload={"farm_uuid": str(self.farm.farm_uuid)},
|
|
)
|
|
|
|
@patch("soil.views.external_api_request")
|
|
def test_anomalies_cache_last_four_responses(self, mock_external_api_request):
|
|
for index in range(5):
|
|
farm = FarmHub.objects.create(owner=self.user, farm_type=self.farm_type, name=f"Soil Farm Cache {index}")
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(farm.farm_uuid),
|
|
"summary": f"summary {index}",
|
|
"anomalies": [],
|
|
}
|
|
},
|
|
)
|
|
|
|
request = self.factory.get(f"/api/soil/anomalies/?farm_uuid={farm.farm_uuid}")
|
|
response = SoilAnomalyDetectionView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
cached_items = cache.get("soil:anomalies:recent")
|
|
|
|
self.assertEqual(len(cached_items), 4)
|
|
self.assertEqual(cached_items[0]["summary"], "summary 4")
|
|
self.assertEqual(cached_items[-1]["summary"], "summary 1")
|
|
|
|
@patch("soil.views.external_api_request")
|
|
def test_anomalies_return_cached_response_for_same_farm(self, mock_external_api_request):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(self.farm.farm_uuid),
|
|
"summary": "cached summary",
|
|
"anomalies": [],
|
|
}
|
|
},
|
|
)
|
|
|
|
for _ in range(2):
|
|
request = self.factory.get(f"/api/soil/anomalies/?farm_uuid={self.farm.farm_uuid}")
|
|
response = SoilAnomalyDetectionView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["data"]["summary"], "cached summary")
|
|
|
|
self.assertEqual(cache.get(f"soil:anomalies:{self.farm.farm_uuid}")["summary"], "cached summary")
|
|
mock_external_api_request.assert_called_once()
|
|
|
|
@patch("soil.views.cache.set")
|
|
@patch("soil.views.external_api_request")
|
|
def test_anomalies_use_env_ttl_for_cache(self, mock_external_api_request, mock_cache_set):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(self.farm.farm_uuid),
|
|
"summary": "summary",
|
|
"anomalies": [],
|
|
}
|
|
},
|
|
)
|
|
|
|
request = self.factory.get(f"/api/soil/anomalies/?farm_uuid={self.farm.farm_uuid}")
|
|
response = SoilAnomalyDetectionView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(any(call.kwargs.get("timeout") == TEST_SOIL_ANOMALIES_CACHE_TTL for call in mock_cache_set.call_args_list))
|
|
|
|
def test_anomalies_require_farm_uuid(self):
|
|
request = self.factory.get("/api/soil/anomalies/")
|
|
response = SoilAnomalyDetectionView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertEqual(response.data["code"], 400)
|
|
self.assertEqual(response.data["data"]["farm_uuid"][0], "This field is required.")
|
|
|
|
def test_anomalies_return_404_for_missing_farm(self):
|
|
request = self.factory.get("/api/soil/anomalies/?farm_uuid=11111111-1111-1111-1111-111111111111")
|
|
response = SoilAnomalyDetectionView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertEqual(response.data["code"], 404)
|
|
self.assertEqual(response.data["data"]["farm_uuid"][0], "Farm not found.")
|
|
|
|
|
|
class SoilMoistureHeatmapViewTests(TestCase):
|
|
def setUp(self):
|
|
self.factory = APIRequestFactory()
|
|
self.user = User.objects.create_user(
|
|
username="soil-heatmap-user",
|
|
password="secret123",
|
|
email="soil-heatmap@example.com",
|
|
phone_number="09120000101",
|
|
)
|
|
self.farm_type = FarmType.objects.create(name="Soil Heatmap Farm Type")
|
|
self.farm = FarmHub.objects.create(
|
|
owner=self.user,
|
|
farm_type=self.farm_type,
|
|
name="Heatmap Farm",
|
|
)
|
|
|
|
@patch("soil.views.external_api_request")
|
|
def test_heatmap_proxies_to_soile_moisture_heatmap(self, mock_external_api_request):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(self.farm.farm_uuid),
|
|
"location": {},
|
|
"current_sensor": {},
|
|
"soil_profile": [],
|
|
"timestamp": "2026-04-26T10:00:00Z",
|
|
"grid_resolution": {},
|
|
"grid_cells": [],
|
|
"sensor_points": [],
|
|
"quality_legend": {},
|
|
"depth_layers": [],
|
|
"model_metadata": {},
|
|
"summary": {},
|
|
}
|
|
},
|
|
)
|
|
|
|
request = self.factory.get(f"/api/soil/moisture-heatmap/?farm_uuid={self.farm.farm_uuid}")
|
|
response = SoilMoistureHeatmapView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["code"], 200)
|
|
self.assertEqual(response.data["data"]["farm_uuid"], str(self.farm.farm_uuid))
|
|
mock_external_api_request.assert_called_once_with(
|
|
"ai",
|
|
"/api/soile/moisture-heatmap/",
|
|
method="POST",
|
|
payload={"farm_uuid": str(self.farm.farm_uuid)},
|
|
)
|
|
|
|
def test_heatmap_requires_farm_uuid(self):
|
|
request = self.factory.get("/api/soil/moisture-heatmap/")
|
|
response = SoilMoistureHeatmapView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertEqual(response.data["code"], 400)
|
|
self.assertEqual(response.data["data"]["farm_uuid"][0], "This field is required.")
|
|
|
|
def test_heatmap_returns_404_for_missing_farm(self):
|
|
request = self.factory.get("/api/soil/moisture-heatmap/?farm_uuid=11111111-1111-1111-1111-111111111111")
|
|
response = SoilMoistureHeatmapView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertEqual(response.data["code"], 404)
|
|
self.assertEqual(response.data["data"]["farm_uuid"][0], "Farm not found.")
|
|
|
|
|
|
@override_settings(
|
|
CACHES=TEST_CACHES,
|
|
SOIL_SUMMARY_CACHE_TTL=TEST_SOIL_SUMMARY_CACHE_TTL,
|
|
)
|
|
class SoilSummaryViewTests(TestCase):
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.factory = APIRequestFactory()
|
|
self.user = User.objects.create_user(
|
|
username="soil-summary-user",
|
|
password="secret123",
|
|
email="soil-summary@example.com",
|
|
phone_number="09120000102",
|
|
)
|
|
self.farm_type = FarmType.objects.create(name="Soil Summary Farm Type")
|
|
self.farm = FarmHub.objects.create(
|
|
owner=self.user,
|
|
farm_type=self.farm_type,
|
|
name="Summary Farm",
|
|
)
|
|
|
|
@patch("soil.views.external_api_request")
|
|
def test_summary_proxies_to_soile_health_summary(self, mock_external_api_request):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(self.farm.farm_uuid),
|
|
"healthScore": 82,
|
|
"profileSource": "Tomato",
|
|
"healthScoreDetails": {},
|
|
"healthLanguage": {},
|
|
"avgSoilMoisture": 46,
|
|
"avgSoilMoistureRaw": 46.0,
|
|
"avgSoilMoistureStatus": "بهینه",
|
|
}
|
|
},
|
|
)
|
|
|
|
request = self.factory.get(f"/api/soil/summary/?farm_uuid={self.farm.farm_uuid}")
|
|
response = SoilSummaryView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["code"], 200)
|
|
self.assertEqual(response.data["data"]["healthScore"], 82)
|
|
mock_external_api_request.assert_called_once_with(
|
|
"ai",
|
|
"/api/soile/health-summary/",
|
|
method="POST",
|
|
payload={"farm_uuid": str(self.farm.farm_uuid)},
|
|
)
|
|
|
|
@patch("soil.views.external_api_request")
|
|
def test_summary_returns_cached_response_for_same_farm(self, mock_external_api_request):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(self.farm.farm_uuid),
|
|
"healthScore": 82,
|
|
"profileSource": "Tomato",
|
|
}
|
|
},
|
|
)
|
|
|
|
for _ in range(2):
|
|
request = self.factory.get(f"/api/soil/summary/?farm_uuid={self.farm.farm_uuid}")
|
|
response = SoilSummaryView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["data"]["healthScore"], 82)
|
|
|
|
self.assertEqual(cache.get(f"soil:summary:{self.farm.farm_uuid}")["healthScore"], 82)
|
|
mock_external_api_request.assert_called_once()
|
|
|
|
@patch("soil.views.cache.set")
|
|
@patch("soil.views.external_api_request")
|
|
def test_summary_uses_env_ttl_for_cache(self, mock_external_api_request, mock_cache_set):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(self.farm.farm_uuid),
|
|
"healthScore": 82,
|
|
}
|
|
},
|
|
)
|
|
|
|
request = self.factory.get(f"/api/soil/summary/?farm_uuid={self.farm.farm_uuid}")
|
|
response = SoilSummaryView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(any(call.kwargs.get("timeout") == TEST_SOIL_SUMMARY_CACHE_TTL for call in mock_cache_set.call_args_list))
|
|
|
|
@patch("soil.views.external_api_request")
|
|
def test_summary_caches_last_four_responses(self, mock_external_api_request):
|
|
for index in range(5):
|
|
farm = FarmHub.objects.create(owner=self.user, farm_type=self.farm_type, name=f"Soil Summary Cache {index}")
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"data": {
|
|
"farm_uuid": str(farm.farm_uuid),
|
|
"healthScore": 80 + index,
|
|
}
|
|
},
|
|
)
|
|
|
|
request = self.factory.get(f"/api/soil/summary/?farm_uuid={farm.farm_uuid}")
|
|
response = SoilSummaryView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
cached_items = cache.get("soil:summary:recent")
|
|
self.assertEqual(len(cached_items), 4)
|
|
self.assertEqual(cached_items[0]["healthScore"], 84)
|
|
self.assertEqual(cached_items[-1]["healthScore"], 81)
|
|
|
|
def test_summary_requires_farm_uuid(self):
|
|
request = self.factory.get("/api/soil/summary/")
|
|
response = SoilSummaryView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertEqual(response.data["code"], 400)
|
|
self.assertEqual(response.data["data"]["farm_uuid"][0], "This field is required.")
|
|
|
|
def test_summary_returns_404_for_missing_farm(self):
|
|
request = self.factory.get("/api/soil/summary/?farm_uuid=11111111-1111-1111-1111-111111111111")
|
|
response = SoilSummaryView.as_view()(request)
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertEqual(response.data["code"], 404)
|
|
self.assertEqual(response.data["data"]["farm_uuid"][0], "Farm not found.")
|