sajad-dev/Backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UPDATE

Browse Source
This commit is contained in:
Mohammad Sajad Pourajam
2026-05-02 06:16:36 +03:30
parent bd02c8342a
commit f34d5dd198
38 changed files with 71 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
irrigation/tests.py
+461
View File
@@ -0,0 +1,461 @@
from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.test import TestCase
from rest_framework.test import APIRequestFactory, force_authenticate
from external_api_adapter.adapter import AdapterResponse
from farm_hub.models import FarmHub, FarmType
from .models import IrrigationRecommendationRequest
from .views import (
IrrigationMethodListView,
PlanFromTextView,
RecommendView,
RecommendationDetailView,
RecommendationListView,
WaterStressView,
)
class WaterStressViewTests(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.user = get_user_model().objects.create_user(
username="farmer",
password="secret123",
email="farmer@example.com",
phone_number="09120000000",
)
self.other_user = get_user_model().objects.create_user(
username="other-farmer",
password="secret123",
email="other@example.com",
phone_number="09120000001",
)
self.farm_type = FarmType.objects.create(name="زراعی")
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")
@patch("irrigation.views.external_api_request")
def test_post_proxies_request_to_ai_service(self, mock_external_api_request):
mock_external_api_request.return_value = AdapterResponse(
status_code=200,
data={
"data": {
"result": {
"waterStressIndex": 12,
"level": "پایین",
"sourceMetric": {"soilMoisture": 24},
}
}
},
)
request = self.factory.post(
"/api/irrigation/water-stress/",
{"farm_uuid": str(self.farm.farm_uuid)},
format="json",
)
force_authenticate(request, user=self.user)
response = WaterStressView.as_view()(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["code"], 200)
self.assertEqual(response.data["msg"], "success")
self.assertEqual(response.data["data"]["farm_uuid"], str(self.farm.farm_uuid))
self.assertEqual(response.data["data"]["waterStressIndex"], 12)
self.assertEqual(response.data["data"]["level"], "پایین")
self.assertEqual(response.data["data"]["sourceMetric"], {"soilMoisture": 24})
mock_external_api_request.assert_called_once_with(
"ai",
"/api/irrigation/water-stress/",
method="POST",
payload={"farm_uuid": str(self.farm.farm_uuid)},
)
def test_post_rejects_foreign_farm_uuid(self):
request = self.factory.post(
"/api/irrigation/water-stress/",
{"farm_uuid": str(self.other_farm.farm_uuid)},
format="json",
)
force_authenticate(request, user=self.user)
response = WaterStressView.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 IrrigationPlanFromTextViewTests(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.user = get_user_model().objects.create_user(
username="plan-parser-user",
password="secret123",
email="plan-parser@example.com",
phone_number="09120000005",
)
self.farm_type = FarmType.objects.create(name="گلخانه ای")
self.farm = FarmHub.objects.create(owner=self.user, farm_type=self.farm_type, name="Plan Parser Farm")
@patch("irrigation.views.external_api_request")
def test_plan_from_text_proxies_to_ai_service(self, mock_external_api_request):
mock_external_api_request.return_value = AdapterResponse(
status_code=200,
data={
"code": 200,
"msg": "موفق",
"data": {
"status": "completed",
"status_fa": "تکمیل شد",
"summary": "done",
"missing_fields": [],
"questions": [],
"collected_data": {"crop_name": "گوجه فرنگی"},
"final_plan": {"crop_name": "گوجه فرنگی"},
},
},
)
request = self.factory.post(
"/api/irrigation/plan-from-text/",
{"message": "متن برنامه", "farm_uuid": str(self.farm.farm_uuid)},
format="json",
)
force_authenticate(request, user=self.user)
response = PlanFromTextView.as_view()(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["data"]["status"], "completed")
mock_external_api_request.assert_called_once_with(
"ai",
"/api/irrigation/plan-from-text/",
method="POST",
payload={"message": "متن برنامه", "farm_uuid": str(self.farm.farm_uuid)},
)
def test_plan_from_text_requires_message_or_answers_or_partial_plan(self):
request = self.factory.post("/api/irrigation/plan-from-text/", {}, format="json")
force_authenticate(request, user=self.user)
response = PlanFromTextView.as_view()(request)
self.assertEqual(response.status_code, 400)
self.assertIn("non_field_errors", response.data)
class IrrigationMethodListViewTests(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
@patch("irrigation.views.external_api_request")
def test_get_proxies_irrigation_methods_from_ai(self, mock_external_api_request):
mock_external_api_request.return_value = AdapterResponse(
status_code=200,
data={
"data": [
{
"id": 1,
"name": "Drip",
"category": "micro",
"description": "Efficient irrigation",
"water_efficiency_percent": 90.0,
}
]
},
)
request = self.factory.get("/api/irrigation/")
response = IrrigationMethodListView.as_view()(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["code"], 200)
self.assertEqual(response.data["data"][0]["name"], "Drip")
mock_external_api_request.assert_called_once_with(
"ai",
"/api/irrigation/",
method="GET",
)
@patch("irrigation.views.external_api_request")
def test_post_proxies_irrigation_method_creation_to_ai(self, mock_external_api_request):
mock_external_api_request.return_value = AdapterResponse(
status_code=201,
data={
"data": {
"id": 1,
"name": "Drip",
"category": "micro",
}
},
)
request = self.factory.post("/api/irrigation/", {"name": "Drip"}, format="json")
response = IrrigationMethodListView.as_view()(request)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.data["data"]["name"], "Drip")
mock_external_api_request.assert_called_once_with(
"ai",
"/api/irrigation/",
method="POST",
payload={"name": "Drip"},
)
class RecommendViewTests(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.user = get_user_model().objects.create_user(
username="recommend-farmer",
password="secret123",
email="recommend@example.com",
phone_number="09120000002",
)
self.farm_type = FarmType.objects.create(name="باغی")
self.farm = FarmHub.objects.create(
owner=self.user,
farm_type=self.farm_type,
name="Recommend Farm",
irrigation_method_id=3,
irrigation_method_name="آبیاری قطره ای",
)
@patch("irrigation.views.external_api_request")
def test_post_returns_full_recommendation_shape(self, mock_external_api_request):
mock_external_api_request.return_value = AdapterResponse(
status_code=200,
data={
"data": {
"result": {
"plan": {
"frequencyPerWeek": 4,
"durationMinutes": 38,
"bestTimeOfDay": "05:30 تا 08:00 صبح",
"moistureLevel": 72,
"warning": "در ساعات گرم روز آبیاری انجام نشود",
},
"water_balance": {
"active_kc": 0.93,
"crop_profile": {
"kc_initial": 0.55,
"kc_mid": 1.05,
"kc_end": 0.78,
},
"daily": [
{
"forecast_date": "2025-02-12",
"et0_mm": 5.4,
"etc_mm": 4.9,
"effective_rainfall_mm": 0,
"gross_irrigation_mm": 17,
"irrigation_timing": "05:30 - 07:00",
}
],
},
"timeline": [
{
"step_number": 1,
"title": "بررسی فشار",
"description": "فشار ابتدا و انتهای لاین کنترل شود",
}
],
"sections": [
{
"title": "هشدار تبخیر بالا",
"icon": "tabler-alert-triangle",
"type": "warning",
"content": "در ساعات گرم روز آبیاری انجام نشود",
},
{
"title": "نکته بهره وری",
"icon": "tabler-bulb",
"type": "tip",
"content": "شست وشوی فیلترها به یکنواختی آبیاری کمک می کند",
},
],
}
}
},
)
request = self.factory.post(
"/api/irrigation/recommend/",
{
"farm_uuid": str(self.farm.farm_uuid),
"plant_name": "گوجه فرنگی",
"growth_stage": "گلدهی",
},
format="json",
)
force_authenticate(request, user=self.user)
response = RecommendView.as_view()(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["code"], 200)
self.assertIn("recommendation_uuid", response.data["data"])
self.assertEqual(response.data["data"]["status"], IrrigationRecommendationRequest.STATUS_PENDING_CONFIRMATION)
self.assertEqual(response.data["data"]["status_label"], "منتظر تایید")
self.assertEqual(response.data["data"]["plan"]["durationMinutes"], 38)
self.assertEqual(response.data["data"]["water_balance"]["active_kc"], 0.93)
self.assertEqual(response.data["data"]["timeline"][0]["step_number"], 1)
self.assertEqual(response.data["data"]["sections"][0]["type"], "warning")
mock_external_api_request.assert_called_once_with(
"ai",
"/api/irrigation/recommend/",
method="POST",
payload={
"farm_uuid": str(self.farm.farm_uuid),
"plant_name": "گوجه فرنگی",
"growth_stage": "گلدهی",
"irrigation_method_id": 3,
"irrigation_type": "آبیاری قطره ای",
"irrigation_method_name": "آبیاری قطره ای",
},
)
class IrrigationRecommendationHistoryTests(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.user = get_user_model().objects.create_user(
username="history-farmer",
password="secret123",
email="history@example.com",
phone_number="09120000003",
)
self.other_user = get_user_model().objects.create_user(
username="other-history-farmer",
password="secret123",
email="other-history@example.com",
phone_number="09120000004",
)
self.farm_type = FarmType.objects.create(name="گلخانه ای")
self.farm = FarmHub.objects.create(owner=self.user, farm_type=self.farm_type, name="History Farm")
self.other_farm = FarmHub.objects.create(owner=self.other_user, farm_type=self.farm_type, name="Other History Farm")
def test_recommendation_list_returns_paginated_items(self):
first = IrrigationRecommendationRequest.objects.create(
farm=self.farm,
crop_id="گندم",
growth_stage="vegetative",
status=IrrigationRecommendationRequest.STATUS_COMPLETED,
request_payload={"irrigation_method_name": "بارانی"},
response_payload={"data": {"plan": {"durationMinutes": 20}}},
)
second = IrrigationRecommendationRequest.objects.create(
farm=self.farm,
crop_id="ذرت",
growth_stage="flowering",
status=IrrigationRecommendationRequest.STATUS_PENDING_CONFIRMATION,
request_payload={"irrigation_method_name": "قطره ای"},
response_payload={"data": {"plan": {"durationMinutes": 35}}},
)
request = self.factory.get(
f"/api/irrigation/recommendations/?farm_uuid={self.farm.farm_uuid}&page=1&page_size=1"
)
force_authenticate(request, user=self.user)
response = RecommendationListView.as_view()(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["code"], 200)
self.assertEqual(len(response.data["data"]), 1)
self.assertEqual(response.data["pagination"]["total_items"], 2)
self.assertEqual(response.data["data"][0]["recommendation_uuid"], str(second.uuid))
self.assertEqual(response.data["data"][0]["plant_name"], "ذرت")
self.assertEqual(response.data["data"][0]["growth_stage"], "flowering")
self.assertEqual(response.data["data"][0]["irrigation_method_name"], "قطره ای")
self.assertEqual(response.data["data"][0]["status"], IrrigationRecommendationRequest.STATUS_PENDING_CONFIRMATION)
self.assertEqual(response.data["data"][0]["status_label"], "منتظر تایید")
self.assertNotEqual(response.data["data"][0]["recommendation_uuid"], str(first.uuid))
def test_recommendation_detail_returns_saved_shape(self):
recommendation = IrrigationRecommendationRequest.objects.create(
farm=self.farm,
crop_id="گوجه فرنگی",
growth_stage="fruiting",
status=IrrigationRecommendationRequest.STATUS_COMPLETED,
request_payload={"irrigation_method_name": "قطره ای"},
response_payload={
"data": {
"result": {
"plan": {"frequencyPerWeek": 4, "durationMinutes": 30},
"water_balance": {"active_kc": 0.93, "daily": []},
"timeline": [{"step_number": 1, "title": "مرحله اول", "description": "اجرا شود"}],
"sections": [{"type": "tip", "title": "نکته", "content": "صبح زود آبیاری شود"}],
}
}
},
)
request = self.factory.get(f"/api/irrigation/recommendations/{recommendation.uuid}/")
force_authenticate(request, user=self.user)
response = RecommendationDetailView.as_view()(request, recommendation_uuid=recommendation.uuid)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["code"], 200)
self.assertEqual(response.data["data"]["recommendation_uuid"], str(recommendation.uuid))
self.assertEqual(response.data["data"]["crop_id"], "گوجه فرنگی")
self.assertEqual(response.data["data"]["plant_name"], "گوجه فرنگی")
self.assertEqual(response.data["data"]["growth_stage"], "fruiting")
self.assertEqual(response.data["data"]["irrigation_method_name"], "قطره ای")
self.assertEqual(response.data["data"]["status"], IrrigationRecommendationRequest.STATUS_COMPLETED)
self.assertEqual(response.data["data"]["status_label"], "پایان یافته")
self.assertEqual(response.data["data"]["plan"]["durationMinutes"], 30)
self.assertEqual(response.data["data"]["timeline"][0]["step_number"], 1)
self.assertEqual(response.data["data"]["sections"][0]["type"], "tip")
def test_recommendation_detail_rejects_foreign_recommendation(self):
recommendation = IrrigationRecommendationRequest.objects.create(
farm=self.other_farm,
crop_id="خیار",
status=IrrigationRecommendationRequest.STATUS_PENDING_CONFIRMATION,
)
request = self.factory.get(f"/api/irrigation/recommendations/{recommendation.uuid}/")
force_authenticate(request, user=self.user)
response = RecommendationDetailView.as_view()(request, recommendation_uuid=recommendation.uuid)
self.assertEqual(response.status_code, 404)
self.assertEqual(response.data["msg"], "Recommendation not found.")
@patch("irrigation.views.external_api_request")
def test_post_accepts_sensor_uuid_as_farm_uuid_alias(self, mock_external_api_request):
mock_external_api_request.return_value = AdapterResponse(
status_code=200,
data={"data": {"result": {"sections": []}}},
)
request = self.factory.post(
"/api/irrigation/recommend/",
{
"sensor_uuid": str(self.farm.farm_uuid),
},
format="json",
)
force_authenticate(request, user=self.user)
response = RecommendView.as_view()(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["data"]["plan"]["frequencyPerWeek"], 4)
mock_external_api_request.assert_called_once_with(
"ai",
"/api/irrigation/recommend/",
method="POST",
payload={
"farm_uuid": str(self.farm.farm_uuid),
"irrigation_method_id": 3,
"irrigation_type": "آبیاری قطره ای",
"irrigation_method_name": "آبیاری قطره ای",
},
)