feat: add database schema with WAL mode and FTS5 triggers
SQLite database layer with thread-local connections, WAL journal mode, foreign key enforcement, and FTS5 full-text search on utterances via content-table triggers. TDD: 5 tests written first, all passing. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,39 @@
|
||||
import pytest
|
||||
import os
|
||||
from app.shared.config import Settings
|
||||
from app.shared.database import init_schema, get_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings(tmp_path):
|
||||
db_file = str(tmp_path / "test.db")
|
||||
clips_dir = str(tmp_path / "clips")
|
||||
os.makedirs(clips_dir, exist_ok=True)
|
||||
s = Settings(
|
||||
db_path=db_file,
|
||||
clips_dir=clips_dir,
|
||||
speaches_url="http://localhost:9999",
|
||||
ollama_url="http://localhost:9999",
|
||||
livekit_api_key="test",
|
||||
livekit_api_secret="test",
|
||||
)
|
||||
init_schema(s)
|
||||
return s
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db(settings):
|
||||
return get_db(settings)
|
||||
|
||||
|
||||
# Add to conftest.py
|
||||
from app.shared import database as _db_module
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_db_connection():
|
||||
"""Reset thread-local DB connection between tests."""
|
||||
_db_module._local.conn = None
|
||||
yield
|
||||
if hasattr(_db_module._local, 'conn') and _db_module._local.conn:
|
||||
_db_module._local.conn.close()
|
||||
_db_module._local.conn = None
|
||||
@@ -0,0 +1,41 @@
|
||||
def test_schema_creates_rooms_table(db):
|
||||
rows = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='rooms'").fetchall()
|
||||
assert len(rows) == 1
|
||||
|
||||
|
||||
def test_schema_creates_utterances_fts(db):
|
||||
rows = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='utterances_fts'").fetchall()
|
||||
assert len(rows) == 1
|
||||
|
||||
|
||||
def test_insert_and_fetch_room(db):
|
||||
db.execute("INSERT INTO rooms (name) VALUES (?)", ("Kitchen",))
|
||||
db.commit()
|
||||
row = db.execute("SELECT name FROM rooms WHERE name='Kitchen'").fetchone()
|
||||
assert row["name"] == "Kitchen"
|
||||
|
||||
|
||||
def test_insert_speaker_and_embedding(db):
|
||||
db.execute("INSERT INTO speakers (name) VALUES (?)", ("Jeremy",))
|
||||
db.commit()
|
||||
speaker = db.execute("SELECT id FROM speakers WHERE name='Jeremy'").fetchone()
|
||||
db.execute(
|
||||
"INSERT INTO voice_embeddings (speaker_id, embedding) VALUES (?, ?)",
|
||||
(speaker["id"], b"\x00\x01\x02\x03"),
|
||||
)
|
||||
db.commit()
|
||||
emb = db.execute("SELECT speaker_id FROM voice_embeddings").fetchone()
|
||||
assert emb["speaker_id"] == speaker["id"]
|
||||
|
||||
|
||||
def test_fts_trigger_on_utterance_insert(db):
|
||||
db.execute("INSERT INTO rooms (name) VALUES (?)", ("Living Room",))
|
||||
db.commit()
|
||||
room = db.execute("SELECT id FROM rooms WHERE name='Living Room'").fetchone()
|
||||
db.execute(
|
||||
"INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)",
|
||||
(room["id"], "hello world test", "unreviewed", "2026-01-01 10:00:00", "2026-01-01 10:00:03"),
|
||||
)
|
||||
db.commit()
|
||||
results = db.execute("SELECT rowid FROM utterances_fts WHERE transcript MATCH 'hello'").fetchall()
|
||||
assert len(results) == 1
|
||||
Reference in New Issue
Block a user