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

"""Integration: migration 0017 (folder_summaries) schema contract
(phase 94, task 01).

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

* upgrade 0016 → 0017 → the ``folder_summaries`` table exists with the
  full column contract — PK ``(source, folder_path)`` (VARCHAR(120) /
  VARCHAR(1000) NOT NULL, mirroring ``documents.source`` /
  ``documents.path``), ``summary`` TEXT NOT NULL, ``updated_at``
  TIMESTAMPTZ NOT NULL with the now() server default (house style) —
  while the 0016 ``ui_settings`` schema survives;
* inserted rows round-trip their values (a source-root row with
  ``folder_path = ''`` and a nested-folder row);
* downgrade to 0016 → the table is GONE (A13 — reversible), the rest of
  the schema (``ui_settings`` + ``api_tokens``) survives;
* upgrade back to 0017 → the table 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

from collections.abc import Iterator
from typing import Any

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

from alembic import command
from app.db import db_available


@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:
        command.upgrade(cfg, "head")


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


def _table_exists(db: Session, table: str) -> bool:
    return (
        db.execute(
            text("SELECT 1 FROM information_schema.tables WHERE table_name = :t"),
            {"t": table},
        ).scalar()
        is not None
    )


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


def _pk_columns(db: Session, table: str) -> list[str]:
    """The table's PRIMARY KEY columns in ordinal position."""
    rows = db.execute(
        text(
            "SELECT kcu.column_name"
            " FROM information_schema.table_constraints tc"
            " JOIN information_schema.key_column_usage kcu"
            "   ON kcu.constraint_name = tc.constraint_name"
            "  AND kcu.table_name = tc.table_name"
            " WHERE tc.table_name = :t"
            "   AND tc.constraint_type = 'PRIMARY KEY'"
            " ORDER BY kcu.ordinal_position"
        ),
        {"t": table},
    ).fetchall()
    return [r[0] for r in rows]


def _clear_rows(db: Session) -> None:
    db.execute(text("DELETE FROM folder_summaries"))
    db.commit()


def test_upgrade_to_0017_creates_folder_summaries(db: Session, alembic: Config) -> None:
    """Upgrade 0016 → 0017: the table exists with the full column
    contract (PK ``(source, folder_path)`` mirroring
    ``documents.source`` / ``documents.path``; TEXT summary NOT NULL;
    TIMESTAMPTZ updated_at NOT NULL with the now() server default), and
    the table is ABSENT at 0016 while the 0016 ``ui_settings`` schema
    survives the upgrade."""
    command.downgrade(alembic, "0016")  # start from the pre-0017 state
    assert _version(db) == "0016"
    assert not _table_exists(db, "folder_summaries"), (
        "folder_summaries must be absent at 0016"
    )

    command.upgrade(alembic, "0017")
    assert _version(db) == "0017", "alembic_version must be at 0017"
    assert _table_exists(db, "folder_summaries"), "the table must exist at 0017"

    source = _column(db, "folder_summaries", "source")
    assert source is not None, "folder_summaries.source is missing"
    assert source[0] == "character varying", "source must be VARCHAR"
    assert source[1] == "NO", "source must be NOT NULL (PK part 1)"
    assert source[2] is None, "source must have no server default"
    assert source[3] == 120, "source must be String(120) — documents.source"

    folder = _column(db, "folder_summaries", "folder_path")
    assert folder is not None, "folder_summaries.folder_path is missing"
    assert folder[0] == "character varying", "folder_path must be VARCHAR"
    assert folder[1] == "NO", "folder_path must be NOT NULL (PK part 2)"
    assert folder[2] is None, "folder_path must have no server default"
    assert folder[3] == 1000, "folder_path must be String(1000) — documents.path"

    summary = _column(db, "folder_summaries", "summary")
    assert summary is not None, "folder_summaries.summary is missing"
    assert summary[0] == "text", "summary must be TEXT"
    assert summary[1] == "NO", "summary must be NOT NULL (never stored empty)"

    updated = _column(db, "folder_summaries", "updated_at")
    assert updated is not None, "folder_summaries.updated_at is missing"
    assert updated[0] == "timestamp with time zone", "updated_at must be TIMESTAMPTZ"
    assert updated[1] == "NO", "updated_at must be NOT NULL"
    assert updated[2] is not None and "now" in str(updated[2]), (
        "updated_at must carry the now() server default (house style)"
    )

    assert _pk_columns(db, "folder_summaries") == ["source", "folder_path"]

    # The 0016 schema survives the additive upgrade.
    semantic = _column(db, "ui_settings", "ok_bg")
    assert semantic is not None and semantic[3] == 7, (
        "ui_settings.ok_bg (0016) must survive the upgrade"
    )


def test_inserted_rows_round_trip(db: Session, alembic: Config) -> None:
    """At 0017, a source-root row (``folder_path = ''``) and a nested
    folder row round-trip their values, and the composite PK rejects a
    duplicate (source, folder_path) pair."""
    command.upgrade(alembic, "head")
    try:
        db.execute(
            text(
                "INSERT INTO folder_summaries (source, folder_path, summary)"
                " VALUES ('Homelab', '', 'Root summary.')"
            )
        )
        db.execute(
            text(
                "INSERT INTO folder_summaries (source, folder_path, summary)"
                " VALUES ('Homelab', 'deployments/ansible', 'Ansible summary.')"
            )
        )
        db.commit()
        rows = db.execute(
            text(
                "SELECT source, folder_path, summary, updated_at"
                " FROM folder_summaries ORDER BY folder_path"
            )
        ).fetchall()
        assert len(rows) == 2, "both rows must be stored"
        assert rows[0][0] == "Homelab" and rows[0][1] == "", (
            "the source-root row uses folder_path = ''"
        )
        assert rows[0][2] == "Root summary.", "summary must round-trip verbatim"
        assert rows[0][3] is not None, "updated_at must be stamped (server default)"
        assert rows[1][1] == "deployments/ansible", (
            "a nested folder path must round-trip verbatim"
        )
        assert rows[1][2] == "Ansible summary."

        with pytest.raises(IntegrityError):
            db.execute(
                text(
                    "INSERT INTO folder_summaries (source, folder_path, summary)"
                    " VALUES ('Homelab', 'deployments/ansible', 'dup')"
                )
            )
        db.rollback()  # the IntegrityError aborts the open transaction
    finally:
        _clear_rows(db)


def test_downgrade_to_0016_drops_the_table(db: Session, alembic: Config) -> None:
    """Downgrade 0017 → 0016: the table is gone (A13 — fully
    reversible) while the rest of the schema survives (the 0016
    ``ui_settings`` semantic columns, ``api_tokens``, ``documents``)."""
    command.downgrade(alembic, "0016")
    assert _version(db) == "0016"
    assert not _table_exists(db, "folder_summaries"), (
        "folder_summaries must be dropped"
    )

    semantic = _column(db, "ui_settings", "accent_line")
    assert semantic is not None and semantic[3] == 7, (
        "ui_settings.accent_line (0016) must survive the downgrade"
    )
    token_col = _column(db, "api_tokens", "token_hash")
    assert token_col is not None and token_col[0] == "character varying", (
        "api_tokens.token_hash must survive the downgrade"
    )
    doc_path = _column(db, "documents", "path")
    assert doc_path is not None and doc_path[3] == 1000, (
        "documents.path must survive the downgrade"
    )


def test_upgrade_round_trip_restores_the_table(db: Session, alembic: Config) -> None:
    """Downgrade to 0016, then upgrade back to 0017: the table is back
    with the column contract and PK intact."""
    command.downgrade(alembic, "0016")
    command.upgrade(alembic, "0017")
    assert _version(db) == "0017", "round-trip upgrade must land at 0017"
    assert _table_exists(db, "folder_summaries"), "the table must be back"

    folder = _column(db, "folder_summaries", "folder_path")
    assert folder is not None, "folder_summaries.folder_path must be back"
    assert folder[0] == "character varying", "folder_path must be VARCHAR"
    assert folder[1] == "NO", "folder_path must be NOT NULL after the round-trip"
    assert folder[3] == 1000, "folder_path must be String(1000) after the round-trip"
    assert _pk_columns(db, "folder_summaries") == ["source", "folder_path"]