feat(S3): face profile, embedding, and visit CRUD queries
Add create/get/update/delete_cascade for face_profiles, insert/get for face_embeddings, and insert/get/get_active for visits. All seven query functions covered by unit tests (327 passing). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
57
tests/unit/test_visitor_queries.py
Normal file
57
tests/unit/test_visitor_queries.py
Normal file
@@ -0,0 +1,57 @@
|
||||
import time
|
||||
import pytest
|
||||
from vigilar.storage.queries import (
|
||||
create_face_profile, get_face_profile, get_all_profiles, update_face_profile,
|
||||
delete_face_profile_cascade, insert_face_embedding, get_embeddings_for_profile,
|
||||
insert_visit, get_visits, get_active_visits,
|
||||
)
|
||||
|
||||
|
||||
def test_create_and_get_profile(test_db):
|
||||
now = time.time()
|
||||
pid = create_face_profile(test_db, first_seen_at=now, last_seen_at=now)
|
||||
assert pid > 0
|
||||
profile = get_face_profile(test_db, pid)
|
||||
assert profile is not None
|
||||
assert profile["name"] is None
|
||||
|
||||
|
||||
def test_get_all_profiles(test_db):
|
||||
now = time.time()
|
||||
create_face_profile(test_db, first_seen_at=now, last_seen_at=now)
|
||||
create_face_profile(test_db, name="Bob", first_seen_at=now, last_seen_at=now)
|
||||
assert len(get_all_profiles(test_db)) == 2
|
||||
|
||||
|
||||
def test_update_profile(test_db):
|
||||
pid = create_face_profile(test_db, first_seen_at=0, last_seen_at=0)
|
||||
update_face_profile(test_db, pid, name="Alice")
|
||||
assert get_face_profile(test_db, pid)["name"] == "Alice"
|
||||
|
||||
|
||||
def test_insert_embedding(test_db):
|
||||
pid = create_face_profile(test_db, first_seen_at=0, last_seen_at=0)
|
||||
eid = insert_face_embedding(test_db, pid, "AAAA", "front", time.time())
|
||||
assert eid > 0
|
||||
assert len(get_embeddings_for_profile(test_db, pid)) == 1
|
||||
|
||||
|
||||
def test_insert_and_get_visit(test_db):
|
||||
pid = create_face_profile(test_db, first_seen_at=0, last_seen_at=0)
|
||||
insert_visit(test_db, pid, "front", time.time())
|
||||
assert len(get_visits(test_db)) == 1
|
||||
|
||||
|
||||
def test_get_active_visits(test_db):
|
||||
pid = create_face_profile(test_db, first_seen_at=0, last_seen_at=0)
|
||||
insert_visit(test_db, pid, "front", time.time())
|
||||
assert len(get_active_visits(test_db)) == 1
|
||||
|
||||
|
||||
def test_delete_cascade(test_db):
|
||||
pid = create_face_profile(test_db, first_seen_at=0, last_seen_at=0)
|
||||
insert_face_embedding(test_db, pid, "AAAA", "front", time.time())
|
||||
insert_visit(test_db, pid, "front", time.time())
|
||||
delete_face_profile_cascade(test_db, pid)
|
||||
assert get_face_profile(test_db, pid) is None
|
||||
assert len(get_embeddings_for_profile(test_db, pid)) == 0
|
||||
Reference in New Issue
Block a user