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, aggregate_weight_by_date, coop_for_kandang, location_prefix_for_kandang, sync_chicken_counting_for_cycle, sync_chicken_weight_for_cycle, ) WEIGHT_PREDICTIONS_SAMPLE = [ { "date": "2026-09-02", "camera_id": "FUSED", "cycle_day": 103, "predicted_weight_g": 1626.9, "doc_weight_g": 42.0, "n_valid_samples": 3215, "coop": "K1", "location": "K1-L1", "uniformity": 85.0, }, { "date": "2026-09-02", "camera_id": "FUSED", "cycle_day": 103, "predicted_weight_g": 1626.9, "doc_weight_g": 42.0, "n_valid_samples": 3215, "coop": "K1", "location": "K1-L2", "uniformity": 85.0, }, { "date": "2026-09-02", "camera_id": "CC1", "cycle_day": 103, "predicted_weight_g": 1600.0, "n_valid_samples": 100, "coop": "K1", "location": "K1-L1", "uniformity": 80.0, }, { "date": "2026-09-03", "camera_id": "FUSED", "cycle_day": 1, "predicted_weight_g": 45.3, "n_valid_samples": 18580, "coop": "K1", "location": "K1-L1", "uniformity": 18.7, }, { "date": "2026-09-03", "camera_id": "FUSED", "cycle_day": 1, "predicted_weight_g": 45.3, "n_valid_samples": 18580, "coop": "K1", "location": "K1-L2", "uniformity": 18.7, }, { "date": "2026-09-03", "camera_id": "FUSED", "cycle_day": 1, "predicted_weight_g": 99.0, "n_valid_samples": 10, "coop": "K2", "location": "K2-L1", "uniformity": 50.0, }, { "date": "2026-09-04", "camera_id": "FUSED", "cycle_day": 2, "predicted_weight_g": 58.2, "n_valid_samples": 20751, "coop": "K1", "location": "K1-L1", "uniformity": 17.2, }, { "date": "2026-08-20", "camera_id": "FUSED", "cycle_day": 0, "predicted_weight_g": 40.0, "n_valid_samples": 100, "coop": "K1", "location": "K1-L1", "uniformity": 90.0, }, ] 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) def test_weight_uses_fused_coop_rows_and_averages_floors(self): by_date = aggregate_weight_by_date( WEIGHT_PREDICTIONS_SAMPLE, kandang_name="Kandang 1" ) self.assertEqual(by_date[date(2026, 9, 2)]["average_weight"], 1626.9) self.assertEqual(by_date[date(2026, 9, 2)]["uniformity"], 85.0) # Identical floor duplicates must not double-count samples. self.assertEqual(by_date[date(2026, 9, 2)]["chicken_count"], 3215) self.assertEqual(by_date[date(2026, 9, 3)]["average_weight"], 45.3) self.assertEqual(by_date[date(2026, 9, 3)]["uniformity"], 18.7) self.assertNotIn(date(2026, 9, 3), aggregate_weight_by_date([], kandang_name="Kandang 1")) # K2 row must not leak into K1. self.assertNotEqual(by_date[date(2026, 9, 3)]["average_weight"], 99.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, ) class SyncChickenCountingTests(TestCase): def setUp(self): user = User.objects.create_user(user_name="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") @patch( "apps.operations.services.visibility.visible_through_date", return_value=date(2026, 9, 3), # publish cutoff still yesterday ) def test_open_cycle_syncs_through_today_not_publish_cutoff( self, _mock_visible, mock_db, mock_mort, mock_today ): """Open cycles must ingest today's edge rows before DASHBOARD_PUBLISH_HOUR.""" mock_today.return_value = date(2026, 9, 4) open_cycle = Cycle.objects.create( kandang=self.kandang, total_days=1, start_date=date(2026, 9, 2), end_date=None, doc_in_weight=35, doc_in_count=50000, status=Cycle.STATUS_ACTIVE, ) mock_db.return_value = DB_HISTORY_SAMPLE + [ {"date": "2026-09-04", "location": "K1-L1", "total": 100}, {"date": "2026-09-04", "location": "K1-L2", "total": 100}, ] mock_mort.return_value = MORTALITY_HISTORY_SAMPLE + [ {"date": "2026-09-04", "coop": "K1", "total_mortality_count": 3}, ] summary = sync_chicken_counting_for_cycle(open_cycle) self.assertIn(date(2026, 9, 4), summary["dates"]) row = ChickenCounting.objects.get(cycle=open_cycle, date=date(2026, 9, 4)) self.assertEqual(row.total_count, 200) self.assertEqual(row.mortality_count, 3) @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_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=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") @override_settings( CHICKEN_COUNTING_EDGE_BASE_URL="http://edge.internal", CHICKEN_COUNTING_EDGE_WEIGHT_SYNC_ENABLED=True, ) class SyncChickenWeightTests(TestCase): def setUp(self): user = User.objects.create_user(user_name="weight-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_weight_predictions" ) def test_day0_uses_doc_weight_and_zero_uniformity(self, mock_pred, mock_today): mock_today.return_value = date(2026, 9, 4) mock_pred.return_value = WEIGHT_PREDICTIONS_SAMPLE summary = sync_chicken_weight_for_cycle(self.cycle) self.assertEqual( sorted(summary["dates"]), [date(2026, 9, 2), date(2026, 9, 3), date(2026, 9, 4)], ) day0 = ChickenWeight.objects.get(cycle=self.cycle, date=date(2026, 9, 2)) self.assertEqual(day0.age, 0) self.assertEqual(day0.doc_weight, 35) self.assertEqual(day0.average_weight, 35.0) self.assertEqual(day0.uniformity, 0.0) self.assertEqual(day0.average_daily_gain, 0.0) self.assertEqual(day0.chicken_count, 50000) day1 = ChickenWeight.objects.get(cycle=self.cycle, date=date(2026, 9, 3)) self.assertEqual(day1.age, 1) self.assertEqual(day1.average_weight, 45.3) self.assertEqual(day1.uniformity, 18.7) self.assertAlmostEqual(day1.average_daily_gain, 10.3) self.assertEqual(day1.chicken_count, 18580) day2 = ChickenWeight.objects.get(cycle=self.cycle, date=date(2026, 9, 4)) self.assertEqual(day2.average_weight, 58.2) self.assertAlmostEqual(day2.average_daily_gain, 12.9) self.assertFalse( ChickenWeight.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_weight_predictions" ) def test_creates_day0_even_without_edge_row(self, mock_pred, mock_today): mock_today.return_value = date(2026, 9, 4) mock_pred.return_value = [ row for row in WEIGHT_PREDICTIONS_SAMPLE if row["date"] != "2026-09-02" ] sync_chicken_weight_for_cycle(self.cycle) day0 = ChickenWeight.objects.get(cycle=self.cycle, date=date(2026, 9, 2)) self.assertEqual(day0.average_weight, 35.0) self.assertEqual(day0.uniformity, 0.0) @patch( "apps.operations.services.chicken_counting_edge.ChickenCountingEdgeClient.fetch_weight_predictions" ) def test_skips_when_weight_sync_disabled(self, mock_pred): with override_settings(CHICKEN_COUNTING_EDGE_WEIGHT_SYNC_ENABLED=False): summary = sync_chicken_weight_for_cycle(self.cycle) self.assertTrue(summary["skipped"]) mock_pred.assert_not_called() self.assertEqual(ChickenWeight.objects.filter(cycle=self.cycle).count(), 0) @override_settings( CHICKEN_COUNTING_EDGE_BASE_URL="http://edge.internal", CHICKEN_COUNTING_EDGE_WEIGHT_SYNC_ENABLED=True, ) class ChickenWeightSyncApiTests(TestCase): def setUp(self): self.user = User.objects.create_user(user_name="weight-api", 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_weight_predictions" ) def test_sync_action_upserts(self, mock_pred, mock_today): mock_today.return_value = date(2026, 9, 4) mock_pred.return_value = WEIGHT_PREDICTIONS_SAMPLE response = self.client.post( "/api/v1/chicken-weights/sync/", {"cycle_id": self.cycle.pk}, format="json", ) self.assertEqual(response.status_code, 200, response.data) self.assertEqual(ChickenWeight.objects.filter(cycle=self.cycle).count(), 3) day0 = ChickenWeight.objects.get(cycle=self.cycle, date=date(2026, 9, 2)) self.assertEqual(day0.average_weight, 35.0) self.assertEqual(day0.uniformity, 0.0) @patch( "apps.operations.views.sync_chicken_weight_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-weights/sync/", {"cycle_id": self.cycle.pk}, format="json", ) self.assertEqual(response.status_code, 502) self.assertIn("edge down", response.data["detail"])