chicken_babies_site/tests/test_db_migrations.py

"""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}"