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

"""Integration: migration 0018 (folder_summaries.manually_edited)
schema contract (phase 97, task 01).

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

* upgrade 0017 → 0018 → the ``manually_edited`` column exists with the
  full contract — BOOLEAN NOT NULL, server default ``false`` — while
  the 0017 ``folder_summaries`` schema (PK, summary, updated_at)
  survives;
* pre-0018 rows backfill ``false`` (an AI-written row stays an AI row)
  and a row written without the column takes the default;
* a row written with ``manually_edited = true`` round-trips the flag;
* downgrade to 0017 → the column is GONE (A13 — reversible) while the
  rows + their summaries survive;
* upgrade back to 0018 → 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

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 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:
        # Release the test session's open transaction BEFORE the repair
        # DDL: an idle-in-transaction SELECT holds an ACCESS SHARE lock
        # on ``folder_summaries``, which would deadlock the repair's
        # ``ALTER TABLE`` (0018) 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, 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 _flag(db: Session, source: str, folder_path: str) -> Any:
    return db.execute(
        text(
            "SELECT manually_edited FROM folder_summaries"
            " WHERE source = :s AND folder_path = :f"
        ),
        {"s": source, "f": folder_path},
    ).scalar_one()


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


def test_upgrade_to_0018_adds_manually_edited(db: Session, alembic: Config) -> None:
    """Upgrade 0017 → 0018: the column exists with the full contract
    (BOOLEAN NOT NULL, server default ``false``), is ABSENT at 0017,
    pre-0018 rows backfill ``false`` (an AI row stays an AI row), a new
    row without the column takes the default, and an explicit ``true``
    round-trips — while the 0017 table contract survives."""
    command.downgrade(alembic, "0017")  # start from the pre-0018 state
    assert _version(db) == "0017"
    assert _column(db, "folder_summaries", "manually_edited") is None, (
        "the flag must be absent at 0017"
    )

    try:
        # A pre-0018 AI-written row — must backfill ``false``.
        db.execute(
            text(
                "INSERT INTO folder_summaries (source, folder_path, summary)"
                " VALUES ('OldSource', 'old/folder', 'pre-0018 summary')"
            )
        )
        db.commit()

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

        flag = _column(db, "folder_summaries", "manually_edited")
        assert flag is not None, "folder_summaries.manually_edited is missing"
        assert flag[0] == "boolean", "manually_edited must be BOOLEAN"
        assert flag[1] == "NO", "manually_edited must be NOT NULL"
        assert flag[2] is not None and "false" in str(flag[2]), (
            "manually_edited must carry the `false` server default"
        )

        # The pre-0018 row backfilled ``false`` — an AI row stays an AI row.
        assert _flag(db, "OldSource", "old/folder") is False

        # A row written without the column takes the server default.
        db.execute(
            text(
                "INSERT INTO folder_summaries (source, folder_path, summary)"
                " VALUES ('NewSource', '', 'root summary')"
            )
        )
        db.commit()
        assert _flag(db, "NewSource", "") is False, (
            "an omitted flag takes the `false` server default"
        )

        # The flag round-trips through an explicit ``true``.
        db.execute(
            text(
                "UPDATE folder_summaries SET manually_edited = true"
                " WHERE source = 'NewSource'"
            )
        )
        db.commit()
        assert _flag(db, "NewSource", "") is True, (
            "manually_edited = true must round-trip"
        )

        # The 0017 schema survives the additive upgrade.
        summary = _column(db, "folder_summaries", "summary")
        assert summary is not None and summary[0] == "text" and summary[1] == "NO", (
            "folder_summaries.summary (0017) must survive the upgrade"
        )
        folder = _column(db, "folder_summaries", "folder_path")
        assert folder is not None and folder[3] == 1000, (
            "folder_summaries.folder_path (0017) must survive the upgrade"
        )
    finally:
        _clear_rows(db)


def test_downgrade_to_0017_drops_the_column(db: Session, alembic: Config) -> None:
    """Downgrade 0018 → 0017: the column is gone (A13 — fully
    reversible) while the rows + their summaries survive, and the rest
    of the schema (the 0017 table contract, ``documents``) is intact."""
    command.upgrade(alembic, "head")
    try:
        db.execute(
            text(
                "INSERT INTO folder_summaries"
                " (source, folder_path, summary, manually_edited)"
                " VALUES ('ManualSrc', '', 'owner text', true)"
            )
        )
        db.commit()
        command.downgrade(alembic, "0017")
        assert _version(db) == "0017"
        assert _column(db, "folder_summaries", "manually_edited") is None, (
            "the flag must be dropped"
        )
        row = db.execute(
            text(
                "SELECT source, folder_path, summary FROM folder_summaries"
                " WHERE source = 'ManualSrc'"
            )
        ).fetchone()
        assert row is not None and row[2] == "owner text", (
            "the row and its summary must survive the column drop"
        )

        summary = _column(db, "folder_summaries", "summary")
        assert summary is not None and summary[0] == "text", (
            "the 0017 table contract 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"
        )
    finally:
        _clear_rows(db)
    # Repair: the fixture teardown re-upgrades to head.


def test_upgrade_round_trip_restores_the_flag(db: Session, alembic: Config) -> None:
    """Downgrade to 0017, then upgrade back to 0018: the column is back
    with the full contract (BOOLEAN NOT NULL, the `false` default)."""
    command.downgrade(alembic, "0017")
    command.upgrade(alembic, "0018")
    assert _version(db) == "0018", "round-trip upgrade must land at 0018"

    flag = _column(db, "folder_summaries", "manually_edited")
    assert flag is not None, "folder_summaries.manually_edited must be back"
    assert flag[0] == "boolean", "manually_edited must be BOOLEAN after the round-trip"
    assert flag[1] == "NO", "manually_edited must be NOT NULL after the round-trip"
    assert flag[2] is not None and "false" in str(flag[2]), (
        "the `false` server default must survive the round-trip"
    )