audrix/dashboard-cpsp
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update iot panel will now listen to push from panel

Browse Source
This commit is contained in:
Alberto-Audrix committed 2026-09-21 14:38:17 +07:00
1 parent 20d04acb3e
commit 520c4bcaad
10 files changed
+158 -754
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/apps/operations/tests_iot_sync.py
+49 -122
View File
@@ -1,5 +1,4 @@
from datetime import timedelta
from unittest.mock import patch
from django.test import Client, TestCase
from django.utils import timezone
@@ -7,42 +6,26 @@ 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": {},
}
},
}
]
SAMPLE_PUSH_PAYLOAD = {
"suhu": "30.0",
"kelembapan": "70.0",
"wind": "5.0",
"flow": "12.0",
"total": "0.00",
}
class SyncIotPanelsPerFlockTests(TestCase):
class IotPanelPushTests(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.owner = User.objects.create_user(user_name="owner", password="secret")
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_a = Flock.objects.create(
self.flock = 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="",
iot_api_url="http://192.168.1.100:1123/",
)
today = timezone.localdate()
Cycle.objects.create(
@@ -52,108 +35,52 @@ class SyncIotPanelsPerFlockTests(TestCase):
doc_in_weight=40,
doc_in_count=10000,
)
self.client = Client()
@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"},
def test_push_creates_panel(self):
response = self.client.post(
f"/api/v1/panel-iot/push/{self.flock.pk}/",
SAMPLE_PUSH_PAYLOAD,
content_type="application/json",
**self._csrf_headers(client),
)
self.assertEqual(response.status_code, 200, response.content)
data = response.json()
self.assertTrue(data["ok"])
self.assertIsNotNone(data["panel_id"])
panel = IotPanel.objects.get(pk=data["panel_id"])
self.assertEqual(panel.flock_id, self.flock.pk)
self.assertAlmostEqual(panel.average_temperature, 30.0)
self.assertAlmostEqual(panel.humidity, 70.0)
self.assertAlmostEqual(panel.wind_speed, 5.0)
self.assertAlmostEqual(panel.water_total, 12.0)
@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},
def test_push_divides_by_10_when_over_100(self):
response = self.client.post(
f"/api/v1/panel-iot/push/{self.flock.pk}/",
{"suhu": "300", "kelembapan": "700", "wind": "50", "flow": "12.0", "total": "0"},
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)
panel = IotPanel.objects.get(pk=response.json()["panel_id"])
self.assertAlmostEqual(panel.average_temperature, 30.0)
self.assertAlmostEqual(panel.humidity, 70.0)
self.assertAlmostEqual(panel.wind_speed, 50.0)
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},
def test_push_nonexistent_flock_returns_404(self):
response = self.client.post(
"/api/v1/panel-iot/push/99999/",
SAMPLE_PUSH_PAYLOAD,
content_type="application/json",
**self._csrf_headers(client),
)
self.assertEqual(response.status_code, 400, response.content)
self.assertEqual(response.status_code, 404)
def test_push_computes_experience_temperature(self):
response = self.client.post(
f"/api/v1/panel-iot/push/{self.flock.pk}/",
SAMPLE_PUSH_PAYLOAD,
content_type="application/json",
)
self.assertEqual(response.status_code, 200)
panel = IotPanel.objects.get(pk=response.json()["panel_id"])
self.assertIsNotNone(panel.experience_temperature)
self.assertIsInstance(panel.experience_temperature, float)