160 lines
6.1 KiB
Python
160 lines
6.1 KiB
Python
from datetime import timedelta
|
|
from unittest.mock import patch
|
|
|
|
from django.test import Client, TestCase
|
|
from django.utils import timezone
|
|
|
|
from apps.accounts.models import User
|
|
from apps.farms.models import Cycle, Flock, Kandang, Site
|
|
from apps.operations.models import IotPanel
|
|
from apps.operations.services.iot_api import IotApiClient, sync_iot_panels
|
|
|
|
|
|
SAMPLE_PAYLOAD = {
|
|
"results": [
|
|
{
|
|
"flock_id": "other-external-id",
|
|
"fetched_at": timezone.now().isoformat().replace("+00:00", "Z"),
|
|
"payload": {
|
|
"data": {
|
|
"actualTemperature": 300,
|
|
"humidity": 700,
|
|
"wind": 5,
|
|
"water": 12,
|
|
"sensors": {},
|
|
}
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
class SyncIotPanelsPerFlockTests(TestCase):
|
|
def setUp(self):
|
|
owner = User.objects.create_user(user_name="owner", password="secret")
|
|
self.site = Site.objects.create(site_name="Site A", user=owner)
|
|
self.kandang = Kandang.objects.create(kandang_name="Kandang 1", site=self.site)
|
|
self.flock_a = Flock.objects.create(
|
|
flock_name="Lantai 1",
|
|
kandang=self.kandang,
|
|
iot_api_url="https://example.test/iot/a?token=1",
|
|
)
|
|
self.flock_b = Flock.objects.create(
|
|
flock_name="Lantai 2",
|
|
kandang=self.kandang,
|
|
iot_api_url="",
|
|
)
|
|
today = timezone.localdate()
|
|
Cycle.objects.create(
|
|
kandang=self.kandang,
|
|
start_date=today - timedelta(days=5),
|
|
end_date=today + timedelta(days=30),
|
|
doc_in_weight=40,
|
|
doc_in_count=10000,
|
|
)
|
|
|
|
@patch.object(IotApiClient, "fetch_page")
|
|
def test_upserts_on_flock_with_url_and_ignores_upstream_flock_id(self, mock_fetch):
|
|
mock_fetch.return_value = SAMPLE_PAYLOAD
|
|
summary = sync_iot_panels(max_pages=1)
|
|
self.assertEqual(summary.upserted, 1)
|
|
self.assertEqual(IotPanel.objects.filter(flock=self.flock_a).count(), 1)
|
|
self.assertEqual(IotPanel.objects.filter(flock=self.flock_b).count(), 0)
|
|
mock_fetch.assert_called()
|
|
|
|
@patch.object(IotApiClient, "fetch_page")
|
|
def test_blank_url_flock_never_fetched(self, mock_fetch):
|
|
mock_fetch.return_value = SAMPLE_PAYLOAD
|
|
summary = sync_iot_panels(flock_id=self.flock_b.pk, max_pages=1)
|
|
self.assertEqual(summary.upserted, 0)
|
|
self.assertEqual(summary.pages, 0)
|
|
mock_fetch.assert_not_called()
|
|
|
|
@patch.object(IotApiClient, "fetch_page")
|
|
def test_flock_id_filter_only_syncs_that_flock(self, mock_fetch):
|
|
mock_fetch.return_value = SAMPLE_PAYLOAD
|
|
flock_c = Flock.objects.create(
|
|
flock_name="Lantai 3",
|
|
kandang=self.kandang,
|
|
iot_api_url="https://example.test/iot/c",
|
|
)
|
|
summary = sync_iot_panels(flock_id=flock_c.pk, max_pages=1)
|
|
self.assertEqual(summary.upserted, 1)
|
|
self.assertEqual(IotPanel.objects.filter(flock=flock_c).count(), 1)
|
|
self.assertEqual(IotPanel.objects.filter(flock=self.flock_a).count(), 0)
|
|
|
|
def test_fetch_page_merges_existing_query_string(self):
|
|
client = IotApiClient("https://example.test/iot/a?token=1")
|
|
with patch("apps.operations.services.iot_api.httpx.Client") as mock_client_cls:
|
|
mock_client = mock_client_cls.return_value.__enter__.return_value
|
|
mock_response = mock_client.get.return_value
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"results": []}
|
|
client.fetch_page(2)
|
|
called_url = mock_client.get.call_args.args[0]
|
|
self.assertIn("token=1", called_url)
|
|
self.assertIn("page=2", called_url)
|
|
|
|
|
|
class PanelIotSyncActionTests(TestCase):
|
|
def setUp(self):
|
|
self.owner = User.objects.create_user(user_name="owner", password="secret")
|
|
self.staff = User.objects.create_user(
|
|
user_name="staff",
|
|
password="staff123",
|
|
status=User.STATUS_ACTIVE,
|
|
is_staff=True,
|
|
)
|
|
self.site = Site.objects.create(site_name="Site A", user=self.owner)
|
|
self.kandang = Kandang.objects.create(kandang_name="Kandang 1", site=self.site)
|
|
self.flock = Flock.objects.create(
|
|
flock_name="Lantai 1",
|
|
kandang=self.kandang,
|
|
iot_api_url="https://example.test/iot/a",
|
|
)
|
|
self.blank = Flock.objects.create(
|
|
flock_name="Lantai 2",
|
|
kandang=self.kandang,
|
|
iot_api_url="",
|
|
)
|
|
|
|
def _csrf_headers(self, client: Client) -> dict[str, str]:
|
|
token = client.cookies["csrftoken"].value
|
|
return {"HTTP_X_CSRFTOKEN": token}
|
|
|
|
def _login(self, client: Client) -> None:
|
|
client.get("/api/v1/auth/csrf/")
|
|
response = client.post(
|
|
"/api/v1/auth/login/",
|
|
{"user_name": "staff", "password": "staff123"},
|
|
content_type="application/json",
|
|
**self._csrf_headers(client),
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.content)
|
|
|
|
@patch("apps.operations.views.sync_iot_panels")
|
|
def test_sync_action_ok_for_staff(self, mock_sync):
|
|
mock_sync.return_value = type("S", (), {"upserted": 2, "skipped": 0, "pages": 1})()
|
|
client = Client(enforce_csrf_checks=True)
|
|
self._login(client)
|
|
response = client.post(
|
|
"/api/v1/panel-iot/sync/",
|
|
{"flock_id": self.flock.pk},
|
|
content_type="application/json",
|
|
**self._csrf_headers(client),
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.content)
|
|
self.assertEqual(response.json()["upserted"], 2)
|
|
mock_sync.assert_called_once_with(flock_id=self.flock.pk)
|
|
|
|
def test_sync_action_rejects_blank_url(self):
|
|
client = Client(enforce_csrf_checks=True)
|
|
self._login(client)
|
|
response = client.post(
|
|
"/api/v1/panel-iot/sync/",
|
|
{"flock_id": self.blank.pk},
|
|
content_type="application/json",
|
|
**self._csrf_headers(client),
|
|
)
|
|
self.assertEqual(response.status_code, 400, response.content)
|