Peter Woolery
910508194a
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 <noreply@anthropic.com>
92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
# 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"
|