yangyl/cold_display_guard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: upload alarm snapshots to webhook payloads

Browse Source
This commit is contained in:
skye.yue
2026-06-09 13:01:15 +08:00
parent 523f928303
commit 04729a0fd1
14 changed files with 853 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
tests/test_webhooks.py
View File

@@ -68,6 +68,24 @@ class WebhookTests(unittest.TestCase):
self.assertEqual(payload["event"], "time_alarm")
self.assertEqual(payload["zone_label"], "区域 1")
def test_build_batch_event_payload_includes_uploaded_snapshot_path(self) -> None:
payload = build_batch_event_payload(
{
"event": "time_alarm",
"ts": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
"batch_id": "batch_000001",
"camera_id": "cam_01",
"zone_id": "1",
"zone_label": "区域 1",
"severity": "alarm",
"state": "alerted",
},
snapshot_upload={"status": "uploaded", "object_key": "uploads/alarms/a.jpg", "batch_ids": ["batch_000001"]},
)
self.assertEqual(payload["snapshot_upload_status"], "uploaded")
self.assertEqual(payload["snapshot_object_key"], "uploads/alarms/a.jpg")
def test_build_case_event_payload_wraps_case_snapshot(self) -> None:
payload = build_case_event_payload(
{
@@ -85,6 +103,23 @@ class WebhookTests(unittest.TestCase):
self.assertEqual(payload["action"], "updated")
self.assertEqual(payload["case_id"], "case_batch_000001")
def test_build_case_event_payload_includes_uploaded_snapshot_path(self) -> None:
payload = build_case_event_payload(
{
"case_id": "case_batch_000001",
"case_type": "warning_escalated",
"case_status": "open",
"batch_id": "batch_000001",
"source_event": "warning_escalated",
"handled_source": "",
"updated_at": datetime(2026, 6, 9, 9, 5, tzinfo=UTC).isoformat(),
},
snapshot_upload={"status": "uploaded", "object_key": "uploads/alarms/a.jpg", "batch_ids": ["batch_000001"]},
)
self.assertEqual(payload["snapshot_upload_status"], "uploaded")
self.assertEqual(payload["snapshot_object_key"], "uploads/alarms/a.jpg")
def test_send_batch_event_webhooks_delivers_payload(self) -> None:
deliveries: list[tuple[str, dict[str, object], tuple[float, float]]] = []