431 lines
16 KiB
Python
431 lines
16 KiB
Python
"""Tests for Package 3.1 SQLite infrastructure — prompts.db + history.db.
|
||
|
||
Covers: connection factories, WAL mode, foreign keys, table creation, seed data, idempotency.
|
||
Uses tmp_path for isolated test databases — no real filesystem pollution.
|
||
"""
|
||
|
||
import sqlite3
|
||
|
||
import pytest
|
||
|
||
|
||
# ── Helpers ────────────────────────────────────────────────────────────────
|
||
|
||
DEFAULT_PROMPTS_DB = "/tmp/test_prompts.db"
|
||
DEFAULT_HISTORY_DB = "/tmp/test_history.db"
|
||
|
||
|
||
def _patch_settings(monkeypatch, prompts_path: str, history_path: str):
|
||
"""Patch Settings to use test-specific DB paths."""
|
||
monkeypatch.setenv("PROMPTS_DB_PATH", prompts_path)
|
||
monkeypatch.setenv("HISTORY_DB_PATH", history_path)
|
||
from app.core.config import get_settings
|
||
get_settings.cache_clear()
|
||
|
||
|
||
# ── Config Tests ───────────────────────────────────────────────────────────
|
||
|
||
def test_config_default_db_paths(monkeypatch):
|
||
"""New config fields should have sensible defaults."""
|
||
monkeypatch.delenv("PROMPTS_DB_PATH", raising=False)
|
||
monkeypatch.delenv("HISTORY_DB_PATH", raising=False)
|
||
|
||
from app.core.config import Settings
|
||
|
||
settings = Settings()
|
||
assert settings.prompts_db_path == "./data/prompts.db"
|
||
assert settings.history_db_path == "./data/history.db"
|
||
|
||
|
||
def test_config_db_paths_from_env(tmp_path, monkeypatch):
|
||
"""DB paths should be configurable via environment variables."""
|
||
prompts_path = str(tmp_path / "my_prompts.db")
|
||
history_path = str(tmp_path / "my_history.db")
|
||
monkeypatch.setenv("PROMPTS_DB_PATH", prompts_path)
|
||
monkeypatch.setenv("HISTORY_DB_PATH", history_path)
|
||
|
||
from app.core.config import Settings
|
||
|
||
settings = Settings()
|
||
assert settings.prompts_db_path == prompts_path
|
||
assert settings.history_db_path == history_path
|
||
|
||
|
||
# ── Connection Factory Tests ───────────────────────────────────────────────
|
||
|
||
def test_get_prompts_db_creates_file_and_dir(tmp_path, monkeypatch):
|
||
"""get_prompts_db() should create the DB file and any missing parent dirs."""
|
||
prompts_path = str(tmp_path / "subdir" / "prompts.db")
|
||
history_path = str(tmp_path / "subdir" / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db
|
||
|
||
conn = get_prompts_db()
|
||
assert conn is not None
|
||
import os
|
||
assert os.path.isfile(prompts_path)
|
||
conn.close()
|
||
|
||
|
||
def test_get_history_db_creates_file_and_dir(tmp_path, monkeypatch):
|
||
"""get_history_db() should create the DB file and any missing parent dirs."""
|
||
prompts_path = str(tmp_path / "subdir" / "prompts.db")
|
||
history_path = str(tmp_path / "subdir" / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_history_db
|
||
|
||
conn = get_history_db()
|
||
assert conn is not None
|
||
import os
|
||
assert os.path.isfile(history_path)
|
||
conn.close()
|
||
|
||
|
||
def test_wal_mode_enabled(tmp_path, monkeypatch):
|
||
"""WAL journal mode should be enabled for better concurrency."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db
|
||
|
||
conn = get_prompts_db()
|
||
result = conn.execute("PRAGMA journal_mode").fetchone()
|
||
assert result[0].upper() == "WAL"
|
||
conn.close()
|
||
|
||
|
||
def test_foreign_keys_enabled(tmp_path, monkeypatch):
|
||
"""Foreign key enforcement should be enabled."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db
|
||
|
||
conn = get_prompts_db()
|
||
result = conn.execute("PRAGMA foreign_keys").fetchone()
|
||
assert result[0] == 1
|
||
conn.close()
|
||
|
||
|
||
def test_row_factory_is_row(tmp_path, monkeypatch):
|
||
"""Row factory should be sqlite3.Row for dict-like access."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db
|
||
|
||
conn = get_prompts_db()
|
||
assert conn.row_factory is sqlite3.Row
|
||
conn.close()
|
||
|
||
|
||
# ── Table Creation Tests ───────────────────────────────────────────────────
|
||
|
||
def test_init_prompts_db_creates_tables(tmp_path, monkeypatch):
|
||
"""init_prompts_db() should create system_prompt_profiles + system_prompts tables."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db, init_prompts_db
|
||
|
||
conn = get_prompts_db()
|
||
init_prompts_db(conn)
|
||
|
||
tables = conn.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
||
).fetchall()
|
||
table_names = {t["name"] for t in tables}
|
||
assert "system_prompt_profiles" in table_names
|
||
assert "system_prompts" in table_names
|
||
conn.close()
|
||
|
||
|
||
def test_init_prompts_db_idempotent(tmp_path, monkeypatch):
|
||
"""Calling init_prompts_db() twice should not raise errors."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db, init_prompts_db
|
||
|
||
conn = get_prompts_db()
|
||
init_prompts_db(conn)
|
||
init_prompts_db(conn) # second call — must not raise
|
||
conn.close()
|
||
|
||
|
||
def test_init_history_db_creates_table_and_index(tmp_path, monkeypatch):
|
||
"""init_history_db() should create query_history table + created_at index."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_history_db, init_history_db
|
||
|
||
conn = get_history_db()
|
||
init_history_db(conn)
|
||
|
||
tables = conn.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
||
).fetchall()
|
||
table_names = {t["name"] for t in tables}
|
||
assert "query_history" in table_names
|
||
|
||
indexes = conn.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='index' ORDER BY name"
|
||
).fetchall()
|
||
index_names = {i["name"] for i in indexes}
|
||
assert "idx_query_history_created_at" in index_names
|
||
conn.close()
|
||
|
||
|
||
def test_init_history_db_idempotent(tmp_path, monkeypatch):
|
||
"""Calling init_history_db() twice should not raise errors."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_history_db, init_history_db
|
||
|
||
conn = get_history_db()
|
||
init_history_db(conn)
|
||
init_history_db(conn) # second call — must not raise
|
||
conn.close()
|
||
|
||
|
||
# ── Seed Data Tests ────────────────────────────────────────────────────────
|
||
|
||
def test_seed_default_profiles_creates_three_profiles(tmp_path, monkeypatch):
|
||
"""seed_default_profiles() should insert exactly 3 profiles: A, B, C."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db, init_prompts_db, seed_default_profiles
|
||
|
||
conn = get_prompts_db()
|
||
init_prompts_db(conn)
|
||
seed_default_profiles(conn)
|
||
|
||
rows = conn.execute(
|
||
"SELECT name, is_active FROM system_prompt_profiles ORDER BY id"
|
||
).fetchall()
|
||
assert len(rows) == 3
|
||
names = [r["name"] for r in rows]
|
||
assert names == ["A", "B", "C"]
|
||
conn.close()
|
||
|
||
|
||
def test_seed_default_profiles_A_is_active(tmp_path, monkeypatch):
|
||
"""Profile A should be active (is_active=1), B and C inactive (is_active=0)."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db, init_prompts_db, seed_default_profiles
|
||
|
||
conn = get_prompts_db()
|
||
init_prompts_db(conn)
|
||
seed_default_profiles(conn)
|
||
|
||
profile_a = conn.execute(
|
||
"SELECT name, is_active FROM system_prompt_profiles WHERE name='A'"
|
||
).fetchone()
|
||
profile_b = conn.execute(
|
||
"SELECT name, is_active FROM system_prompt_profiles WHERE name='B'"
|
||
).fetchone()
|
||
profile_c = conn.execute(
|
||
"SELECT name, is_active FROM system_prompt_profiles WHERE name='C'"
|
||
).fetchone()
|
||
|
||
assert profile_a["is_active"] == 1
|
||
assert profile_b["is_active"] == 0
|
||
assert profile_c["is_active"] == 0
|
||
conn.close()
|
||
|
||
|
||
def test_seed_default_profiles_creates_nine_prompts(tmp_path, monkeypatch):
|
||
"""seed_default_profiles() should insert 21 prompt rows (3 profiles × 7 steps)."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db, init_prompts_db, seed_default_profiles
|
||
|
||
conn = get_prompts_db()
|
||
init_prompts_db(conn)
|
||
seed_default_profiles(conn)
|
||
|
||
rows = conn.execute(
|
||
"""SELECT sp.step_name, spp.name AS profile_name
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
ORDER BY spp.name, sp.step_name"""
|
||
).fetchall()
|
||
assert len(rows) == 21
|
||
|
||
# Each profile should have all 7 steps
|
||
expected_steps = {
|
||
"decompose", "filter", "generate",
|
||
"generate_per_subq", "filter_intro", "filter_section", "filter_outro",
|
||
}
|
||
for profile in ("A", "B", "C"):
|
||
profile_rows = [r for r in rows if r["profile_name"] == profile]
|
||
steps = {r["step_name"] for r in profile_rows}
|
||
assert steps == expected_steps
|
||
conn.close()
|
||
|
||
|
||
def test_seed_default_profiles_contains_expected_templates(tmp_path, monkeypatch):
|
||
"""Templates should contain expected placeholders for each step."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db, init_prompts_db, seed_default_profiles
|
||
|
||
conn = get_prompts_db()
|
||
init_prompts_db(conn)
|
||
seed_default_profiles(conn)
|
||
|
||
# Decompose must have {question}
|
||
decompose_row = conn.execute(
|
||
"""SELECT sp.prompt_template
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
WHERE spp.name='A' AND sp.step_name='decompose'"""
|
||
).fetchone()
|
||
assert decompose_row is not None
|
||
assert "{question}" in decompose_row["prompt_template"]
|
||
|
||
# Filter must have {question} and {chunks}
|
||
filter_row = conn.execute(
|
||
"""SELECT sp.prompt_template
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
WHERE spp.name='A' AND sp.step_name='filter'"""
|
||
).fetchone()
|
||
assert filter_row is not None
|
||
assert "{question}" in filter_row["prompt_template"]
|
||
assert "{chunks}" in filter_row["prompt_template"]
|
||
|
||
# Generate must have {question} and {context}
|
||
generate_row = conn.execute(
|
||
"""SELECT sp.prompt_template
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
WHERE spp.name='A' AND sp.step_name='generate'"""
|
||
).fetchone()
|
||
assert generate_row is not None
|
||
assert "{question}" in generate_row["prompt_template"]
|
||
assert "{context}" in generate_row["prompt_template"]
|
||
|
||
# Generate per-subq must have {context_sections}
|
||
gen_per_subq_row = conn.execute(
|
||
"""SELECT sp.prompt_template
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
WHERE spp.name='A' AND sp.step_name='generate_per_subq'"""
|
||
).fetchone()
|
||
assert gen_per_subq_row is not None
|
||
assert "{context_sections}" in gen_per_subq_row["prompt_template"]
|
||
|
||
# Filter intro exists (no required placeholders)
|
||
filter_intro_row = conn.execute(
|
||
"""SELECT sp.prompt_template
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
WHERE spp.name='A' AND sp.step_name='filter_intro'"""
|
||
).fetchone()
|
||
assert filter_intro_row is not None
|
||
assert len(filter_intro_row["prompt_template"]) > 0
|
||
|
||
# Filter section must have per-subq placeholders
|
||
filter_section_row = conn.execute(
|
||
"""SELECT sp.prompt_template
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
WHERE spp.name='A' AND sp.step_name='filter_section'"""
|
||
).fetchone()
|
||
assert filter_section_row is not None
|
||
assert "{subq_idx}" in filter_section_row["prompt_template"]
|
||
assert "{subq_question}" in filter_section_row["prompt_template"]
|
||
assert "{chunks}" in filter_section_row["prompt_template"]
|
||
|
||
# Filter outro exists (no required placeholders)
|
||
filter_outro_row = conn.execute(
|
||
"""SELECT sp.prompt_template
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
WHERE spp.name='A' AND sp.step_name='filter_outro'"""
|
||
).fetchone()
|
||
assert filter_outro_row is not None
|
||
assert len(filter_outro_row["prompt_template"]) > 0
|
||
|
||
conn.close()
|
||
|
||
|
||
def test_seed_default_profiles_idempotent(tmp_path, monkeypatch):
|
||
"""Calling seed_default_profiles() twice should not duplicate rows."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db, init_prompts_db, seed_default_profiles
|
||
|
||
conn = get_prompts_db()
|
||
init_prompts_db(conn)
|
||
seed_default_profiles(conn)
|
||
seed_default_profiles(conn)
|
||
|
||
profile_count = conn.execute(
|
||
"SELECT COUNT(*) FROM system_prompt_profiles"
|
||
).fetchone()[0]
|
||
prompt_count = conn.execute(
|
||
"SELECT COUNT(*) FROM system_prompts"
|
||
).fetchone()[0]
|
||
|
||
assert profile_count == 3
|
||
assert prompt_count == 21
|
||
conn.close()
|
||
|
||
|
||
def test_all_three_profiles_have_identical_seed_templates(tmp_path, monkeypatch):
|
||
"""All 3 profiles should start with identical seed templates."""
|
||
prompts_path = str(tmp_path / "prompts.db")
|
||
history_path = str(tmp_path / "history.db")
|
||
_patch_settings(monkeypatch, prompts_path, history_path)
|
||
|
||
from app.core.sqlite_db import get_prompts_db, init_prompts_db, seed_default_profiles
|
||
|
||
conn = get_prompts_db()
|
||
init_prompts_db(conn)
|
||
seed_default_profiles(conn)
|
||
|
||
for step in (
|
||
"decompose", "filter", "generate",
|
||
"generate_per_subq", "filter_intro", "filter_section", "filter_outro",
|
||
):
|
||
rows = conn.execute(
|
||
"""SELECT spp.name, sp.prompt_template
|
||
FROM system_prompts sp
|
||
JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
|
||
WHERE sp.step_name = ?
|
||
ORDER BY spp.name""",
|
||
(step,),
|
||
).fetchall()
|
||
|
||
templates = [r["prompt_template"] for r in rows]
|
||
# All 3 must match
|
||
assert templates[0] == templates[1] == templates[2], (
|
||
f"Step '{step}' templates differ across profiles: {templates}"
|
||
)
|
||
|
||
conn.close()
|