From 910508194a845527f6eb31bb45b44df274bc5403 Mon Sep 17 00:00:00 2001 From: Peter Woolery Date: Tue, 14 Apr 2026 07:01:40 -0700 Subject: [PATCH] feat: camera batch endpoint implementation and tests Self-contained server stub and pytest tests for the /api/v1/camera/events/batch endpoint, mirroring the BLE batch pattern with idempotent INSERT on (device_id, period_start). Co-Authored-By: Claude Sonnet 4.6 --- server/__init__.py | 0 server/camera_endpoint.py | 69 ++++++++++++++++++++++++++ server/test_camera_endpoint.py | 91 ++++++++++++++++++++++++++++++++++ 3 files changed, 160 insertions(+) create mode 100644 server/__init__.py create mode 100644 server/camera_endpoint.py create mode 100644 server/test_camera_endpoint.py diff --git a/server/__init__.py b/server/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/camera_endpoint.py b/server/camera_endpoint.py new file mode 100644 index 0000000..d5ebe62 --- /dev/null +++ b/server/camera_endpoint.py @@ -0,0 +1,69 @@ +# server/camera_endpoint.py +# Add these models and endpoint to the server's main.py alongside the existing BLE endpoints. +# Requires: camera_records table (see migrations/004_camera_records.sql) +# +# IMPORTANT: Before deploying, verify the HMAC message format in verify_device_hmac +# matches what the firmware computes: +# HMAC-SHA256(secret, f"{device_id}:{timestamp}:{sha256_hex(body)}") +# Headers expected: X-Device-Id, X-Timestamp, X-HMAC-Signature + +import sqlite3 +from typing import List + +from fastapi import Depends +from pydantic import BaseModel + + +class CameraRecord(BaseModel): + period_start: int + period_end: int + entries: int + exits: int + + +class CameraEventsRequest(BaseModel): + location_id: str + records: List[CameraRecord] + + +class CameraEventsResponse(BaseModel): + status: str + accepted: int + + +# Add this endpoint to your FastAPI app (alongside receive_batch_events): +# +# @app.post("/api/v1/camera/events/batch", response_model=CameraEventsResponse) +# async def receive_camera_events( +# batch: CameraEventsRequest, +# device_id: str = Depends(verify_device_hmac), +# db: sqlite3.Connection = Depends(get_db), +# ): +def receive_camera_events_impl( + batch: CameraEventsRequest, + device_id: str, + db: sqlite3.Connection, +) -> CameraEventsResponse: + """Receive hourly camera entry/exit records; idempotent on (device_id, period_start).""" + cursor = db.cursor() + accepted = 0 + for record in batch.records: + try: + cursor.execute( + """INSERT INTO camera_records + (device_id, location_id, period_start, period_end, entries, exits) + VALUES (?, ?, ?, ?, ?, ?)""", + ( + device_id, + batch.location_id, + record.period_start, + record.period_end, + record.entries, + record.exits, + ), + ) + accepted += 1 + except sqlite3.IntegrityError: + pass # duplicate (device_id, period_start) — idempotent + db.commit() + return CameraEventsResponse(status="ok", accepted=accepted) diff --git a/server/test_camera_endpoint.py b/server/test_camera_endpoint.py new file mode 100644 index 0000000..159fbc7 --- /dev/null +++ b/server/test_camera_endpoint.py @@ -0,0 +1,91 @@ +# server/test_camera_endpoint.py +# Template tests for the camera batch endpoint. +# Adapt imports and fixtures to match the actual server's test structure. +# +# To run against the actual server (once integrated): +# pytest server/test_camera_endpoint.py -v + +import json +import sqlite3 + +import pytest + +# These imports will need to match the actual server module structure: +# from main import app, get_db, verify_device_hmac +# from fastapi.testclient import TestClient + + +def make_camera_batch_body(location_id: str, period_start: int, + period_end: int, entries: int, exits: int) -> str: + return json.dumps({ + "location_id": location_id, + "records": [{ + "period_start": period_start, + "period_end": period_end, + "entries": entries, + "exits": exits, + }] + }) + + +def _make_db() -> sqlite3.Connection: + db = sqlite3.connect(":memory:") + db.execute(""" + CREATE TABLE camera_records ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + device_id TEXT NOT NULL, + location_id TEXT NOT NULL, + period_start INTEGER NOT NULL, + period_end INTEGER NOT NULL, + entries INTEGER NOT NULL, + exits INTEGER NOT NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + UNIQUE(device_id, period_start) + ) + """) + db.commit() + return db + + +def test_insert_logic_idempotent(): + """Unit test for the insert logic — no FastAPI needed.""" + db = _make_db() + + from server.camera_endpoint import CameraRecord, CameraEventsRequest, receive_camera_events_impl + + batch = CameraEventsRequest( + location_id="test-loc", + records=[CameraRecord(period_start=1712000000, period_end=1712003600, + entries=5, exits=3)] + ) + + resp1 = receive_camera_events_impl(batch, "dc-test-01", db) + assert resp1.status == "ok" + assert resp1.accepted == 1 + + # Second identical call — idempotent + resp2 = receive_camera_events_impl(batch, "dc-test-01", db) + assert resp2.status == "ok" + assert resp2.accepted == 0 + + +def test_entries_exits_stored_correctly(): + """Verify entries and exits are stored as submitted.""" + db = _make_db() + + from server.camera_endpoint import CameraRecord, CameraEventsRequest, receive_camera_events_impl + + batch = CameraEventsRequest( + location_id="retailer-123", + records=[CameraRecord(period_start=1712007200, period_end=1712010800, + entries=42, exits=39)] + ) + receive_camera_events_impl(batch, "dc-0042", db) + + row = db.execute( + "SELECT entries, exits, location_id FROM camera_records WHERE device_id=?", + ("dc-0042",) + ).fetchone() + assert row[0] == 42 + assert row[1] == 39 + assert row[2] == "retailer-123"