pluto
031b9eaaec
Migrate from raw HTTPServer to FastAPI, add SQLite database with schema for known objects, embeddings, events, and settings. Add centralized config, Pydantic models, and 22 database integration tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
322 lines
10 KiB
Python
322 lines
10 KiB
Python
"""Tests for the database module."""
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from database import close_connection, get_connection, init_db, set_db_path
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_db(db_path):
|
|
"""Initialize a fresh database for each test."""
|
|
set_db_path(db_path)
|
|
init_db()
|
|
yield
|
|
close_connection()
|
|
|
|
|
|
class TestTableCreation:
|
|
"""Tests for database schema initialization."""
|
|
|
|
def test_known_objects_table_exists(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='known_objects'"
|
|
)
|
|
assert cursor.fetchone() is not None
|
|
|
|
def test_embeddings_table_exists(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='embeddings'"
|
|
)
|
|
assert cursor.fetchone() is not None
|
|
|
|
def test_events_table_exists(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='events'"
|
|
)
|
|
assert cursor.fetchone() is not None
|
|
|
|
def test_settings_table_exists(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='settings'"
|
|
)
|
|
assert cursor.fetchone() is not None
|
|
|
|
def test_wal_mode_enabled(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute("PRAGMA journal_mode")
|
|
mode = cursor.fetchone()[0]
|
|
assert mode == "wal"
|
|
|
|
def test_foreign_keys_enabled(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute("PRAGMA foreign_keys")
|
|
assert cursor.fetchone()[0] == 1
|
|
|
|
def test_indexes_created(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='index' AND name LIKE 'idx_%'"
|
|
)
|
|
indexes = {row[0] for row in cursor.fetchall()}
|
|
expected = {
|
|
"idx_events_created_at",
|
|
"idx_events_match_status",
|
|
"idx_events_camera",
|
|
"idx_embeddings_known_object_id",
|
|
}
|
|
assert expected.issubset(indexes)
|
|
|
|
|
|
class TestDefaultSettings:
|
|
"""Tests for default settings insertion."""
|
|
|
|
def test_vision_backend_default(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute(
|
|
"SELECT value FROM settings WHERE key = 'vision_backend'"
|
|
)
|
|
assert cursor.fetchone()[0] == "haiku"
|
|
|
|
def test_recognition_enabled_default(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute(
|
|
"SELECT value FROM settings WHERE key = 'recognition_enabled'"
|
|
)
|
|
assert cursor.fetchone()[0] == "false"
|
|
|
|
def test_event_retention_days_default(self):
|
|
conn = get_connection()
|
|
cursor = conn.execute(
|
|
"SELECT value FROM settings WHERE key = 'event_retention_days'"
|
|
)
|
|
assert cursor.fetchone()[0] == "90"
|
|
|
|
def test_reinit_does_not_overwrite_settings(self):
|
|
"""Calling init_db() again should not reset changed settings."""
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"UPDATE settings SET value = 'ollama' WHERE key = 'vision_backend'"
|
|
)
|
|
conn.commit()
|
|
|
|
# Re-init
|
|
init_db()
|
|
|
|
cursor = conn.execute(
|
|
"SELECT value FROM settings WHERE key = 'vision_backend'"
|
|
)
|
|
assert cursor.fetchone()[0] == "ollama"
|
|
|
|
|
|
class TestKnownObjectsCRUD:
|
|
"""Tests for known_objects CRUD operations."""
|
|
|
|
def test_insert_known_object(self):
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"INSERT INTO known_objects (name, type, action) VALUES (?, ?, ?)",
|
|
("John", "person", "quiet"),
|
|
)
|
|
conn.commit()
|
|
|
|
cursor = conn.execute("SELECT * FROM known_objects WHERE name = 'John'")
|
|
row = cursor.fetchone()
|
|
assert row is not None
|
|
assert row["name"] == "John"
|
|
assert row["type"] == "person"
|
|
assert row["action"] == "quiet"
|
|
|
|
def test_update_known_object(self):
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"INSERT INTO known_objects (name, type, action) VALUES (?, ?, ?)",
|
|
("Rover", "pet", "quiet"),
|
|
)
|
|
conn.commit()
|
|
|
|
conn.execute(
|
|
"UPDATE known_objects SET action = 'suppress' WHERE name = 'Rover'"
|
|
)
|
|
conn.commit()
|
|
|
|
cursor = conn.execute(
|
|
"SELECT action FROM known_objects WHERE name = 'Rover'"
|
|
)
|
|
assert cursor.fetchone()[0] == "suppress"
|
|
|
|
def test_delete_known_object(self):
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"INSERT INTO known_objects (name, type, action) VALUES (?, ?, ?)",
|
|
("Delete Me", "person", "quiet"),
|
|
)
|
|
conn.commit()
|
|
|
|
conn.execute("DELETE FROM known_objects WHERE name = 'Delete Me'")
|
|
conn.commit()
|
|
|
|
cursor = conn.execute(
|
|
"SELECT * FROM known_objects WHERE name = 'Delete Me'"
|
|
)
|
|
assert cursor.fetchone() is None
|
|
|
|
def test_type_check_constraint(self):
|
|
conn = get_connection()
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"INSERT INTO known_objects (name, type, action) VALUES (?, ?, ?)",
|
|
("Bad", "alien", "quiet"),
|
|
)
|
|
|
|
def test_action_check_constraint(self):
|
|
conn = get_connection()
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"INSERT INTO known_objects (name, type, action) VALUES (?, ?, ?)",
|
|
("Bad", "person", "explode"),
|
|
)
|
|
|
|
|
|
class TestEventsCRUD:
|
|
"""Tests for events CRUD operations."""
|
|
|
|
def test_insert_event(self):
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"""INSERT INTO events (frigate_event_id, camera, label, score, ai_description, notified)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
("evt-001", "front_door", "person", 0.95, "A person at the door.", 1),
|
|
)
|
|
conn.commit()
|
|
|
|
cursor = conn.execute(
|
|
"SELECT * FROM events WHERE frigate_event_id = 'evt-001'"
|
|
)
|
|
row = cursor.fetchone()
|
|
assert row is not None
|
|
assert row["camera"] == "front_door"
|
|
assert row["label"] == "person"
|
|
assert row["score"] == 0.95
|
|
assert row["match_status"] == "unreviewed"
|
|
assert row["notified"] == 1
|
|
|
|
def test_frigate_event_id_unique(self):
|
|
conn = get_connection()
|
|
conn.execute(
|
|
"""INSERT INTO events (frigate_event_id, camera, label, score, notified)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
("evt-dup", "cam1", "car", 0.8, 1),
|
|
)
|
|
conn.commit()
|
|
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"""INSERT INTO events (frigate_event_id, camera, label, score, notified)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
("evt-dup", "cam2", "dog", 0.9, 0),
|
|
)
|
|
|
|
def test_match_status_check_constraint(self):
|
|
conn = get_connection()
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"""INSERT INTO events (frigate_event_id, camera, label, score, match_status, notified)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
("evt-bad", "cam1", "person", 0.8, "invalid_status", 1),
|
|
)
|
|
|
|
def test_matched_object_id_foreign_key_set_null(self):
|
|
"""Deleting a known object should set matched_object_id to NULL on events."""
|
|
conn = get_connection()
|
|
|
|
# Create known object
|
|
conn.execute(
|
|
"INSERT INTO known_objects (name, type, action) VALUES (?, ?, ?)",
|
|
("Jane", "person", "quiet"),
|
|
)
|
|
conn.commit()
|
|
obj_id = conn.execute(
|
|
"SELECT id FROM known_objects WHERE name = 'Jane'"
|
|
).fetchone()[0]
|
|
|
|
# Create event linked to the known object
|
|
conn.execute(
|
|
"""INSERT INTO events (frigate_event_id, camera, label, score, matched_object_id, match_status, notified)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
|
("evt-fk", "cam1", "person", 0.9, obj_id, "known", 1),
|
|
)
|
|
conn.commit()
|
|
|
|
# Delete the known object
|
|
conn.execute("DELETE FROM known_objects WHERE id = ?", (obj_id,))
|
|
conn.commit()
|
|
|
|
# Event should still exist with matched_object_id = NULL
|
|
cursor = conn.execute(
|
|
"SELECT matched_object_id FROM events WHERE frigate_event_id = 'evt-fk'"
|
|
)
|
|
row = cursor.fetchone()
|
|
assert row is not None
|
|
assert row["matched_object_id"] is None
|
|
|
|
|
|
class TestEmbeddingsCascade:
|
|
"""Tests for embeddings cascade delete."""
|
|
|
|
def test_cascade_delete_embeddings(self):
|
|
"""Deleting a known object should cascade delete its embeddings."""
|
|
conn = get_connection()
|
|
|
|
# Create known object
|
|
conn.execute(
|
|
"INSERT INTO known_objects (name, type, action) VALUES (?, ?, ?)",
|
|
("Bob", "person", "alert"),
|
|
)
|
|
conn.commit()
|
|
obj_id = conn.execute(
|
|
"SELECT id FROM known_objects WHERE name = 'Bob'"
|
|
).fetchone()[0]
|
|
|
|
# Create embeddings for that object
|
|
fake_embedding = b"\x00\x01\x02\x03"
|
|
conn.execute(
|
|
"INSERT INTO embeddings (known_object_id, embedding) VALUES (?, ?)",
|
|
(obj_id, fake_embedding),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO embeddings (known_object_id, embedding) VALUES (?, ?)",
|
|
(obj_id, fake_embedding),
|
|
)
|
|
conn.commit()
|
|
|
|
# Verify embeddings exist
|
|
cursor = conn.execute(
|
|
"SELECT COUNT(*) FROM embeddings WHERE known_object_id = ?", (obj_id,)
|
|
)
|
|
assert cursor.fetchone()[0] == 2
|
|
|
|
# Delete the known object
|
|
conn.execute("DELETE FROM known_objects WHERE id = ?", (obj_id,))
|
|
conn.commit()
|
|
|
|
# Embeddings should be gone
|
|
cursor = conn.execute(
|
|
"SELECT COUNT(*) FROM embeddings WHERE known_object_id = ?", (obj_id,)
|
|
)
|
|
assert cursor.fetchone()[0] == 0
|
|
|
|
def test_embedding_foreign_key_constraint(self):
|
|
"""Inserting an embedding with a non-existent known_object_id should fail."""
|
|
conn = get_connection()
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"INSERT INTO embeddings (known_object_id, embedding) VALUES (?, ?)",
|
|
(99999, b"\x00"),
|
|
)
|