442 lines
18 KiB
Python
442 lines
18 KiB
Python
from datetime import date, timedelta
|
|
from unittest.mock import patch
|
|
|
|
from django.test import TestCase, override_settings
|
|
from rest_framework.test import APIClient
|
|
|
|
from apps.accounts.models import User
|
|
from apps.farms.models import Cycle, Flock, Kandang, Site
|
|
from apps.operations.models import ChickenWeight, FeedSacks, IotPanel
|
|
|
|
|
|
@override_settings(DASHBOARD_PUBLISH_HOUR=0)
|
|
class OperationsAPITests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(user_name="tester", password="secret")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
site = Site.objects.create(site_name="Sukawarna", user=self.user)
|
|
self.kandang = Kandang.objects.create(kandang_name="Kandang 1", site=site)
|
|
self.cycle = Cycle.objects.create(
|
|
kandang=self.kandang,
|
|
total_days=35,
|
|
start_date=date.today() - timedelta(days=2),
|
|
end_date=date.today() + timedelta(days=32),
|
|
doc_in_weight=40,
|
|
doc_in_count=10000,
|
|
status=Cycle.STATUS_ACTIVE,
|
|
)
|
|
self.flock = Flock.objects.create(flock_name="Flock A", kandang=self.kandang)
|
|
|
|
def test_karungs_alias_lists_feed_sacks_with_date(self):
|
|
FeedSacks.objects.create(
|
|
cycle=self.cycle,
|
|
date=date.today(),
|
|
in_today=10,
|
|
out_today=8,
|
|
feed_use_today=8,
|
|
feed_use_total=8,
|
|
)
|
|
response = self.client.get("/api/v1/karungs/", {"cycle_id": self.cycle.pk})
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
results = response.data["results"] if isinstance(response.data, dict) else response.data
|
|
self.assertEqual(results[0]["date"], date.today().isoformat())
|
|
self.assertEqual(results[0]["feed_use_total"], 8)
|
|
self.assertEqual(results[0]["kandang"], self.kandang.pk)
|
|
self.assertEqual(results[0]["flock_id"], self.flock.pk)
|
|
|
|
def test_karungs_latest_returns_newest_row(self):
|
|
FeedSacks.objects.create(cycle=self.cycle, date=date.today() - timedelta(days=1), out_today=1)
|
|
newest = FeedSacks.objects.create(cycle=self.cycle, date=date.today(), out_today=9)
|
|
response = self.client.get("/api/v1/karungs/latest/", {"cycle_id": self.cycle.pk})
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
self.assertEqual(response.data["id"], newest.pk)
|
|
|
|
def test_panel_iot_accepts_cycle_id(self):
|
|
IotPanel.objects.create(
|
|
flock=self.flock,
|
|
date=date.today(),
|
|
average_temperature=28.5,
|
|
)
|
|
response = self.client.get("/api/v1/panel-iot/", {"cycle_id": self.cycle.pk})
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
results = response.data["results"] if isinstance(response.data, dict) else response.data
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0]["date"], date.today().isoformat())
|
|
|
|
def test_panel_iot_filters_by_flock(self):
|
|
flock_b = Flock.objects.create(flock_name="Flock B", kandang=self.kandang)
|
|
row = IotPanel.objects.create(
|
|
flock=self.flock,
|
|
date=date.today(),
|
|
average_temperature=28.5,
|
|
)
|
|
IotPanel.objects.create(
|
|
flock=flock_b,
|
|
date=date.today(),
|
|
average_temperature=30.0,
|
|
)
|
|
response = self.client.get(
|
|
"/api/v1/panel-iot/",
|
|
{"cycle_id": self.cycle.pk, "flock_id": self.flock.pk},
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
results = response.data["results"] if isinstance(response.data, dict) else response.data
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0]["id"], row.pk)
|
|
|
|
def test_chicken_counting_uses_date_not_cameras(self):
|
|
response = self.client.post(
|
|
"/api/v1/chicken-countings/",
|
|
{
|
|
"cycle": self.cycle.pk,
|
|
"date": date.today().isoformat(),
|
|
"total_count": 9800,
|
|
"mortality_count": 12,
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 201, response.data)
|
|
self.assertEqual(response.data["date"], date.today().isoformat())
|
|
self.assertNotIn("counting_date", response.data)
|
|
self.assertNotIn("cam1_count", response.data)
|
|
|
|
def test_weights_latest_avg_uses_date(self):
|
|
ChickenWeight.objects.create(
|
|
cycle=self.cycle,
|
|
date=date.today(),
|
|
age=10,
|
|
doc_weight=40,
|
|
average_weight=320.5,
|
|
chicken_count=9800,
|
|
uniformity=82.0,
|
|
average_daily_gain=28.0,
|
|
)
|
|
response = self.client.get(
|
|
"/api/v1/chicken-weights/latest-avg/", {"cycle_id": self.cycle.pk}
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
self.assertEqual(response.data["date"], date.today().isoformat())
|
|
self.assertEqual(response.data["average_weight"], 320.5)
|
|
self.assertNotIn("prediction_date", response.data)
|
|
|
|
def test_weights_request_endpoint_is_removed(self):
|
|
response = self.client.post(
|
|
"/api/v1/chicken-weights/request/",
|
|
{"cycle_id": self.cycle.pk, "mode": "fetch"},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 405)
|
|
|
|
def test_health_does_not_probe_fatchicken(self):
|
|
response = self.client.get("/api/v1/health/")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertNotIn("fatchicken", response.data)
|
|
self.assertNotIn("fatchicken_base_configured", response.data)
|
|
|
|
def test_manual_input_uses_erd_field_names(self):
|
|
response = self.client.post(
|
|
"/api/v1/manual-inputs/",
|
|
{
|
|
"cycle": self.cycle.pk,
|
|
"date": date.today().isoformat(),
|
|
"age_manual": 10,
|
|
"mortality_manual": 5,
|
|
"mortality_manual_total": 5,
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 201, response.data)
|
|
self.assertEqual(response.data["mortality_manual_total"], 5)
|
|
self.assertNotIn("mortality_total_manual", response.data)
|
|
|
|
def test_manual_input_computes_totals_and_ignores_client_derived_fields(self):
|
|
day1 = date.today() - timedelta(days=1)
|
|
day2 = date.today()
|
|
first = self.client.post(
|
|
"/api/v1/manual-inputs/",
|
|
{
|
|
"cycle": self.cycle.pk,
|
|
"date": day1.isoformat(),
|
|
"age_manual": 9,
|
|
"mortality_manual": 5,
|
|
"feed_in_manual": 8,
|
|
"feed_use_manual": 10,
|
|
"harvest_manual": 2,
|
|
"harvest_weight_manual": 8,
|
|
"manual_weight": 320.5,
|
|
"mortality_manual_total": 999,
|
|
"feed_in_manual_total": 999,
|
|
"feed_use_manual_total": 999,
|
|
"harvest_manual_total": 999,
|
|
"harvest_weight_manual_total": 999,
|
|
"fcr_manual": 1.9,
|
|
"eef_manual": 400,
|
|
"average_harvest_day_manual": 12,
|
|
"average_harvest_weight_manual": 99,
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertIn(first.status_code, (200, 201), first.data)
|
|
self.assertEqual(first.data["age_manual"], 9)
|
|
self.assertEqual(first.data["mortality_manual_total"], 5)
|
|
self.assertEqual(first.data["feed_in_manual_total"], 8)
|
|
self.assertEqual(first.data["feed_use_manual_total"], 10)
|
|
self.assertEqual(first.data["harvest_manual_total"], 2)
|
|
self.assertEqual(first.data["harvest_weight_manual_total"], 8)
|
|
self.assertEqual(first.data["fcr_manual"], 0)
|
|
self.assertEqual(first.data["eef_manual"], 0)
|
|
self.assertEqual(first.data["average_harvest_day_manual"], 0)
|
|
self.assertEqual(first.data["average_harvest_weight_manual"], 0)
|
|
|
|
second = self.client.post(
|
|
"/api/v1/manual-inputs/",
|
|
{
|
|
"cycle": self.cycle.pk,
|
|
"date": day2.isoformat(),
|
|
"age_manual": 10,
|
|
"mortality_manual": 3,
|
|
"feed_in_manual": 2,
|
|
"feed_use_manual": 4,
|
|
"harvest_manual": 1,
|
|
"harvest_weight_manual": 2,
|
|
"manual_weight": 330.0,
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertIn(second.status_code, (200, 201), second.data)
|
|
self.assertEqual(second.data["mortality_manual_total"], 8)
|
|
self.assertEqual(second.data["feed_in_manual_total"], 10)
|
|
self.assertEqual(second.data["feed_use_manual_total"], 14)
|
|
self.assertEqual(second.data["harvest_manual_total"], 3)
|
|
self.assertEqual(second.data["harvest_weight_manual_total"], 10)
|
|
|
|
update_day1 = self.client.post(
|
|
"/api/v1/manual-inputs/",
|
|
{
|
|
"cycle": self.cycle.pk,
|
|
"date": day1.isoformat(),
|
|
"age_manual": 9,
|
|
"mortality_manual": 7,
|
|
"feed_in_manual": 8,
|
|
"feed_use_manual": 10,
|
|
"harvest_manual": 2,
|
|
"harvest_weight_manual": 8,
|
|
"manual_weight": 320.5,
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertIn(update_day1.status_code, (200, 201), update_day1.data)
|
|
self.assertEqual(update_day1.data["id"], first.data["id"])
|
|
self.assertEqual(update_day1.data["mortality_manual_total"], 7)
|
|
|
|
refreshed_day2 = self.client.get(
|
|
"/api/v1/manual-inputs/", {"cycle_id": self.cycle.pk, "date": day2.isoformat()}
|
|
)
|
|
self.assertEqual(refreshed_day2.status_code, 200, refreshed_day2.data)
|
|
rows = (
|
|
refreshed_day2.data["results"]
|
|
if isinstance(refreshed_day2.data, dict)
|
|
else refreshed_day2.data
|
|
)
|
|
self.assertEqual(rows[0]["mortality_manual_total"], 10)
|
|
|
|
def test_manual_input_accepts_fractional_harvest_weight(self):
|
|
day1 = date.today() - timedelta(days=1)
|
|
day2 = date.today()
|
|
first = self.client.post(
|
|
"/api/v1/manual-inputs/",
|
|
{
|
|
"cycle": self.cycle.pk,
|
|
"date": day1.isoformat(),
|
|
"harvest_weight_manual": 8.5,
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertIn(first.status_code, (200, 201), first.data)
|
|
self.assertEqual(first.data["harvest_weight_manual"], 8.5)
|
|
self.assertEqual(first.data["harvest_weight_manual_total"], 8.5)
|
|
|
|
second = self.client.post(
|
|
"/api/v1/manual-inputs/",
|
|
{
|
|
"cycle": self.cycle.pk,
|
|
"date": day2.isoformat(),
|
|
"harvest_weight_manual": 1.25,
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertIn(second.status_code, (200, 201), second.data)
|
|
self.assertEqual(second.data["harvest_weight_manual"], 1.25)
|
|
self.assertEqual(second.data["harvest_weight_manual_total"], 9.75)
|
|
|
|
def test_kpi_accepts_fractional_harvest_weight(self):
|
|
from apps.operations.models import ChickenCounting, FeedSacks, KPI, ManualInput
|
|
|
|
cc = ChickenCounting.objects.create(cycle=self.cycle, date=date.today(), total_count=9800)
|
|
cw = ChickenWeight.objects.create(
|
|
cycle=self.cycle,
|
|
date=date.today(),
|
|
age=10,
|
|
doc_weight=40,
|
|
average_weight=320.5,
|
|
chicken_count=9800,
|
|
uniformity=82.0,
|
|
average_daily_gain=28.0,
|
|
)
|
|
fs = FeedSacks.objects.create(cycle=self.cycle, date=date.today())
|
|
manual = ManualInput.objects.create(cycle=self.cycle, date=date.today())
|
|
response = self.client.post(
|
|
"/api/v1/kpis/",
|
|
{
|
|
"cycle": self.cycle.pk,
|
|
"date": date.today().isoformat(),
|
|
"cc": cc.pk,
|
|
"cw": cw.pk,
|
|
"fs": fs.pk,
|
|
"manual": manual.pk,
|
|
"harvest_weight": 12.75,
|
|
"harvest_weight_total": 20.5,
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 201, response.data)
|
|
self.assertEqual(response.data["harvest_weight"], 12.75)
|
|
self.assertEqual(response.data["harvest_weight_total"], 20.5)
|
|
kpi = KPI.objects.get(pk=response.data["id"])
|
|
self.assertEqual(kpi.harvest_weight, 12.75)
|
|
self.assertEqual(kpi.harvest_weight_total, 20.5)
|
|
|
|
|
|
@override_settings(DASHBOARD_PUBLISH_HOUR=17, TIME_ZONE="Asia/Jakarta")
|
|
class DashboardVisibilityTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(user_name="tester", password="secret")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
site = Site.objects.create(site_name="Sukawarna", user=self.user)
|
|
self.kandang = Kandang.objects.create(kandang_name="Kandang 1", site=site)
|
|
self.cycle = Cycle.objects.create(
|
|
kandang=self.kandang,
|
|
total_days=35,
|
|
start_date=date(2026, 8, 26),
|
|
end_date=date(2026, 9, 29),
|
|
doc_in_weight=40,
|
|
doc_in_count=10000,
|
|
status=Cycle.STATUS_ACTIVE,
|
|
)
|
|
|
|
def test_hides_today_before_publish_time(self):
|
|
FeedSacks.objects.create(
|
|
cycle=self.cycle,
|
|
date=date(2026, 8, 27),
|
|
in_today=5,
|
|
out_today=3,
|
|
feed_use_today=3,
|
|
feed_use_total=3,
|
|
)
|
|
FeedSacks.objects.create(
|
|
cycle=self.cycle,
|
|
date=date(2026, 8, 28),
|
|
in_today=9,
|
|
out_today=7,
|
|
feed_use_today=7,
|
|
feed_use_total=10,
|
|
)
|
|
with patch(
|
|
"apps.operations.services.visibility.timezone.localtime",
|
|
) as mock_localtime:
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
wib = ZoneInfo("Asia/Jakarta")
|
|
mock_localtime.return_value = datetime(2026, 8, 28, 16, 30, tzinfo=wib)
|
|
response = self.client.get("/api/v1/karungs/", {"cycle_id": self.cycle.pk})
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
results = response.data["results"] if isinstance(response.data, dict) else response.data
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0]["date"], "2026-08-27")
|
|
|
|
def test_shows_today_after_publish_time(self):
|
|
FeedSacks.objects.create(
|
|
cycle=self.cycle,
|
|
date=date(2026, 8, 28),
|
|
in_today=9,
|
|
out_today=7,
|
|
feed_use_today=7,
|
|
feed_use_total=10,
|
|
)
|
|
with patch(
|
|
"apps.operations.services.visibility.timezone.localtime",
|
|
) as mock_localtime:
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
wib = ZoneInfo("Asia/Jakarta")
|
|
mock_localtime.return_value = datetime(2026, 8, 28, 17, 0, tzinfo=wib)
|
|
response = self.client.get("/api/v1/karungs/latest/", {"cycle_id": self.cycle.pk})
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
self.assertEqual(response.data["date"], "2026-08-28")
|
|
|
|
def test_panel_iot_not_hidden_before_publish_time(self):
|
|
flock = Flock.objects.create(flock_name="Flock A", kandang=self.kandang)
|
|
IotPanel.objects.create(
|
|
flock=flock,
|
|
date=date(2026, 8, 28),
|
|
average_temperature=28.5,
|
|
)
|
|
with patch(
|
|
"apps.operations.services.visibility.timezone.localtime",
|
|
) as mock_localtime:
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
wib = ZoneInfo("Asia/Jakarta")
|
|
mock_localtime.return_value = datetime(2026, 8, 28, 10, 0, tzinfo=wib)
|
|
response = self.client.get("/api/v1/panel-iot/", {"cycle_id": self.cycle.pk})
|
|
self.assertEqual(response.status_code, 200, response.data)
|
|
results = response.data["results"] if isinstance(response.data, dict) else response.data
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0]["date"], "2026-08-28")
|
|
|
|
def test_health_includes_visible_through_date(self):
|
|
with patch(
|
|
"apps.operations.services.visibility.timezone.localtime",
|
|
) as mock_localtime:
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
wib = ZoneInfo("Asia/Jakarta")
|
|
mock_localtime.return_value = datetime(2026, 8, 28, 10, 0, tzinfo=wib)
|
|
response = self.client.get("/api/v1/health/")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["visible_through_date"], "2026-08-27")
|
|
self.assertEqual(response.data["dashboard_publish_time"], "17:00")
|
|
|
|
|
|
class ErdPrimaryKeyTests(TestCase):
|
|
ERD_PRIMARY_KEYS = {
|
|
"user_access": "user_id",
|
|
"sites": "site_id",
|
|
"kandang": "kandang_id",
|
|
"flock": "flock_id",
|
|
"cycles": "cycle_id",
|
|
"iot_panel": "panel_id",
|
|
"ai_insight": "id",
|
|
"feed_sacks": "fs_id",
|
|
"chicken_counting": "cc_id",
|
|
"chicken_weight": "cw_id",
|
|
"manual_input": "manual_id",
|
|
"kpi": "kpi_id",
|
|
}
|
|
|
|
def test_domain_primary_keys_match_erd_names(self):
|
|
from django.db import connection
|
|
|
|
with connection.cursor() as cursor:
|
|
for table, pk_name in self.ERD_PRIMARY_KEYS.items():
|
|
cursor.execute(f'PRAGMA table_info("{table}")')
|
|
live_pks = [row[1] for row in cursor.fetchall() if row[5]]
|
|
self.assertEqual(live_pks, [pk_name], table)
|
|
|