124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
import httpx
|
|
from django.test import SimpleTestCase, TestCase
|
|
|
|
from apps.accounts.models import User
|
|
from apps.sync.models import ActiveSite
|
|
from apps.sync.services.farm_export_client import FarmExportClient, FarmExportError
|
|
|
|
|
|
def _ops_page(*, page: int, page_size: int, count: int, results: list) -> dict:
|
|
return {
|
|
"sites": [],
|
|
"data": {
|
|
"kpis": {
|
|
"count": count,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"results": results,
|
|
},
|
|
"manual_inputs": {"count": 0, "page": page, "page_size": page_size, "results": []},
|
|
"feed_sacks": {"count": 0, "page": page, "page_size": page_size, "results": []},
|
|
"chicken_countings": {
|
|
"count": 0,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"results": [],
|
|
},
|
|
"chicken_weights": {
|
|
"count": 0,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"results": [],
|
|
},
|
|
"ai_insights": {"count": 0, "page": page, "page_size": page_size, "results": []},
|
|
},
|
|
}
|
|
|
|
|
|
class FarmExportClientTests(TestCase):
|
|
def setUp(self):
|
|
self.gm = User.objects.create_user(
|
|
user_name="gm-export", password="x", status=User.STATUS_GM
|
|
)
|
|
self.active = ActiveSite.objects.create(
|
|
code="sukawarna",
|
|
name="Sukawarna",
|
|
source_site_id="sukawarna-0002",
|
|
api_base_url="https://farm.example/api/v1",
|
|
api_key="site-secret",
|
|
managed_by=self.gm,
|
|
)
|
|
|
|
def test_401_raises_farm_export_error(self):
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(401, json={"detail": "nope"})
|
|
|
|
client = FarmExportClient(
|
|
self.active, transport=httpx.MockTransport(handler)
|
|
)
|
|
with self.assertRaises(FarmExportError) as ctx:
|
|
client.fetch_ops_export(site_id="sukawarna-0002")
|
|
self.assertEqual(ctx.exception.status_code, 401)
|
|
|
|
def test_collect_ops_series_paginates_until_done(self):
|
|
calls = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
page = int(request.url.params.get("page", "1"))
|
|
calls.append(page)
|
|
self.assertEqual(request.url.params.get("site_id"), "sukawarna-0002")
|
|
self.assertEqual(request.url.params.get("cycle_id"), "9")
|
|
self.assertEqual(request.headers.get("X-API-Key"), "site-secret")
|
|
if page == 1:
|
|
body = _ops_page(
|
|
page=1,
|
|
page_size=2,
|
|
count=3,
|
|
results=[{"date": "2026-09-01"}, {"date": "2026-09-02"}],
|
|
)
|
|
else:
|
|
body = _ops_page(
|
|
page=2,
|
|
page_size=2,
|
|
count=3,
|
|
results=[{"date": "2026-09-03"}],
|
|
)
|
|
return httpx.Response(200, json=body)
|
|
|
|
client = FarmExportClient(
|
|
self.active, transport=httpx.MockTransport(handler)
|
|
)
|
|
series = client.collect_ops_series(
|
|
site_id="sukawarna-0002", cycle_id=9, page_size=2
|
|
)
|
|
self.assertEqual(calls, [1, 2])
|
|
self.assertEqual(len(series["kpis"]), 3)
|
|
|
|
def test_missing_api_key_rejected(self):
|
|
self.active.api_key = ""
|
|
with self.assertRaises(FarmExportError):
|
|
FarmExportClient(self.active)
|
|
|
|
|
|
class SelectCycleIdsUnitTests(SimpleTestCase):
|
|
def test_selects_open_and_last_three_closed(self):
|
|
from apps.sync.services.mirror_pull import select_source_cycle_ids
|
|
|
|
rows = [
|
|
{"id": 1, "kandang": 10, "status": "active"},
|
|
{"id": 2, "kandang": 10, "status": "pending_close"},
|
|
{"id": 3, "kandang": 10, "status": "closed", "end_date": "2026-01-01"},
|
|
{"id": 4, "kandang": 10, "status": "closed", "end_date": "2026-02-01"},
|
|
{"id": 5, "kandang": 10, "status": "closed", "end_date": "2026-03-01"},
|
|
{"id": 6, "kandang": 10, "status": "closed", "end_date": "2026-04-01"},
|
|
{"id": 7, "kandang": 11, "status": "closed", "end_date": "2026-05-01"},
|
|
]
|
|
selected = select_source_cycle_ids(rows, closed_limit=3)
|
|
self.assertIn(1, selected)
|
|
self.assertIn(2, selected)
|
|
self.assertIn(6, selected)
|
|
self.assertIn(5, selected)
|
|
self.assertIn(4, selected)
|
|
self.assertNotIn(3, selected)
|
|
self.assertIn(7, selected)
|