5a438fdb32
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>
58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
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
|