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

"""Integration: migration 0022 (documents.is_image + image_path) schema
contract (phase 122, task 02).

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

* upgrade 0021 → 0022 → ``is_image`` exists with the full contract —
  BOOLEAN, NOT NULL, server default ``false`` — and ``image_path`` —
  TEXT, NULLABLE, no server default — while the 0021 ``documents``
  schema (``content``/``content_hash`` NOT NULL, ``summary`` NULLABLE,
  ``created_at`` + ``created_at_manual``, the (source, path) unique
  constraint — asserted column-based, since the suite's table
  self-heal renames copied constraints) survives;
* a ``documents`` row inserted while the DB is at 0021 backfills
  ``is_image`` to ``false`` and ``image_path`` to NULL (every
  pre-phase-122 row is a text doc — the LOCKED A3 default);
* the ORM contract agrees: a freshly inserted ``Document`` without the
  image fields reads ``is_image is False`` / ``image_path is None``,
  and one with them round-trips through a fresh session;
* downgrade to 0021 → both columns are GONE (A13) while the row
  survives; upgrade back to 0022 → the columns are 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 = "mig0022"
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`` (0022) 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-0022 column set (the 0021
    shape — the image columns, when present, are omitted so their
    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_0022_adds_image_columns(db: Session, alembic: Config) -> None:
    """Upgrade 0021 → 0022: ``is_image`` exists with the full contract
    (BOOLEAN, NOT NULL, server default ``false`` — every pre-phase-122
    row is a text doc) and ``image_path`` (TEXT, NULLABLE, no server
    default — NULL for text docs), both ABSENT at 0021; a pre-0022 row
    backfills ``is_image`` to ``false`` + ``image_path`` to NULL; and
    the 0021 table contract survives the additive upgrade."""
    command.downgrade(alembic, "0021")  # start from the pre-0022 state
    assert _version(db) == "0021"
    assert _column(db, "is_image") is None, "is_image must be absent at 0021"
    assert _column(db, "image_path") is None, "image_path must be absent at 0021"

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

        is_image = _column(db, "is_image")
        assert is_image is not None, "documents.is_image is missing"
        assert is_image[0] == "boolean", "is_image must be BOOLEAN"
        assert is_image[1] == "NO", "is_image must be NOT NULL"
        assert is_image[2] == "false", (
            "is_image must carry the `false` server default — every"
            " pre-phase-122 row is a text doc"
        )

        image_path = _column(db, "image_path")
        assert image_path is not None, "documents.image_path is missing"
        assert image_path[0] == "text", "image_path must be TEXT"
        assert image_path[1] == "YES", "image_path must be NULLABLE"
        assert image_path[2] is None, (
            "image_path must carry NO server default — NULL is the"
            " text-doc value"
        )

        # The pre-0022 row backfilled to (false, NULL) — a text doc.
        row = db.execute(
            text("SELECT is_image, image_path FROM documents WHERE id = :id"),
            {"id": pre_id},
        ).fetchone()
        assert row is not None, "the pre-0022 row must survive the upgrade"
        assert row[0] is False, "the backfilled is_image must be false"
        assert row[1] is None, "the backfilled image_path must be NULL"

        # A row written without the image columns reads the same
        # (the Python-side defaults are False/None — same values).
        new_id = _insert_sql(db, PATH_IMAGE)
        try:
            backfilled = db.execute(
                text("SELECT is_image, image_path FROM documents WHERE id = :id"),
                {"id": new_id},
            ).fetchone()
            assert backfilled == (False, None), (
                "an omitted image state must read (false, NULL)"
            )
        finally:
            _delete(db, new_id)

        # The 0021 schema survives the additive upgrade.
        content = _column(db, "content")
        assert content is not None and content[0] == "text" and content[1] == "NO", (
            "documents.content (0001) must keep its 0021 contract"
        )
        hash_col = _column(db, "content_hash")
        assert (
            hash_col is not None
            and hash_col[0] == "character varying"
            and hash_col[1] == "NO"
        ), "documents.content_hash (0001) must survive the upgrade"
        summary = _column(db, "summary")
        assert summary is not None and summary[0] == "text" and summary[1] == "YES", (
            "documents.summary (phase 30) must survive the upgrade"
        )
        created = _column(db, "created_at")
        assert created is not None and created[0] == "timestamp with time zone"
        assert created[1] == "NO" and "now()" in str(created[2]), (
            "documents.created_at (0020) must keep its `now()` server default"
        )
        manual = _column(db, "created_at_manual")
        assert manual is not None and manual[0] == "boolean" and manual[1] == "NO"
        assert manual[2] == "false", (
            "documents.created_at_manual (0020) must keep its `false` default"
        )
        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_fields_round_trip(db: Session, alembic: Config) -> None:
    """The ORM contract agrees with the column contract: a freshly
    inserted ``Document`` WITHOUT the image fields reads ``is_image is
    False`` / ``image_path is None`` (the text-doc default state), and
    one WITH them round-trips the pair through a FRESH session."""
    command.upgrade(alembic, "head")
    text_doc = Document(
        source=SOURCE,
        path=PATH_TEXT,
        full_path=f"/tmp/{SOURCE}/{PATH_TEXT}",
        title="readme",
        content="# readme\n",
        content_hash="1" * 64,
    )
    image_doc = Document(
        source=SOURCE,
        path=PATH_IMAGE,
        full_path=f"/tmp/{SOURCE}/{PATH_IMAGE}",
        title="diagram",
        content="A description of the diagram.",
        content_hash="2" * 64,
        is_image=True,
        image_path="/srv/bor-images/diagram.png",
    )
    db.add(text_doc)
    db.add(image_doc)
    db.commit()
    try:
        with SessionLocal() as fresh:
            reloaded_text = fresh.get(Document, text_doc.id)
            assert reloaded_text is not None, "the text row must be readable"
            assert reloaded_text.is_image is False, (
                "an omitted is_image must read the False default"
            )
            assert reloaded_text.image_path is None, (
                "an omitted image_path must read NULL"
            )
            reloaded_image = fresh.get(Document, image_doc.id)
            assert reloaded_image is not None, "the image row must be readable"
            assert reloaded_image.is_image is True
            assert reloaded_image.image_path == "/srv/bor-images/diagram.png"
    finally:
        _delete(db, text_doc.id)
        _delete(db, image_doc.id)


def test_downgrade_to_0021_drops_the_columns(db: Session, alembic: Config) -> None:
    """Downgrade 0022 → 0021: both image columns are gone (A13 — fully
    reversible) while the row + its 0021 columns survive, and the rest
    of the 0021 table contract (``content``, ``content_hash``,
    ``created_at``) is intact."""
    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",
    )
    db.add(row)
    db.commit()
    try:
        command.downgrade(alembic, "0021")
        assert _version(db) == "0021"
        assert _column(db, "is_image") is None, "is_image must be dropped"
        assert _column(db, "image_path") is None, "image_path must be dropped"

        surviving = db.execute(
            text(
                "SELECT source, path, title, content, content_hash, created_at"
                " FROM documents WHERE id = :id"
            ),
            {"id": row.id},
        ).fetchone()
        assert surviving is not None, "the row must survive the column drops"
        assert surviving[0] == SOURCE and surviving[1] == PATH_IMAGE
        assert surviving[3] == "A description of the diagram."
        assert surviving[4] == "3" * 64
        assert surviving[5] is not None, "created_at must survive the drops"

        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_columns(db: Session, alembic: Config) -> None:
    """Downgrade to 0021, then upgrade back to 0022: both columns are
    back with the full contract (``is_image`` BOOLEAN NOT NULL default
    ``false``; ``image_path`` TEXT NULLABLE no default)."""
    command.downgrade(alembic, "0021")
    command.upgrade(alembic, "0022")
    assert _version(db) == "0022", "round-trip upgrade must land at 0022"

    is_image = _column(db, "is_image")
    assert is_image is not None, "documents.is_image must be back"
    assert is_image[0] == "boolean", "is_image must be BOOLEAN after the round-trip"
    assert is_image[1] == "NO", "is_image must be NOT NULL after the round-trip"
    assert is_image[2] == "false", (
        "is_image must still carry the `false` server default"
    )

    image_path = _column(db, "image_path")
    assert image_path is not None, "documents.image_path must be back"
    assert image_path[0] == "text"
    assert image_path[1] == "YES"
    assert image_path[2] is None, "image_path must still carry NO server default"