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

"""Integration: migration 0023 (documents.image_context) schema contract
(phase 125, task 01).

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

* upgrade 0022 → 0023 → ``image_context`` exists with the full
  contract — TEXT, NULLABLE, no server default (NULL is the
  standalone-image / pre-phase value, LOCKED A1/A4) — while the 0022
  ``documents`` schema (``is_image`` + ``image_path`` from phase 122,
  ``content``/``content_hash`` NOT NULL, the (source, path) unique
  constraint) survives;
* a ``documents`` row inserted while the DB is at 0022 (a pre-phase
  image doc) reads ``image_context IS NULL`` after the upgrade — the
  pre-phase round-trip pin;
* the ORM contract agrees: a freshly inserted image ``Document`` WITH
  ``image_context`` round-trips it through a fresh session, and one
  WITHOUT (standalone image / text doc) reads ``image_context is None``;
* downgrade to 0022 → the column is GONE (A13) while the row and its
  0022 columns survive; upgrade back to 0023 → 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 Document

SOURCE = "mig0023"
PATH_TEXT = "notes/readme.md"
PATH_IMAGE = "notes/diagram.png"


@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 ``documents``, which would deadlock the repair's
        # ``ALTER TABLE`` (0023) 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 documents
    column."""
    row = db.execute(
        text(
            "SELECT data_type, is_nullable, column_default"
            " FROM information_schema.columns"
            " WHERE table_name = 'documents' AND column_name = :c"
        ),
        {"c": column},
    ).fetchone()
    return tuple(row) if row is not None else None


def _insert_sql(db: Session, path: str) -> uuid.UUID:
    """Insert one documents row with the PRE-0023 column set (the 0022
    shape — ``image_context``, when present, is omitted so its NULL
    backfill is what the row reads)."""
    row_id = uuid.uuid4()
    db.execute(
        text(
            "INSERT INTO documents (id, source, path, full_path, title,"
            " content, content_hash)"
            " VALUES (:id, :s, :p, :f, :t, :c, :h)"
        ),
        {
            "id": row_id,
            "s": SOURCE,
            "p": path,
            "f": f"/tmp/{SOURCE}/{path}",
            "t": path.rsplit("/", 1)[-1],
            "c": "content",
            "h": "0" * 64,
        },
    )
    db.commit()
    return row_id


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


def _unique_column_sets(db: Session) -> set[tuple[str, ...]]:
    """The column tuples of every UNIQUE constraint on ``documents``.

    Column-based (not name-based): the integration suite's table
    self-heal (``tests/integration/conftest.py``) rewrites bloated
    tables via ``CREATE TABLE (LIKE …)``, which renames copied
    constraints (PG auto-names them) — the (source, path) uniqueness
    contract is what must hold, not the original name.
    """
    rows = db.execute(
        text(
            "SELECT (SELECT string_agg(a.attname, ',' ORDER BY k.ord)"
            "   FROM unnest(c.conkey) WITH ORDINALITY k(attnum, ord)"
            "   JOIN pg_attribute a"
            "     ON a.attrelid = c.conrelid AND a.attnum = k.attnum)"
            " FROM pg_constraint c"
            " WHERE c.contype = 'u' AND c.conrelid = 'documents'::regclass"
        )
    ).fetchall()
    return {tuple(r[0].split(",")) for r in rows}


def test_upgrade_to_0023_adds_image_context(db: Session, alembic: Config) -> None:
    """Upgrade 0022 → 0023: ``image_context`` exists with the full
    contract (TEXT, NULLABLE, no server default — NULL is the
    standalone-image / pre-phase value, LOCKED A1) and is ABSENT at
    0022; a pre-0023 row (a phase-122 image doc) backfills to NULL; and
    the 0022 table contract survives the additive upgrade."""
    command.downgrade(alembic, "0022")  # start from the pre-0023 state
    assert _version(db) == "0022"
    assert _column(db, "image_context") is None, (
        "image_context must be absent at 0022"
    )

    pre_id = _insert_sql(db, PATH_IMAGE)  # the 0022 column set (a pre-125 row)
    try:
        command.upgrade(alembic, "0023")
        assert _version(db) == "0023", "alembic_version must be at 0023"

        context = _column(db, "image_context")
        assert context is not None, "documents.image_context is missing"
        assert context[0] == "text", "image_context must be TEXT"
        assert context[1] == "YES", "image_context must be NULLABLE"
        assert context[2] is None, (
            "image_context must carry NO server default — NULL is the"
            " standalone-image / pre-phase value (A1, A4)"
        )

        # The pre-0023 row (inserted with the 0022 column set) reads
        # NULL — every pre-phase row keeps its byte-identical state.
        row = db.execute(
            text("SELECT image_context FROM documents WHERE id = :id"),
            {"id": pre_id},
        ).fetchone()
        assert row is not None, "the pre-0023 row must survive the upgrade"
        assert row[0] is None, (
            "a pre-phase image doc must backfill image_context to NULL"
        )

        # A row written without the column reads the same (an omitted
        # image_context inserts NULL — the text-doc / standalone value).
        new_id = _insert_sql(db, PATH_TEXT)
        try:
            backfilled = db.execute(
                text("SELECT image_context FROM documents WHERE id = :id"),
                {"id": new_id},
            ).fetchone()
            assert backfilled == (None,), (
                "an omitted image_context must read NULL"
            )
        finally:
            _delete(db, new_id)

        # The 0022 schema survives the additive upgrade.
        is_image = _column(db, "is_image")
        assert (
            is_image is not None
            and is_image[0] == "boolean"
            and is_image[1] == "NO"
            and is_image[2] == "false"
        ), "documents.is_image (0022) must keep its 0022 contract"
        image_path = _column(db, "image_path")
        assert (
            image_path is not None
            and image_path[0] == "text"
            and image_path[1] == "YES"
        ), "documents.image_path (0022) must keep its 0022 contract"
        content = _column(db, "content")
        assert content is not None and content[0] == "text" and content[1] == "NO", (
            "documents.content (0001) must keep its 0022 contract"
        )
        assert ("source", "path") in _unique_column_sets(db), (
            "the (source, path) unique constraint must survive the upgrade"
        )
    finally:
        _delete(db, pre_id)


def test_orm_image_context_round_trip(db: Session, alembic: Config) -> None:
    """The ORM contract agrees with the column contract: a freshly
    inserted image ``Document`` WITH ``image_context`` round-trips the
    string through a FRESH session, and one WITHOUT it (a standalone
    image, LOCKED A1/A4) reads ``image_context is None``."""
    command.upgrade(alembic, "head")
    enriched = Document(
        source=SOURCE,
        path=PATH_IMAGE,
        full_path=f"/tmp/{SOURCE}/{PATH_IMAGE}",
        title="diagram",
        content="A description of the diagram.",
        content_hash="1" * 64,
        is_image=True,
        image_path="/srv/bor-images/diagram.png",
        image_context=(
            "![Deployment diagram](diagram.png)\n\nThe diagram\n"
            "shows the cluster layout and how the VMs connect."
        ),
    )
    standalone = Document(
        source=SOURCE,
        path="standalone.png",
        full_path=f"/tmp/{SOURCE}/standalone.png",
        title="standalone",
        content="A description of a standalone image.",
        content_hash="2" * 64,
        is_image=True,
        image_path="/srv/bor-images/standalone.png",
    )
    db.add(enriched)
    db.add(standalone)
    db.commit()
    try:
        with SessionLocal() as fresh:
            reloaded_enriched = fresh.get(Document, enriched.id)
            assert reloaded_enriched is not None, "the image row must be readable"
            assert reloaded_enriched.image_context == enriched.image_context, (
                "image_context must round-trip the exact context text"
            )
            reloaded_standalone = fresh.get(Document, standalone.id)
            assert reloaded_standalone is not None
            assert reloaded_standalone.image_context is None, (
                "an omitted image_context must read NULL (standalone)"
            )
    finally:
        _delete(db, enriched.id)
        _delete(db, standalone.id)


def test_downgrade_to_0022_drops_the_column(db: Session, alembic: Config) -> None:
    """Downgrade 0023 → 0022: the ``image_context`` column is gone
    (A13 — fully reversible) while the row, its 0022 columns (including
    the phase-122 image pair) and the table's unique constraint
    survive."""
    command.upgrade(alembic, "head")
    row = Document(
        source=SOURCE,
        path=PATH_IMAGE,
        full_path=f"/tmp/{SOURCE}/{PATH_IMAGE}",
        title="diagram",
        content="A description of the diagram.",
        content_hash="3" * 64,
        is_image=True,
        image_path="/srv/bor-images/diagram.png",
        image_context="![alt](diagram.png)\n\nSurrounding paragraph.",
    )
    db.add(row)
    db.commit()
    try:
        command.downgrade(alembic, "0022")
        assert _version(db) == "0022"
        assert _column(db, "image_context") is None, (
            "image_context must be dropped"
        )

        surviving = db.execute(
            text(
                "SELECT source, path, content, content_hash, is_image,"
                " image_path FROM documents WHERE id = :id"
            ),
            {"id": row.id},
        ).fetchone()
        assert surviving is not None, "the row must survive the column drop"
        assert surviving[0] == SOURCE and surviving[1] == PATH_IMAGE
        assert surviving[2] == "A description of the diagram."
        assert surviving[3] == "3" * 64
        assert surviving[4] is True, "is_image (0022) must survive the drop"
        assert surviving[5] == "/srv/bor-images/diagram.png"

        assert ("source", "path") in _unique_column_sets(db), (
            "the (source, path) unique constraint 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 0022, then upgrade back to 0023: the column is back
    with the full contract (TEXT, NULLABLE, no server default) — a
    clean upgrade → downgrade → upgrade round-trip (A13)."""
    command.downgrade(alembic, "0022")
    command.upgrade(alembic, "head")
    assert _version(db) == "0023", (
        "round-trip upgrade must land at head (0023)"
    )

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