brain-of-reese/tests/integration/test_migration_0021.py

"""Integration: migration 0021 (git_sources.token) schema contract
(phase 121, task 01).

Drives the **real Alembic engine** against the live dev database
(``podman compose up -d db``), mirroring the house pattern of
``test_migration_0020.py`` (information_schema assertions on the state
the migration must leave). The tests target the 0020 → 0021 step
explicitly so later migrations cannot break the pins:

* upgrade 0020 → 0021 → ``token`` exists with the full contract —
  TEXT, NULLABLE, no server default — while the 0020 ``git_sources``
  schema (``url`` NOT NULL + the unique index, ``kind``, ``path``,
  ``ignore_paths``, ``include_hidden``, ``added_at``) survives;
* a ``git_sources`` row inserted while the DB is at 0020 backfills
  ``token`` to NULL (a pre-phase-121 row is a public repo — or a
  legacy embedded-token row whose credential lives in ``url``);
* the ORM contract agrees: a freshly inserted ``GitSource`` with an
  explicit ``token`` round-trips it through a fresh session, and a row
  without one reads ``token is None``;
* downgrade to 0020 → the column is GONE (A13) while the row + its
  ``url`` survive; upgrade back to 0021 → the column is back
  (round-trip).

The ``alembic`` fixture guarantees the DB ends at head even if a test
fails or the process is interrupted.
"""
from __future__ import annotations

import uuid
from collections.abc import Iterator
from typing import Any

import pytest
from alembic.config import Config
from sqlalchemy import text
from sqlalchemy.orm import Session

from alembic import command
from app.db import SessionLocal, db_available
from app.models import GitSource

URL_BARE = "https://github.com/mig0021/bare.git"
URL_TOKENED = "https://github.com/mig0021/tokened.git"
TOKEN = "ghp_mig0021secret"


@pytest.fixture()
def alembic(db: Session) -> Iterator[Config]:
    """Real Alembic config bound to the dev DB (URL from app settings).

    Starts at head (repairs an interrupted earlier run); teardown
    upgrades to head no matter what happened, so the dev DB is never
    left below head.
    """
    if not db_available():
        pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
    cfg = Config()  # no alembic.ini file — env.py gets the URL from app config
    cfg.set_main_option("script_location", "alembic")
    command.upgrade(cfg, "head")
    try:
        yield cfg
    finally:
        # Release the test session's open transaction BEFORE the repair
        # DDL: an idle-in-transaction SELECT holds an ACCESS SHARE lock
        # on ``git_sources``, which would deadlock the repair's
        # ``ALTER TABLE`` (0021) forever.
        db.rollback()
        command.upgrade(cfg, "head")


def _version(db: Session) -> str | None:
    return db.execute(text("SELECT version_num FROM alembic_version")).scalar()


def _column(db: Session, column: str) -> tuple[Any, ...] | None:
    """(data_type, is_nullable, column_default) for one git_sources
    column."""
    row = db.execute(
        text(
            "SELECT data_type, is_nullable, column_default"
            " FROM information_schema.columns"
            " WHERE table_name = 'git_sources' AND column_name = :c"
        ),
        {"c": column},
    ).fetchone()
    return tuple(row) if row is not None else None


def _insert_sql(db: Session, url: str) -> uuid.UUID:
    """Insert one git_sources row with the PRE-0021 column set (the
    0020 shape — the token column, when present, is omitted so a NULL
    backfill is what the row reads)."""
    row_id = uuid.uuid4()
    db.execute(
        text(
            "INSERT INTO git_sources (id, url, kind, path, ignore_paths,"
            " include_hidden)"
            " VALUES (:id, :u, 'git', NULL, '[]', false)"
        ),
        {"id": row_id, "u": url},
    )
    db.commit()
    return row_id


def _delete(db: Session, row_id: uuid.UUID) -> None:
    db.execute(text("DELETE FROM git_sources WHERE id = :id"), {"id": row_id})
    db.commit()


def test_upgrade_to_0021_adds_token(db: Session, alembic: Config) -> None:
    """Upgrade 0020 → 0021: ``token`` exists with the full contract
    (TEXT, NULLABLE, no server default — NULL = public/legacy row), is
    ABSENT at 0020, a pre-0021 row backfills ``token`` to NULL, and an
    omitted token on a new row stays NULL — while the 0020 table
    contract (``url`` + unique index, ``kind``, ``ignore_paths``,
    ``include_hidden``, ``added_at``) survives."""
    command.downgrade(alembic, "0020")  # start from the pre-0021 state
    assert _version(db) == "0020"
    assert _column(db, "token") is None, "token must be absent at 0020"

    pre_id = _insert_sql(db, URL_BARE)  # the 0020 column set
    try:
        command.upgrade(alembic, "0021")
        assert _version(db) == "0021", "alembic_version must be at 0021"

        token = _column(db, "token")
        assert token is not None, "git_sources.token is missing"
        assert token[0] == "text", "token must be TEXT"
        assert token[1] == "YES", "token must be NULLABLE"
        assert token[2] is None, (
            "token must carry NO server default — NULL is the"
            " public/legacy value"
        )

        # The pre-0021 row backfilled to NULL (a public repo, or a
        # legacy embedded-token row whose credential lives in url).
        row = db.execute(
            text("SELECT url, token FROM git_sources WHERE id = :id"),
            {"id": pre_id},
        ).fetchone()
        assert row is not None and row[0] == URL_BARE, (
            "the pre-0021 row must survive the upgrade"
        )
        assert row[1] is None, "the backfilled token must be NULL"

        # A row written without the column stays NULL (no default to
        # fill it in — the Python-side default is None, same value).
        new_id = _insert_sql(db, URL_TOKENED)
        try:
            tokened = db.execute(
                text("SELECT token FROM git_sources WHERE id = :id"),
                {"id": new_id},
            ).scalar()
            assert tokened is None, "an omitted token must stay NULL"
        finally:
            _delete(db, new_id)

        # The 0020 schema survives the additive upgrade.
        url = _column(db, "url")
        assert url is not None and url[0] == "text" and url[1] == "NO", (
            "git_sources.url (0006) must keep its 0020 contract"
        )
        kind = _column(db, "kind")
        assert kind is not None and kind[0] == "text" and kind[1] == "NO", (
            "git_sources.kind (0007) must survive the upgrade"
        )
        ignore = _column(db, "ignore_paths")
        assert ignore is not None and ignore[0] == "jsonb" and ignore[1] == "NO", (
            "git_sources.ignore_paths (0013) must survive the upgrade"
        )
        hidden = _column(db, "include_hidden")
        assert hidden is not None and hidden[0] == "boolean" and hidden[1] == "NO", (
            "git_sources.include_hidden (0019) must survive the upgrade"
        )
        added = _column(db, "added_at")
        assert added is not None and added[0] == "timestamp with time zone", (
            "git_sources.added_at (0006) must survive the upgrade"
        )
        assert added[1] == "NO" and "now()" in str(added[2]), (
            "git_sources.added_at must keep its `now()` server default"
        )
        index = db.execute(
            text(
                "SELECT indexname FROM pg_indexes"
                " WHERE tablename = 'git_sources'"
                " AND indexname = 'uq_git_sources_url'"
            )
        ).scalar()
        assert index is not None, (
            "the uq_git_sources_url unique index must survive the upgrade"
        )
    finally:
        _delete(db, pre_id)


def test_orm_token_round_trips(db: Session, alembic: Config) -> None:
    """The ORM contract agrees with the column contract: a freshly
    inserted ``GitSource`` with an explicit ``token`` round-trips the
    credential through a FRESH session, and a row without one reads
    ``token is None`` (the NULL public/legacy state)."""
    command.upgrade(alembic, "head")
    row_tokened = GitSource(url=URL_TOKENED, kind="git", token=TOKEN)
    row_bare = GitSource(url=URL_BARE, kind="git")
    db.add(row_tokened)
    db.add(row_bare)
    db.commit()
    try:
        with SessionLocal() as fresh:
            reloaded_tokened = fresh.get(GitSource, row_tokened.id)
            assert reloaded_tokened is not None, "the tokened row must be readable"
            assert reloaded_tokened.token == TOKEN, (
                "the explicit token must round-trip through the DB"
            )
            reloaded_bare = fresh.get(GitSource, row_bare.id)
            assert reloaded_bare is not None, "the bare row must be readable"
            assert reloaded_bare.token is None, (
                "a row without a token must read token is None"
            )
    finally:
        _delete(db, row_tokened.id)
        _delete(db, row_bare.id)


def test_downgrade_to_0020_drops_the_column(db: Session, alembic: Config) -> None:
    """Downgrade 0021 → 0020: the token column is gone (A13 — fully
    reversible) while the row + its ``url`` survive, and the rest of
    the 0020 table contract (``url`` unique index, ``kind``,
    ``added_at``) is intact."""
    command.upgrade(alembic, "head")
    row = GitSource(url=URL_TOKENED, kind="git", token=TOKEN)
    db.add(row)
    db.commit()
    try:
        command.downgrade(alembic, "0020")
        assert _version(db) == "0020"
        assert _column(db, "token") is None, "token must be dropped"

        surviving = db.execute(
            text(
                "SELECT url, kind, ignore_paths, include_hidden, added_at"
                " FROM git_sources WHERE id = :id"
            ),
            {"id": row.id},
        ).fetchone()
        assert surviving is not None and surviving[0] == URL_TOKENED, (
            "the row must survive the column drop"
        )
        assert surviving[1] == "git" and surviving[2] == [] and surviving[3] is False, (
            "kind + ignore_paths + include_hidden must survive the drop"
        )
        assert surviving[4] is not None, "added_at must survive the drop"

        index = db.execute(
            text(
                "SELECT indexname FROM pg_indexes"
                " WHERE tablename = 'git_sources'"
                " AND indexname = 'uq_git_sources_url'"
            )
        ).scalar()
        assert index is not None, "the unique index must survive the downgrade"
    finally:
        _delete(db, row.id)
    # Repair: the fixture teardown re-upgrades to head.


def test_upgrade_round_trip_restores_the_column(db: Session, alembic: Config) -> None:
    """Downgrade to 0020, then upgrade back to 0021: ``token`` is back
    with the full contract (TEXT, NULLABLE, no server default)."""
    command.downgrade(alembic, "0020")
    command.upgrade(alembic, "0021")
    assert _version(db) == "0021", "round-trip upgrade must land at 0021"

    token = _column(db, "token")
    assert token is not None, "git_sources.token must be back"
    assert token[0] == "text", "token must be TEXT after the round-trip"
    assert token[1] == "YES", "token must be NULLABLE after the round-trip"
    assert token[2] is None, (
        "token must still carry NO server default after the round-trip"
    )