jeremy/frigate-notify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: scaffold v2 with FastAPI, SQLite database, and project structure

Browse Source 
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>
This commit is contained in:
Pluto
2026-03-07 17:02:17 -06:00
parent 6086ed3cc0
commit 031b9eaaec
10 changed files with 721 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
View File
tests/conftest.py
+16
View File
@@ -0,0 +1,16 @@
"""Shared fixtures for frigate-notify tests."""
import os
import sys
import tempfile
import pytest
# Add app/ to sys.path so test modules can import application code
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
@pytest.fixture
def db_path(tmp_path):
"""Provide a temporary database path for testing."""
return str(tmp_path / "test-frigate-notify.db")
tests/test_database.py
+321
View File
@@ -0,0 +1,321 @@
"""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"),
)