from datetime import date 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, Kandang, Site from apps.operations.models import ChickenCounting, ChickenWeight from apps.operations.services.chicken_counting_edge import ( ChickenCountingEdgeError, aggregate_counting_by_date, aggregate_mortality_by_date, coop_for_kandang, location_prefix_for_kandang, sync_chicken_counting_for_cycle, ) DB_HISTORY_SAMPLE = [ {"date": "2026-09-02", "location": "K1-L1", "total": 14591}, {"date": "2026-09-02", "location": "K1-L2", "total": 14591}, {"date": "2026-09-02", "location": "K2-L1", "total": 100}, {"date": "2026-09-02", "location": "K2-L2", "total": 100}, {"date": "2026-09-03", "location": "K1-L1", "total": 18736}, {"date": "2026-09-03", "location": "K1-L2", "total": 18736}, {"date": "2026-08-20", "location": "K1-L1", "total": 50}, ] MORTALITY_HISTORY_SAMPLE = [ {"date": "2026-09-02", "coop": "K1", "total_mortality_count": 0}, {"date": "2026-09-02", "coop": "K2", "total_mortality_count": 10}, {"date": "2026-09-03", "coop": "K1", "total_mortality_count": 29}, {"date": "2026-09-03", "coop": "K2", "total_mortality_count": 29}, ] class LocationMappingTests(TestCase): def test_maps_kandang_to_prefix_and_coop(self): self.assertEqual(location_prefix_for_kandang("Kandang 1"), "K1") self.assertEqual(coop_for_kandang("Kandang 2"), "K2") class AggregationTests(TestCase): def test_sums_matching_floors_per_date(self): by_date = aggregate_counting_by_date(DB_HISTORY_SAMPLE, kandang_name="Kandang 1") self.assertEqual(by_date[date(2026, 9, 2)], 29182) self.assertEqual(by_date[date(2026, 9, 3)], 37472) self.assertNotIn(date(2026, 9, 2), aggregate_counting_by_date([], kandang_name="Kandang 1")) def test_mortality_uses_matching_coop(self): by_date = aggregate_mortality_by_date( MORTALITY_HISTORY_SAMPLE, kandang_name="Kandang 1" ) self.assertEqual(by_date[date(2026, 9, 2)], 0) self.assertEqual(by_date[date(2026, 9, 3)], 29) @override_settings( CHICKEN_COUNTING_EDGE_BASE_URL="http://edge.internal", CHICKEN_COUNTING_EDGE_COUNTING_SYNC_ENABLED=True, CHICKEN_COUNTING_EDGE_MORTALITY_SYNC_ENABLED=True, ) class SyncChickenCountingTests(TestCase): def setUp(self): user = User.objects.create_user(user_login="sync-user", password="secret") site = Site.objects.create(site_name="Sukawarna", user=user) self.kandang = Kandang.objects.create(kandang_name="Kandang 1", site=site) self.cycle = Cycle.objects.create( kandang=self.kandang, total_days=33, start_date=date(2026, 9, 2), end_date=date(2026, 10, 4), doc_in_weight=35, doc_in_count=50000, status=Cycle.STATUS_ACTIVE, ) @patch("apps.operations.services.chicken_counting_edge.timezone.localdate") @patch("apps.operations.services.chicken_counting_edge.ChickenCountingEdgeClient.fetch_mortality_history") @patch("apps.operations.services.chicken_counting_edge.ChickenCountingEdgeClient.fetch_db_history") def test_upserts_within_cycle_window(self, mock_db, mock_mort, mock_today): mock_today.return_value = date(2026, 9, 4) mock_db.return_value = DB_HISTORY_SAMPLE mock_mort.return_value = MORTALITY_HISTORY_SAMPLE summary = sync_chicken_counting_for_cycle(self.cycle) self.assertEqual(sorted(summary["dates"]), [date(2026, 9, 2), date(2026, 9, 3)]) row_sep2 = ChickenCounting.objects.get(cycle=self.cycle, date=date(2026, 9, 2)) self.assertEqual(row_sep2.total_count, 29182) self.assertEqual(row_sep2.mortality_count, 0) row_sep3 = ChickenCounting.objects.get(cycle=self.cycle, date=date(2026, 9, 3)) self.assertEqual(row_sep3.total_count, 37472) self.assertEqual(row_sep3.mortality_count, 29) self.assertFalse( ChickenCounting.objects.filter(cycle=self.cycle, date=date(2026, 8, 20)).exists() ) @patch("apps.operations.services.chicken_counting_edge.timezone.localdate") @patch("apps.operations.services.chicken_counting_edge.ChickenCountingEdgeClient.fetch_mortality_history") @patch("apps.operations.services.chicken_counting_edge.ChickenCountingEdgeClient.fetch_db_history") def test_respects_partial_enable_flags(self, mock_db, mock_mort, mock_today): mock_today.return_value = date(2026, 9, 4) mock_db.return_value = DB_HISTORY_SAMPLE mock_mort.return_value = MORTALITY_HISTORY_SAMPLE ChickenCounting.objects.create( cycle=self.cycle, date=date(2026, 9, 2), total_count=1, mortality_count=5, ) with override_settings( CHICKEN_COUNTING_EDGE_COUNTING_SYNC_ENABLED=False, CHICKEN_COUNTING_EDGE_MORTALITY_SYNC_ENABLED=True, ): sync_chicken_counting_for_cycle(self.cycle) row = ChickenCounting.objects.get(cycle=self.cycle, date=date(2026, 9, 2)) self.assertEqual(row.total_count, 1) self.assertEqual(row.mortality_count, 0) @override_settings( CHICKEN_COUNTING_EDGE_BASE_URL="http://edge.internal", CHICKEN_COUNTING_EDGE_COUNTING_SYNC_ENABLED=True, CHICKEN_COUNTING_EDGE_MORTALITY_SYNC_ENABLED=True, DASHBOARD_PUBLISH_HOUR=17, TIME_ZONE="Asia/Jakarta", ) class ChickenCountingSyncApiTests(TestCase): def setUp(self): self.user = User.objects.create_user(user_login="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=33, start_date=date(2026, 9, 2), end_date=date(2026, 10, 4), doc_in_weight=35, doc_in_count=50000, status=Cycle.STATUS_ACTIVE, ) @patch("apps.operations.services.chicken_counting_edge.timezone.localdate") @patch("apps.operations.services.chicken_counting_edge.ChickenCountingEdgeClient.fetch_mortality_history") @patch("apps.operations.services.chicken_counting_edge.ChickenCountingEdgeClient.fetch_db_history") def test_sync_action_upserts(self, mock_db, mock_mort, mock_today): mock_today.return_value = date(2026, 9, 4) mock_db.return_value = DB_HISTORY_SAMPLE mock_mort.return_value = MORTALITY_HISTORY_SAMPLE response = self.client.post( "/api/v1/chicken-countings/sync/", {"cycle_id": self.cycle.pk}, format="json", ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual(ChickenCounting.objects.filter(cycle=self.cycle).count(), 2) @patch( "apps.operations.views.sync_chicken_counting_for_cycle", side_effect=ChickenCountingEdgeError("edge down"), ) def test_sync_action_returns_502_on_edge_error(self, _mock_sync): response = self.client.post( "/api/v1/chicken-countings/sync/", {"cycle_id": self.cycle.pk}, format="json", ) self.assertEqual(response.status_code, 502) self.assertIn("edge down", response.data["detail"]) def test_counting_and_weight_visible_before_publish_time(self): ChickenCounting.objects.create( cycle=self.cycle, date=date(2026, 9, 4), total_count=100, mortality_count=2, ) ChickenWeight.objects.create( cycle=self.cycle, date=date(2026, 9, 4), age=2, doc_weight=35, average_weight=40, chicken_count=10, uniformity=90, average_daily_gain=1, ) 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, 9, 4, 10, 0, tzinfo=wib) counting = self.client.get( "/api/v1/chicken-countings/", {"cycle_id": self.cycle.pk} ) weight = self.client.get( "/api/v1/chicken-weights/", {"cycle_id": self.cycle.pk} ) self.assertEqual(counting.status_code, 200) c_results = counting.data["results"] if isinstance(counting.data, dict) else counting.data self.assertEqual(len(c_results), 1) self.assertEqual(c_results[0]["date"], "2026-09-04") self.assertEqual(weight.status_code, 200) w_results = weight.data["results"] if isinstance(weight.data, dict) else weight.data self.assertEqual(len(w_results), 1) self.assertEqual(w_results[0]["date"], "2026-09-04")