feat: add management web console

tests/test_manage_api.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from cold_display_guard.config import load_config_document, merge_calibration, save_config_document
+from cold_display_guard.manage_api import ManageContext, build_summary
+
+
+class ManageApiTests(unittest.TestCase):
+    def test_merge_calibration_updates_zones_and_trash(self) -> None:
+        data = {
+            "camera_id": "cam",
+            "layout": {"rows": 2, "cols": 4, "zone_ids": ["r1c1"]},
+            "zones": [],
+        }
+
+        merged = merge_calibration(
+            data,
+            [{"id": "r1c1", "polygon": [[0, 0], [0.5, 0], [0.5, 0.5], [0, 0.5]]}],
+            [[0.8, 0.8], [1, 0.8], [1, 1], [0.8, 1]],
+        )
+
+        self.assertEqual(merged["layout"]["zone_ids"], ["r1c1"])
+        self.assertEqual(merged["zones"][0]["id"], "r1c1")
+        self.assertEqual(merged["trash"]["roi"][0], [0.8, 0.8])
+
+    def test_save_config_document_round_trips_manage_fields(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "config.toml"
+            save_config_document(
+                path,
+                {
+                    "camera_id": "cam",
+                    "timezone": "Asia/Shanghai",
+                    "stream": {"rtsp_url": "rtsp://example"},
+                    "thresholds": {"max_dwell_seconds": 30, "trash_confirmation_seconds": 5},
+                    "layout": {"rows": 1, "cols": 1, "zone_ids": ["r1c1"]},
+                    "zones": [{"id": "r1c1", "polygon": [[0, 0], [1, 0], [1, 1]]}],
+                    "trash": {"roi": [[0, 0], [1, 0], [1, 1]]},
+                    "event_sink": {"path": "logs/events.jsonl"},
+                },
+            )
+            loaded = load_config_document(path)
+
+        self.assertEqual(loaded["stream"]["rtsp_url"], "rtsp://example")
+        self.assertEqual(loaded["zones"][0]["polygon"][1], [1.0, 0.0])
+
+    def test_summary_reads_event_jsonl(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            config_path = root / "config" / "local.toml"
+            save_config_document(
+                config_path,
+                {
+                    "event_sink": {"path": "logs/events.jsonl"},
+                    "layout": {"rows": 1, "cols": 1, "zone_ids": ["r1c1"]},
+                },
+            )
+            events_path = root / "logs" / "events.jsonl"
+            events_path.parent.mkdir()
+            events_path.write_text(
+                "\n".join(
+                    [
+                        json.dumps({"event": "batch_started", "ts": "2026-04-27T10:00:00+08:00"}),
+                        json.dumps({"event": "missing_disposal_violation", "ts": "2026-04-27T13:02:00+08:00"}),
+                    ]
+                ),
+                encoding="utf-8",
+            )
+
+            summary = build_summary(ManageContext(config_path=config_path, project_root=root))
+
+        self.assertEqual(summary["metrics"]["event_count"], 2)
+        self.assertEqual(summary["metrics"]["violation_count"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()