test_migration_rls.py

"""Tests for the row-level-security migration, and for its staying complete.

See ../alembic/versions/b3d1c7a94e02_enable_row_level_security.py and
../../LAUNCH_CHECKLIST.md 5a.

Migrations are the one part of this backend nothing else exercises: the
suite builds its schema with `Base.metadata.create_all` (conftest.py), so
no test has ever run one. That is tolerable for `create_table` calls
Alembic autogenerated from the models, and not tolerable for a revision
whose whole job is a security property - hence `rls_statements` being a
pure function these can call directly.

The last test here is the one with a long future. Alembic creates plain
tables every time, so any model added later arrives unprotected - and the
revision that locked the first seven will long since have been written and
forgotten, which is precisely when nobody looks. It compares the models
against every migration and fails while the gap is still on someone's
screen.
"""

import importlib.util
from pathlib import Path

import pytest

from app.db.base import Base

VERSIONS_DIR = Path(__file__).resolve().parents[1] / "alembic" / "versions"

RLS_REVISION = "b3d1c7a94e02_enable_row_level_security"


def _load_revision(path: Path):
    """Import a migration module by path.

    Alembic loads these itself at runtime and they are not importable as
    ordinary modules (the versions directory is not a package), so this
    mirrors what Alembic does rather than adding an `__init__.py` purely to
    make a test tidier.
    """
    spec = importlib.util.spec_from_file_location(path.stem, path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


@pytest.fixture(scope="module")
def rls_migration():
    return _load_revision(VERSIONS_DIR / f"{RLS_REVISION}.py")


def test_postgres_gets_a_statement_for_every_table(rls_migration):
    statements = rls_migration.rls_statements("postgresql", enable=True)

    assert len(statements) == len(rls_migration.RLS_TABLES)
    for table in rls_migration.RLS_TABLES:
        assert f"ALTER TABLE public.{table} ENABLE ROW LEVEL SECURITY" in statements


@pytest.mark.parametrize("dialect", ["sqlite", "mysql", "duckdb"])
def test_a_non_postgres_dialect_gets_nothing(rls_migration, dialect):
    """The guard is an allow-list ('postgresql only'), not a deny-list.

    Every database this backend runs against is Postgres now - local dev
    included, see backend/scripts/local-postgres.sh - so nothing currently
    takes this branch. It is asserted anyway because `rls_statements` builds
    raw DDL for `op.execute`, which passes strings through to whichever
    dialect Alembic is pointed at: an allow-list fails safe on a dialect
    nobody anticipated, and this is the test that says so is deliberate.
    """
    assert rls_migration.rls_statements(dialect, enable=True) == []
    assert rls_migration.rls_statements(dialect, enable=False) == []


def test_downgrade_disables_what_upgrade_enabled(rls_migration):
    enabled = rls_migration.rls_statements("postgresql", enable=True)
    disabled = rls_migration.rls_statements("postgresql", enable=False)

    assert len(disabled) == len(enabled)
    assert all("DISABLE ROW LEVEL SECURITY" in statement for statement in disabled)


def test_never_forces_rls(rls_migration):
    """FORCE ROW LEVEL SECURITY would apply RLS to the table OWNER, which is
    exactly the role the backend connects as - it would break every endpoint
    at once, while looking like a tightening.

    Asserted rather than left to review because the two spellings differ by
    one word and the failure is total."""
    for enable in (True, False):
        for statement in rls_migration.rls_statements("postgresql", enable=enable):
            assert "FORCE" not in statement.upper()


def test_every_model_table_is_locked_by_some_migration():
    """The drift guard.

    A new model means a new table, and Alembic will create it plain. This
    fails until some revision names it, which is the only thing standing
    between "we enabled RLS once" and a table nobody remembered.
    """
    locked: set[str] = set()
    for path in sorted(VERSIONS_DIR.glob("*.py")):
        module = _load_revision(path)
        locked.update(getattr(module, "RLS_TABLES", ()))

    unlocked = sorted(set(Base.metadata.tables) - locked)

    assert not unlocked, (
        "These tables are created by a migration but no migration enables row level security on "
        "them, so Supabase's PostgREST would serve them to anyone holding the public anon key "
        "(LAUNCH_CHECKLIST.md 5a):\n  " + "\n  ".join(unlocked) + "\n\nAdd a revision that does, following "
        f"alembic/versions/{RLS_REVISION}.py."
    )


def test_the_guard_would_actually_fail(rls_migration):
    """Proves the test above can fail, rather than passing because both
    sides are empty or because getattr silently returned nothing."""
    assert len(rls_migration.RLS_TABLES) > 0
    assert set(Base.metadata.tables) == set(rls_migration.RLS_TABLES)


# --- e5b2f7c1a903, Alembic's own version table -----------------------------

VERSION_TABLE_REVISION = "e5b2f7c1a903_lock_alembics_own_version_table"


@pytest.fixture(scope="module")
def version_table_migration():
    return _load_revision(VERSIONS_DIR / f"{VERSION_TABLE_REVISION}.py")


def test_the_version_table_revision_names_whatever_alembic_calls_it(version_table_migration):
    """The table name is an argument, not a literal, and that is the point.

    Alembic's `version_table` and `version_table_schema` are configurable.
    A revision that hardcoded `public.alembic_version` would keep passing
    while locking nothing at all if either were ever set.
    """
    statements = version_table_migration.rls_statement("postgresql", "somewhere_else", "custom", enable=True)

    assert statements == ["ALTER TABLE custom.somewhere_else ENABLE ROW LEVEL SECURITY"]


def test_the_version_table_revision_falls_back_to_public(version_table_migration):
    statements = version_table_migration.rls_statement("postgresql", "alembic_version", None, enable=True)

    assert statements == ["ALTER TABLE alembic_version ENABLE ROW LEVEL SECURITY"]


@pytest.mark.parametrize("dialect", ["sqlite", "duckdb", "mysql"])
def test_the_version_table_revision_does_nothing_off_postgres(version_table_migration, dialect):
    assert version_table_migration.rls_statement(dialect, "alembic_version", "public", enable=True) == []


def test_the_version_table_revision_downgrade_undoes_its_upgrade(version_table_migration):
    up = version_table_migration.rls_statement("postgresql", "alembic_version", "public", enable=True)
    down = version_table_migration.rls_statement("postgresql", "alembic_version", "public", enable=False)

    assert up == ["ALTER TABLE public.alembic_version ENABLE ROW LEVEL SECURITY"]
    assert down == ["ALTER TABLE public.alembic_version DISABLE ROW LEVEL SECURITY"]


def test_the_version_table_revision_never_forces_rls(version_table_migration):
    """Sharper here than in b3d1c7a94e02: forcing RLS on the version table
    applies it to the owner, which is what Alembic connects as, and would
    break every future migration at once rather than every endpoint."""
    for enable in (True, False):
        for statement in version_table_migration.rls_statement("postgresql", "alembic_version", "public", enable=enable):
            assert "FORCE" not in statement.upper()


def test_the_version_table_is_not_smuggled_into_the_model_drift_guard(version_table_migration):
    """`test_every_model_table_is_locked_by_some_migration` unions RLS_TABLES
    across revisions and compares it against Base.metadata. This revision
    locks a table that is not a model, so naming its constant RLS_TABLES
    would make that union larger than the set it is checked against - which
    would still pass, while quietly weakening the guard."""
    assert not hasattr(version_table_migration, "RLS_TABLES")