Suriyakumarvijayanayagam
dad04e8cda
Five components that ship as one product:
- behavision/ the recognition engine. RTSP ingest, YuNet detection, IoU
tracking, ArcFace embeddings, a FAISS/SQLite gallery, and a
FastAPI dashboard. Identity is decided once per TRACK from an
average of at least three embeddings, never per frame.
- agent/ the Go edge agent: supervises the engine, holds a durable
spool, and drains it to MQTT. Nothing is acked before the
broker confirms.
- desktop/ the shop PC application (Wails + React + tray).
- server/ the cloud API, MQTT consumer, reports and assistant.
- web/ platform.loyaly.ai, the head-office app, embedded in the
server binary.
The gallery stores 512-float embeddings and timestamps - no images unless
`app.store_faces` is switched on. Those embeddings are biometric personal
data under GDPR and India's DPDP: template inversion reconstructs a
recognisable face from an ArcFace vector, so data/behavision.db is treated
as a biometric database and DELETE /api/visitors/{id} is a real erasure.
CLAUDE.md carries the reasoning behind every non-obvious decision here,
including the ones that were measured and the ones that were wrong first.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HViLj9gYNRtSr7YVZmW5sn
111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
"""CameraStore — the writable camera list. No camera or models needed."""
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from behavision.cameras import CameraStore, protect, unprotect
|
|
from behavision.config import CameraConfig
|
|
|
|
SECRET = "p@ss:w0rd/with@specials"
|
|
|
|
|
|
def cam(cid="cam1", **kw):
|
|
base = dict(id=cid, host="192.168.0.138", path="/ch0_0.264",
|
|
username="admin", password=SECRET)
|
|
base.update(kw)
|
|
return CameraConfig(**base)
|
|
|
|
|
|
def test_add_get_list_delete(tmp_path):
|
|
s = CameraStore(tmp_path / "cameras.json")
|
|
s.add(cam())
|
|
assert [c.id for c in s.list()] == ["cam1"]
|
|
assert s.get("cam1").password == SECRET
|
|
assert s.delete("cam1") and s.get("cam1") is None
|
|
assert not s.delete("cam1")
|
|
|
|
|
|
def test_duplicate_id_rejected(tmp_path):
|
|
s = CameraStore(tmp_path / "cameras.json")
|
|
s.add(cam())
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
s.add(cam())
|
|
|
|
|
|
def test_camera_with_no_source_is_rejected_at_add(tmp_path):
|
|
"""Better to fail on Save than to silently create a camera that can never
|
|
connect."""
|
|
s = CameraStore(tmp_path / "cameras.json")
|
|
with pytest.raises(ValueError, match="set url, host or webcam"):
|
|
s.add(CameraConfig(id="bad"))
|
|
|
|
|
|
def test_survives_a_restart(tmp_path):
|
|
path = tmp_path / "cameras.json"
|
|
CameraStore(path).add(cam())
|
|
reloaded = CameraStore(path)
|
|
assert reloaded.get("cam1").password == SECRET
|
|
assert reloaded.get("cam1").host == "192.168.0.138"
|
|
|
|
|
|
def test_password_is_not_stored_in_the_clear_verbatim(tmp_path):
|
|
"""On Windows DPAPI encrypts it; elsewhere it is tagged plaintext. Either
|
|
way the stored form must be tagged so the format can evolve."""
|
|
path = tmp_path / "cameras.json"
|
|
CameraStore(path).add(cam())
|
|
stored = json.loads(path.read_text())["cameras"][0]["password"]
|
|
assert stored.startswith(("dpapi:", "plain:"))
|
|
assert unprotect(stored) == SECRET
|
|
|
|
|
|
def test_protect_roundtrip_including_empty():
|
|
assert unprotect(protect(SECRET)) == SECRET
|
|
assert protect("") == ""
|
|
assert unprotect("") == ""
|
|
|
|
|
|
def test_update_cannot_rename_the_id(tmp_path):
|
|
"""id is the engine's key for the running worker; a rename would orphan
|
|
it."""
|
|
s = CameraStore(tmp_path / "cameras.json")
|
|
s.add(cam())
|
|
updated = s.update("cam1", {"id": "renamed", "port": 8554})
|
|
assert updated.id == "cam1" and updated.port == 8554
|
|
assert s.get("renamed") is None
|
|
|
|
|
|
def test_update_unknown_camera_returns_none(tmp_path):
|
|
assert CameraStore(tmp_path / "cameras.json").update("nope", {}) is None
|
|
|
|
|
|
def test_seed_runs_once_only(tmp_path):
|
|
"""A camera deleted in the UI must not reappear from YAML on restart."""
|
|
path = tmp_path / "cameras.json"
|
|
s = CameraStore(path)
|
|
assert s.seed([cam("yaml1")]) is True
|
|
assert s.delete("yaml1")
|
|
assert CameraStore(path).seed([cam("yaml1")]) is False
|
|
assert CameraStore(path).list() == []
|
|
|
|
|
|
def test_unreadable_file_does_not_destroy_it(tmp_path):
|
|
path = tmp_path / "cameras.json"
|
|
path.write_text("{ not json")
|
|
s = CameraStore(path)
|
|
assert s.list() == []
|
|
assert path.read_text() == "{ not json" # left for a human to look at
|
|
|
|
|
|
def test_malformed_entry_is_skipped_not_fatal(tmp_path):
|
|
path = tmp_path / "cameras.json"
|
|
path.write_text(json.dumps({"cameras": [
|
|
{"id": "good", "host": "10.0.0.1"},
|
|
{"host": "10.0.0.2"}, # no id
|
|
]}))
|
|
assert [c.id for c in CameraStore(path).list()] == ["good"]
|
|
|
|
|
|
def test_safe_url_masks_the_password(tmp_path):
|
|
assert SECRET not in cam().safe_url()
|
|
assert "*****" in cam().safe_url()
|