feat: phase 2 content model + cache — SQLite schema, markdown, TTL
Stand up the full SQLite content layer: all 7 tables from the authoritative schema with WAL + foreign-keys enforced per-connection, entity dataclasses plus row mappers, hand-rolled versioned migrations tracked in schema_migrations, and an idempotent Python seed (system user + welcome post + About page). Add a Markdown->HTML service using markdown-it-py with a strict bleach allowlist (tables intentionally omitted on both sides). Add a typed in-process TTLCache[K,V] and wire it into real DB-backed PostService and PageService, both exposing invalidate_all() for Phase 4 admin writes. Rewire / and /about to read from the DB; homepage renders the seeded welcome post, About renders page.title + sanitized body_html_cached. Update the Phase 1 route tests accordingly. Mark Phase 2 complete in docs/ROADMAP.md.
This commit is contained in:
155
tests/test_db_migrations.py
Normal file
155
tests/test_db_migrations.py
Normal file
@@ -0,0 +1,155 @@
|
||||
"""Tests for the DB migration runner and per-connection PRAGMAs.
|
||||
|
||||
Goals:
|
||||
|
||||
- ``run_migrations`` applies every file once and is a no-op on re-run.
|
||||
- Tables and indexes declared in ``001_init.sql`` exist after a fresh
|
||||
migration.
|
||||
- The ``schema_migrations`` tracker records the applied version.
|
||||
- ``journal_mode=WAL`` and ``foreign_keys=ON`` hold on *every* new
|
||||
connection (the whole point of the ``@event.listens_for`` hook).
|
||||
|
||||
Uses a function-scoped temp SQLite file so each test is hermetic.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import Engine, text
|
||||
|
||||
from app.db import build_engine, run_migrations
|
||||
|
||||
|
||||
def _fresh_engine(tmp_path: Path) -> Engine:
|
||||
"""Build a fresh engine on a temp-file SQLite DB.
|
||||
|
||||
Helper kept local to this test module — conftest provides
|
||||
higher-level fixtures, but for migration tests we want direct
|
||||
control over migration order and re-runs.
|
||||
"""
|
||||
return build_engine(f"sqlite:///{tmp_path / 'mig.db'}")
|
||||
|
||||
|
||||
def test_first_migration_creates_all_tables(tmp_path: Path) -> None:
|
||||
"""Running migrations on a fresh DB creates every authoritative table.
|
||||
|
||||
We check for each of the 7 domain tables + ``schema_migrations``;
|
||||
absence of any one would indicate the SQL file got truncated.
|
||||
"""
|
||||
engine = _fresh_engine(tmp_path)
|
||||
applied = run_migrations(engine)
|
||||
|
||||
assert applied == ["001_init"], applied
|
||||
|
||||
expected_tables = {
|
||||
"users",
|
||||
"magic_link_tokens",
|
||||
"sessions",
|
||||
"pages",
|
||||
"posts",
|
||||
"media",
|
||||
"contact_submissions",
|
||||
"auth_events",
|
||||
"schema_migrations",
|
||||
}
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
text("SELECT name FROM sqlite_master WHERE type='table'")
|
||||
).fetchall()
|
||||
found = {r[0] for r in rows}
|
||||
missing = expected_tables - found
|
||||
assert not missing, f"migration did not create: {missing}"
|
||||
|
||||
|
||||
def test_required_indexes_exist(tmp_path: Path) -> None:
|
||||
"""The three named indexes from the ROADMAP schema are present."""
|
||||
engine = _fresh_engine(tmp_path)
|
||||
run_migrations(engine)
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
text("SELECT name FROM sqlite_master WHERE type='index'")
|
||||
).fetchall()
|
||||
names = {r[0] for r in rows}
|
||||
for required in (
|
||||
"idx_magic_email_created",
|
||||
"idx_posts_status_pub",
|
||||
"idx_auth_events_created",
|
||||
):
|
||||
assert required in names, f"missing index: {required} (have {names})"
|
||||
|
||||
|
||||
def test_migrations_are_idempotent(tmp_path: Path) -> None:
|
||||
"""Re-running migrations on an already-migrated DB is a no-op."""
|
||||
engine = _fresh_engine(tmp_path)
|
||||
first = run_migrations(engine)
|
||||
second = run_migrations(engine)
|
||||
|
||||
assert first == ["001_init"]
|
||||
assert second == [], "re-run should not re-apply migrations"
|
||||
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
text("SELECT version FROM schema_migrations")
|
||||
).fetchall()
|
||||
versions = [r[0] for r in rows]
|
||||
# Exactly one tracker row for the initial migration. Seed is not
|
||||
# run in this test, so seed_001 should NOT be present here.
|
||||
assert versions == ["001_init"], versions
|
||||
|
||||
|
||||
def test_posts_status_check_constraint_rejects_invalid(tmp_path: Path) -> None:
|
||||
"""The CHECK constraint on ``posts.status`` rejects unknown values.
|
||||
|
||||
This is a direct smoke test of the authoritative schema — if the
|
||||
migration accidentally drops the CHECK clause, this test fails.
|
||||
"""
|
||||
import sqlalchemy.exc
|
||||
|
||||
engine = _fresh_engine(tmp_path)
|
||||
run_migrations(engine)
|
||||
|
||||
# Seed a minimal user so the FK is satisfied.
|
||||
with engine.connect() as conn:
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO users (id, email, display_name, created_at,"
|
||||
" active) VALUES (1, 'x@x', 'x', '2026-01-01T00:00:00+00:00', 1)"
|
||||
)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
with engine.connect() as conn:
|
||||
try:
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO posts (slug, title, body_md, body_html_cached,"
|
||||
" status, updated_at, author_user_id)"
|
||||
" VALUES ('x', 't', 'm', 'h', 'bogus',"
|
||||
" '2026-01-01T00:00:00+00:00', 1)"
|
||||
)
|
||||
)
|
||||
conn.commit()
|
||||
raised = False
|
||||
except sqlalchemy.exc.IntegrityError:
|
||||
raised = True
|
||||
|
||||
assert raised, "CHECK constraint should have rejected status='bogus'"
|
||||
|
||||
|
||||
def test_pragmas_apply_on_every_connection(tmp_path: Path) -> None:
|
||||
"""Both PRAGMAs are active on every new connection from the pool.
|
||||
|
||||
Issues two separate connect calls; both must show WAL +
|
||||
foreign_keys=1. This is the regression guard for the
|
||||
``@event.listens_for(Engine, "connect")`` contract.
|
||||
"""
|
||||
engine = _fresh_engine(tmp_path)
|
||||
run_migrations(engine)
|
||||
|
||||
for _ in range(2):
|
||||
with engine.connect() as conn:
|
||||
jm = conn.execute(text("PRAGMA journal_mode")).scalar_one()
|
||||
fk = conn.execute(text("PRAGMA foreign_keys")).scalar_one()
|
||||
assert str(jm).lower() == "wal", f"expected WAL, got {jm!r}"
|
||||
assert int(fk) == 1, f"expected foreign_keys=1, got {fk!r}"
|
||||
Reference in New Issue
Block a user