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

feat: enrich webhook payloads with downstream event table fields

Browse Source 
Add missing fields (event_code, camera_ip, started_at, ended_at,
dwell_seconds, is_discarded, alerted_at, etc.) to both batch_event
and case_event payloads. Introduce source_id config for payload
injection and infer_camera_ip to extract IP from RTSP stream URL.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
skye.yue
2026-06-10 17:04:58 +08:00
parent e919ffd561
commit 45e2cf70f7
8 changed files with 193 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/test_config.py
View File

@@ -138,6 +138,7 @@ zone_ids = ["1", "2", "3"]
"enabled": True,
"event_url": "https://example.com/events",
"case_url": "https://example.com/cases",
"source_id": "cold-display-guard",
"callback_token": "secret",
"connect_timeout_seconds": 3,
"read_timeout_seconds": 5,
@@ -159,6 +160,7 @@ zone_ids = ["1", "2", "3"]
self.assertIn("[webhooks]", text)
self.assertIn('event_url = "https://example.com/events"', text)
self.assertIn('case_url = "https://example.com/cases"', text)
self.assertIn('source_id = "cold-display-guard"', text)
self.assertIn('callback_token = "secret"', text)
self.assertIn("retry_max_attempts = 4", text)
self.assertIn("retry_backoff_seconds = 30", text)

58
tests/test_webhooks.py
View File

@@ -28,6 +28,7 @@ class WebhookTests(unittest.TestCase):
"enabled": True,
"event_url": "https://example.com/events",
"case_url": "https://example.com/cases",
"source_id": "cold-display-guard",
"callback_token": "secret",
"connect_timeout_seconds": 4,
"read_timeout_seconds": 6,
@@ -42,6 +43,7 @@ class WebhookTests(unittest.TestCase):
self.assertTrue(settings.enabled)
self.assertEqual(settings.event_url, "https://example.com/events")
self.assertEqual(settings.case_url, "https://example.com/cases")
self.assertEqual(settings.source_id, "cold-display-guard")
self.assertEqual(settings.callback_token, "secret")
self.assertEqual(settings.connect_timeout_seconds, 4)
self.assertEqual(settings.read_timeout_seconds, 6)
@@ -61,12 +63,22 @@ class WebhookTests(unittest.TestCase):
"zone_label": "区域 1",
"severity": "alarm",
"state": "alerted",
}
"started_at": datetime(2026, 6, 9, 8, 40, tzinfo=UTC).isoformat(),
"dwell_seconds": 1200,
"alerted_at": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
},
camera_ip="192.168.3.4",
)
self.assertEqual(payload["kind"], "batch_event")
self.assertEqual(payload["event"], "time_alarm")
self.assertEqual(payload["event_code"], "batch_000001")
self.assertEqual(payload["camera_ip"], "192.168.3.4")
self.assertEqual(payload["zone_label"], "区域 1")
self.assertEqual(payload["started_at"], datetime(2026, 6, 9, 8, 40, tzinfo=UTC).isoformat())
self.assertEqual(payload["dwell_seconds"], 1200)
self.assertFalse(payload["is_discarded"])
self.assertEqual(payload["alerted_at"], datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat())
def test_build_batch_event_payload_includes_uploaded_snapshot_path(self) -> None:
payload = build_batch_event_payload(
@@ -91,17 +103,41 @@ class WebhookTests(unittest.TestCase):
{
"case_id": "case_batch_000001",
"case_type": "warning_escalated",
"case_status": "open",
"case_status": "handled",
"batch_id": "batch_000001",
"camera_id": "cam_01",
"zone_id": "1",
"zone_label": "区域 1",
"source_event": "warning_escalated",
"handled_source": "",
"created_at": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
"handled_at": datetime(2026, 6, 9, 9, 6, tzinfo=UTC).isoformat(),
"handled_source": "auto_closed",
"updated_at": datetime(2026, 6, 9, 9, 5, tzinfo=UTC).isoformat(),
}
"payload": {
"event": {
"started_at": datetime(2026, 6, 9, 8, 40, tzinfo=UTC).isoformat(),
"ended_at": datetime(2026, 6, 9, 9, 4, tzinfo=UTC).isoformat(),
"dwell_seconds": 1440,
"alerted_at": datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat(),
}
},
},
camera_ip="192.168.3.4",
)
self.assertEqual(payload["kind"], "case_event")
self.assertEqual(payload["action"], "updated")
self.assertEqual(payload["action"], "handled")
self.assertEqual(payload["case_id"], "case_batch_000001")
self.assertEqual(payload["event_code"], "batch_000001")
self.assertEqual(payload["camera_id"], "cam_01")
self.assertEqual(payload["camera_ip"], "192.168.3.4")
self.assertEqual(payload["zone_label"], "区域 1")
self.assertEqual(payload["started_at"], datetime(2026, 6, 9, 8, 40, tzinfo=UTC).isoformat())
self.assertEqual(payload["ended_at"], datetime(2026, 6, 9, 9, 4, tzinfo=UTC).isoformat())
self.assertEqual(payload["dwell_seconds"], 1440)
self.assertTrue(payload["is_discarded"])
self.assertEqual(payload["discarded_at"], datetime(2026, 6, 9, 9, 6, tzinfo=UTC).isoformat())
self.assertEqual(payload["alerted_at"], datetime(2026, 6, 9, 9, 0, tzinfo=UTC).isoformat())
def test_build_case_event_payload_includes_uploaded_snapshot_path(self) -> None:
payload = build_case_event_payload(
@@ -146,9 +182,11 @@ class WebhookTests(unittest.TestCase):
"webhooks": {
"enabled": True,
"event_url": "https://example.com/events",
"source_id": "cold-display-guard",
"connect_timeout_seconds": 4,
"read_timeout_seconds": 6,
}
},
"stream": {"rtsp_url": "rtsp://admin:secret@192.168.3.4:554/h264/ch1/main/av_stream"},
},
audit_path,
http_post=fake_post,
@@ -156,6 +194,8 @@ class WebhookTests(unittest.TestCase):
self.assertEqual(deliveries[0][0], "https://example.com/events")
self.assertEqual(deliveries[0][1]["kind"], "batch_event")
self.assertEqual(deliveries[0][1]["camera_ip"], "192.168.3.4")
self.assertEqual(deliveries[0][1]["source_id"], "cold-display-guard")
self.assertEqual(deliveries[0][2], (4.0, 6.0))
def test_send_case_webhooks_delivers_payload(self) -> None:
@@ -183,7 +223,9 @@ class WebhookTests(unittest.TestCase):
"webhooks": {
"enabled": True,
"case_url": "https://example.com/cases",
}
"source_id": "cold-display-guard",
},
"stream": {"rtsp_url": "rtsp://admin:secret@192.168.3.4:554/h264/ch1/main/av_stream"},
},
audit_path,
http_post=fake_post,
@@ -192,6 +234,8 @@ class WebhookTests(unittest.TestCase):
self.assertEqual(deliveries[0][0], "https://example.com/cases")
self.assertEqual(deliveries[0][1]["kind"], "case_event")
self.assertEqual(deliveries[0][1]["action"], "handled")
self.assertEqual(deliveries[0][1]["camera_ip"], "192.168.3.4")
self.assertEqual(deliveries[0][1]["source_id"], "cold-display-guard")
def test_failed_delivery_is_logged_without_raising(self) -> None:
def fake_post(url: str, payload: dict[str, object], timeout: tuple[float, float]) -> tuple[int, str]: