207 lines
8.6 KiB
Python
207 lines
8.6 KiB
Python
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from django.urls import Resolver404, resolve
|
|
from rest_framework.test import APIRequestFactory, force_authenticate
|
|
|
|
from external_api_adapter.adapter import AdapterResponse
|
|
from farm_hub.models import FarmHub, FarmType
|
|
|
|
from .views import (
|
|
ClusterBlockLiveView,
|
|
ClusterRecommendationsView,
|
|
KOptionsActivateView,
|
|
KOptionsView,
|
|
LocationDataNdviHealthView,
|
|
LocationDataRemoteSensingView,
|
|
LocationDataView,
|
|
RunStatusView,
|
|
)
|
|
|
|
|
|
CLUSTER_UUID = "11111111-2222-3333-4444-555555555555"
|
|
|
|
|
|
class LocationDataProxyViewTests(TestCase):
|
|
def setUp(self):
|
|
self.factory = APIRequestFactory()
|
|
self.user = get_user_model().objects.create_user(
|
|
username="location-user",
|
|
password="secret123",
|
|
email="location@example.com",
|
|
phone_number="09120000030",
|
|
)
|
|
self.other_user = get_user_model().objects.create_user(
|
|
username="location-other-user",
|
|
password="secret123",
|
|
email="location-other@example.com",
|
|
phone_number="09120000031",
|
|
)
|
|
self.farm_type = FarmType.objects.create(name="Location Farm Type")
|
|
self.farm = FarmHub.objects.create(owner=self.user, farm_type=self.farm_type, name="Farm 1")
|
|
self.other_farm = FarmHub.objects.create(owner=self.other_user, farm_type=self.farm_type, name="Farm 2")
|
|
|
|
def _get(self, path):
|
|
request = self.factory.get(path)
|
|
force_authenticate(request, user=self.user)
|
|
return request
|
|
|
|
def _post(self, path, data):
|
|
request = self.factory.post(path, data, format="json")
|
|
force_authenticate(request, user=self.user)
|
|
return request
|
|
|
|
@patch("crop_zoning.views.external_api_request")
|
|
def test_get_location_data_proxies_query_params_to_ai(self, mock_external_api_request):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"code": 200,
|
|
"msg": "success",
|
|
"data": {"source": "database", "id": 12, "lon": "51.389000", "lat": "35.689200"},
|
|
},
|
|
)
|
|
|
|
response = LocationDataView.as_view()(self._get("/api/location-data/?lat=35.6892&lon=51.389"))
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["data"]["id"], 12)
|
|
mock_external_api_request.assert_called_once_with(
|
|
"ai",
|
|
"/api/location-data/",
|
|
method="GET",
|
|
payload=None,
|
|
query={"lat": "35.6892", "lon": "51.389"},
|
|
)
|
|
|
|
def test_post_location_data_rejects_foreign_farm_uuid(self):
|
|
response = LocationDataView.as_view()(
|
|
self._post("/api/location-data/", {"farm_uuid": str(self.other_farm.farm_uuid), "lat": "35.6892", "lon": "51.389"})
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertEqual(response.data, {"code": 404, "msg": "location پیدا نشد.", "data": None})
|
|
|
|
@patch("crop_zoning.views.external_api_request")
|
|
def test_post_ndvi_health_proxies_owned_farm_to_ai(self, mock_external_api_request):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=200,
|
|
data={
|
|
"code": 200,
|
|
"msg": "success",
|
|
"data": {"ndviIndex": 0.63, "vegetation_health_class": "healthy"},
|
|
},
|
|
)
|
|
|
|
response = LocationDataNdviHealthView.as_view()(
|
|
self._post("/api/location-data/ndvi-health/", {"farm_uuid": str(self.farm.farm_uuid)})
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["data"]["ndviIndex"], 0.63)
|
|
mock_external_api_request.assert_called_once_with(
|
|
"ai",
|
|
"/api/location-data/ndvi-health/",
|
|
method="POST",
|
|
payload={"farm_uuid": str(self.farm.farm_uuid)},
|
|
query=None,
|
|
)
|
|
|
|
def test_get_remote_sensing_rejects_foreign_farm_uuid(self):
|
|
response = LocationDataRemoteSensingView.as_view()(
|
|
self._get(f"/api/location-data/remote-sensing/?farm_uuid={self.other_farm.farm_uuid}")
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertEqual(response.data, {"code": 404, "msg": "مزرعه پیدا نشد.", "data": None})
|
|
|
|
@patch("crop_zoning.views.external_api_request")
|
|
def test_post_remote_sensing_passes_through_202_response(self, mock_external_api_request):
|
|
mock_external_api_request.return_value = AdapterResponse(
|
|
status_code=202,
|
|
data={
|
|
"code": 202,
|
|
"msg": "queued",
|
|
"data": {"status": "processing", "task_id": "task-123"},
|
|
},
|
|
)
|
|
|
|
response = LocationDataRemoteSensingView.as_view()(
|
|
self._post("/api/location-data/remote-sensing/", {"farm_uuid": str(self.farm.farm_uuid)})
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 202)
|
|
self.assertEqual(response.data["data"]["task_id"], "task-123")
|
|
mock_external_api_request.assert_called_once_with(
|
|
"ai",
|
|
"/api/location-data/remote-sensing/",
|
|
method="POST",
|
|
payload={"farm_uuid": str(self.farm.farm_uuid)},
|
|
query=None,
|
|
)
|
|
|
|
@patch("crop_zoning.views.external_api_request")
|
|
def test_auxiliary_location_data_endpoints_proxy_to_ai(self, mock_external_api_request):
|
|
mock_external_api_request.side_effect = [
|
|
AdapterResponse(status_code=200, data={"code": 200, "msg": "success", "data": {"status": "success"}}),
|
|
AdapterResponse(status_code=200, data={"code": 200, "msg": "success", "data": {"cluster_count": 2}}),
|
|
AdapterResponse(status_code=200, data={"code": 200, "msg": "success", "data": {"result_id": 5}}),
|
|
AdapterResponse(status_code=200, data={"code": 200, "msg": "success", "data": {"activated_requested_k": 4}}),
|
|
AdapterResponse(status_code=200, data={"code": 200, "msg": "success", "data": {"status": "running"}}),
|
|
]
|
|
|
|
cluster_response = ClusterBlockLiveView.as_view()(
|
|
self._get(f"/api/location-data/remote-sensing/cluster-blocks/{CLUSTER_UUID}/live/"),
|
|
cluster_uuid=CLUSTER_UUID,
|
|
)
|
|
recommendation_response = ClusterRecommendationsView.as_view()(
|
|
self._get(f"/api/location-data/remote-sensing/cluster-recommendations/?farm_uuid={self.farm.farm_uuid}")
|
|
)
|
|
k_options_response = KOptionsView.as_view()(
|
|
self._get("/api/location-data/remote-sensing/results/5/k-options/"),
|
|
result_id=5,
|
|
)
|
|
activate_response = KOptionsActivateView.as_view()(
|
|
self._post("/api/location-data/remote-sensing/results/5/k-options/activate/", {"requested_k": 4}),
|
|
result_id=5,
|
|
)
|
|
run_status_response = RunStatusView.as_view()(
|
|
self._get("/api/location-data/remote-sensing/runs/9/status/"),
|
|
run_id=9,
|
|
)
|
|
|
|
self.assertEqual(cluster_response.status_code, 200)
|
|
self.assertEqual(recommendation_response.data["data"]["cluster_count"], 2)
|
|
self.assertEqual(k_options_response.data["data"]["result_id"], 5)
|
|
self.assertEqual(activate_response.data["data"]["activated_requested_k"], 4)
|
|
self.assertEqual(run_status_response.data["data"]["status"], "running")
|
|
|
|
def test_new_routes_exist_and_old_crop_zoning_routes_are_removed(self):
|
|
self.assertIs(resolve("/api/location-data/").func.view_class, LocationDataView)
|
|
self.assertIs(resolve("/api/location-data/ndvi-health/").func.view_class, LocationDataNdviHealthView)
|
|
self.assertIs(resolve("/api/location-data/remote-sensing/").func.view_class, LocationDataRemoteSensingView)
|
|
self.assertIs(
|
|
resolve(f"/api/location-data/remote-sensing/cluster-blocks/{CLUSTER_UUID}/live/").func.view_class,
|
|
ClusterBlockLiveView,
|
|
)
|
|
self.assertIs(
|
|
resolve("/api/location-data/remote-sensing/cluster-recommendations/").func.view_class,
|
|
ClusterRecommendationsView,
|
|
)
|
|
self.assertIs(
|
|
resolve("/api/location-data/remote-sensing/results/5/k-options/").func.view_class,
|
|
KOptionsView,
|
|
)
|
|
self.assertIs(
|
|
resolve("/api/location-data/remote-sensing/results/5/k-options/activate/").func.view_class,
|
|
KOptionsActivateView,
|
|
)
|
|
self.assertIs(
|
|
resolve("/api/location-data/remote-sensing/runs/9/status/").func.view_class,
|
|
RunStatusView,
|
|
)
|
|
|
|
with self.assertRaises(Resolver404):
|
|
resolve("/api/crop-zoning/area/")
|