"""Integration: migration 0027 (documents.page_strip_mode) schema contract
(phase 163, task 01).

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

* upgrade 0026 → 0027 → ``page_strip_mode`` exists with the full
  contract — VARCHAR(16), NULLABLE, NO server default (NULL = "no strip"
  for every pre-phase row, every non-qualifying document and every
  fail-soft miss) and UNINDEXED (a display fact, never a query surface —
  the content endpoint reads it from the ROW, the ``file_hash`` /
  ``page_count`` convention) — while the 0026 ``documents`` schema
  survives, in particular ``page_count`` keeping its 0026 contract and
  ``content_hash`` staying NOT NULL and INDEXED;
* rows inserted while the DB is at 0026 backfill ``page_strip_mode`` to
  NULL — including a row whose ``page_count`` IS set (the pre-phase strip
  with no stamped class: the wire omits the key and the viewer keeps
  today's "primary" DOM until the task-03 re-gate stamps it);
* the ORM contract agrees: a freshly inserted ``Document`` with no mode
  reads ``page_strip_mode is None`` through a fresh session, and each of
  the two classes (``"primary"`` / ``"additive"``) round-trips verbatim;
* downgrade to 0026 → the column is GONE (A13) while the row + its
  content + its ``page_count`` survive; upgrade back to 0027 → 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 hashlib
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 = "mig0027"
CONTENT = "content here"
CONTENT_HASH = hashlib.sha256(CONTENT.encode()).hexdigest()


@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`` 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, character_maximum_length)
    for one documents column."""
    row = db.execute(
        text(
            "SELECT data_type, is_nullable, column_default, character_maximum_length"
            " 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 _index_covering(db: Session, column: str) -> str | None:
    """The name of any documents index whose definition mentions the
    column (None = UNINDEXED)."""
    return db.execute(
        text(
            "SELECT indexname FROM pg_indexes"
            " WHERE tablename = 'documents' AND indexdef ILIKE :like"
        ),
        {"like": f"%{column}%"},
    ).scalar()


def _insert_sql(db: Session, path: str, page_count: int | None = None) -> uuid.UUID:
    """Insert one documents row with the PRE-0027 column set (the 0026
    shape — the mode column, when present, is omitted so a NULL backfill
    is what the row reads; ``page_count`` is 0026's own column and may be
    set to model a pre-phase strip)."""
    doc_id = uuid.uuid4()
    db.execute(
        text(
            "INSERT INTO documents (id, source, path, full_path, title, content,"
            " content_hash, page_count)"
            " VALUES (:id, :s, :p, :fp, :t, :c, :h, :pc)"
        ),
        {
            "id": doc_id,
            "s": SOURCE,
            "p": path,
            "fp": f"/tmp/{path}",
            "t": f"Doc {path}",
            "c": CONTENT,
            "h": CONTENT_HASH,
            "pc": page_count,
        },
    )
    db.commit()
    return doc_id


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


def _assert_0026_schema_survives(db: Session) -> None:
    """The 0026 documents contract must survive the additive upgrade —
    ``page_count`` keeps its INTEGER / NULLABLE / NO-default / UNINDEXED
    shape (the sibling column this one rides) and ``content_hash`` stays
    NOT NULL and indexed (the retriever's identity)."""
    page_count = _column(db, "page_count")
    assert page_count is not None, "documents.page_count (0026) must survive the upgrade"
    assert page_count[0] == "integer", "page_count must stay INTEGER"
    assert page_count[1] == "YES", "page_count must stay NULLABLE"
    assert page_count[2] is None, "page_count must keep NO server default"
    assert _index_covering(db, "page_count") is None, "page_count must stay UNINDEXED"

    content = _column(db, "content_hash")
    assert content is not None and content[0] == "character varying" and content[1] == "NO", (
        "documents.content_hash (0001) must keep its contract"
    )
    assert (
        _index_covering(db, "content_hash") is not None
    ), "the ix_documents_content_hash index must survive the upgrade"

    indexed = _column(db, "indexed_at")
    assert indexed is not None and indexed[0] == "timestamp with time zone", (
        "documents.indexed_at (0001) must survive the upgrade"
    )
    is_image = _column(db, "is_image")
    assert is_image is not None and is_image[0] == "boolean" and is_image[1] == "NO", (
        "documents.is_image (0022) must survive the upgrade"
    )


def test_upgrade_to_0027_adds_page_strip_mode(db: Session, alembic: Config) -> None:
    """Upgrade 0026 → 0027: ``page_strip_mode`` exists with the full
    contract (VARCHAR(16), NULLABLE, NO server default — NULL = "no
    strip" — and UNINDEXED: a display fact, never a query surface), is
    ABSENT at 0026, and every pre-0027 row backfills it to NULL —
    including the row whose ``page_count`` IS set (the pre-phase strip
    with no stamped class), while the 0026 table contract (the
    ``page_count`` sibling, the indexed ``content_hash``) survives.
    """
    command.downgrade(alembic, "0026")  # start from the pre-0027 state
    assert _version(db) == "0026"
    assert _column(db, "page_strip_mode") is None, "page_strip_mode must be absent at 0026"

    plain_id = _insert_sql(db, "plain.pdf")  # the 0026 column set, no strip
    pre_strip_id = _insert_sql(db, "prephase.pdf", page_count=36)  # a live pre-phase strip

    try:
        command.upgrade(alembic, "0027")
        assert _version(db) == "0027", "alembic_version must be at 0027"

        col = _column(db, "page_strip_mode")
        assert col is not None, "documents.page_strip_mode is missing"
        assert col[0] == "character varying", "page_strip_mode must be VARCHAR"
        assert col[3] == 16, "page_strip_mode must be VARCHAR(16) — the two class names"
        assert col[1] == "YES", (
            "page_strip_mode must be NULLABLE (NULL = no strip, and the pre-phase shape)"
        )
        assert col[2] is None, (
            "page_strip_mode must carry NO server default — NULL is the meaningful"
            " pre-phase state"
        )
        assert _index_covering(db, "page_strip_mode") is None, (
            "page_strip_mode must be UNINDEXED — a display fact, never a query"
            " surface (the content endpoint reads it from the row)"
        )

        # Every pre-0027 row backfills to NULL — no backfill, no default.
        assert (
            db.execute(
                text("SELECT page_count, page_strip_mode FROM documents WHERE id = :id"),
                {"id": plain_id},
            ).fetchone()
            == (None, None)
        ), "a pre-0027 row without a strip must read both columns NULL"
        assert (
            db.execute(
                text("SELECT page_count, page_strip_mode FROM documents WHERE id = :id"),
                {"id": pre_strip_id},
            ).fetchone()
            == (36, None)
        ), (
            "a pre-phase strip (page_count set) must backfill page_strip_mode to"
            " NULL — the wire omits the key and the viewer keeps today's DOM"
        )

        _assert_0026_schema_survives(db)
    finally:
        _delete(db, plain_id)
        _delete(db, pre_strip_id)


@pytest.mark.parametrize("mode", ["primary", "additive"])
def test_orm_page_strip_mode_round_trips(db: Session, alembic: Config, mode: str) -> None:
    """The ORM contract agrees with the column contract: each class
    (``"primary"`` = the fallback class, ``"additive"`` = the pictorial
    class) round-trips VERBATIM through a fresh session, and a row
    created with no mode reads ``page_strip_mode is None`` (the
    pre-phase shape — NULL until the importer stamps it)."""
    command.upgrade(alembic, "head")
    stamped = Document(
        source=SOURCE,
        path="stamped.pdf",
        full_path="/tmp/stamped.pdf",
        title="stamped",
        content=CONTENT,
        content_hash=CONTENT_HASH,
        page_count=36,
        page_strip_mode=mode,
    )
    unstamped = Document(
        source=SOURCE,
        path="unstamped.md",
        full_path="/tmp/unstamped.md",
        title="unstamped",
        content=CONTENT,
        content_hash=CONTENT_HASH,
    )
    db.add(stamped)
    db.add(unstamped)
    db.commit()
    try:
        with SessionLocal() as fresh:
            reloaded = fresh.get(Document, stamped.id)
            assert reloaded is not None, "the stamped row must be readable"
            assert reloaded.page_strip_mode == mode, (
                f"the {mode!r} class must round-trip through the DB verbatim"
            )
            assert reloaded.page_count == 36, "the sibling page_count must round-trip too"
            reloaded_unstamped = fresh.get(Document, unstamped.id)
            assert reloaded_unstamped is not None, "the unstamped row must be readable"
            assert reloaded_unstamped.page_strip_mode is None, (
                "a row created without a mode must read NULL (the pre-phase shape)"
            )
            assert reloaded_unstamped.content_hash == CONTENT_HASH, (
                "content_hash must be untouched by the page_strip_mode addition"
            )
    finally:
        _delete(db, stamped.id)
        _delete(db, unstamped.id)


def test_downgrade_to_0026_drops_the_column(db: Session, alembic: Config) -> None:
    """Downgrade 0027 → 0026: the page_strip_mode column is gone (A13 —
    fully reversible, the class is re-derivable from the source file by
    the deterministic re-gate) while the row + its content + its 0026
    ``page_count`` survive, and the rest of the table contract is intact."""
    command.upgrade(alembic, "head")
    row = Document(
        source=SOURCE,
        path="down.pdf",
        full_path="/tmp/down.pdf",
        title="down",
        content=CONTENT,
        content_hash=CONTENT_HASH,
        page_count=36,
        page_strip_mode="additive",
    )
    db.add(row)
    db.commit()
    try:
        command.downgrade(alembic, "0026")
        assert _version(db) == "0026"
        assert _column(db, "page_strip_mode") is None, "page_strip_mode must be dropped"

        surviving = db.execute(
            text(
                "SELECT source, path, content, content_hash, page_count, is_image"
                " FROM documents WHERE id = :id"
            ),
            {"id": row.id},
        ).fetchone()
        assert surviving is not None and surviving[0] == SOURCE, (
            "the row must survive the column drop"
        )
        assert surviving[1] == "down.pdf" and surviving[2] == CONTENT, (
            "path + content must survive the drop"
        )
        assert surviving[3] == CONTENT_HASH, (
            "content_hash (the retriever's identity) must survive the drop"
        )
        assert surviving[4] == 36, (
            "page_count (the 0026 sibling) must survive the drop — the strip"
            " itself is untouched by this migration's reversal"
        )
        assert surviving[5] is False, "is_image must survive the drop"
        assert _index_covering(db, "content_hash") is not None, (
            "the content_hash 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 0026, then upgrade back to 0027: ``page_strip_mode``
    is back with the full contract (VARCHAR(16), NULLABLE, NO server
    default, UNINDEXED), a row seeded at 0026 — strip included — reads
    the mode as NULL (the pre-phase shape on the round-trip), and the
    0026 schema (the ``page_count`` sibling, the indexed
    ``content_hash``) is intact."""
    command.downgrade(alembic, "0026")
    pre_id = _insert_sql(db, "round.pdf", page_count=36)
    try:
        command.upgrade(alembic, "0027")
        assert _version(db) == "0027", "round-trip upgrade must land at 0027"

        col = _column(db, "page_strip_mode")
        assert col is not None, "documents.page_strip_mode must be back"
        assert col[0] == "character varying", "page_strip_mode must be VARCHAR after the round-trip"
        assert col[3] == 16, "page_strip_mode must be VARCHAR(16) after the round-trip"
        assert col[1] == "YES", "page_strip_mode must be NULLABLE after the round-trip"
        assert col[2] is None, (
            "page_strip_mode must still carry NO server default after the round-trip"
        )
        assert _index_covering(db, "page_strip_mode") is None, (
            "page_strip_mode must stay UNINDEXED after the round-trip"
        )

        assert (
            db.execute(
                text("SELECT page_count, page_strip_mode FROM documents WHERE id = :id"),
                {"id": pre_id},
            ).fetchone()
            == (36, None)
        ), (
            "a row seeded at 0026 must read page_strip_mode NULL after the"
            " round-trip while its page_count stands"
        )

        _assert_0026_schema_survives(db)
    finally:
        _delete(db, pre_id)