learningcircuit--local-deep-research/tests/database/test_migration_0015_drop_document_notes.py

"""Tests for migration 0015: drop the dead ``documents.notes`` legacy column.

Covers:
- The drop removes ``notes`` from ``documents``.
- Other columns, indexes, and row data on the rebuilt table survive.
- Idempotency (re-run, and tables that never had the column).
- Missing-table guard.
- Downgrade re-adds ``notes`` as nullable Text, and a round-trip is stable.

(The head-alignment guard moved to the newer 0016 migration's test file, per
the convention that it lives in the newest migration's tests.)

Note on test setup
==================
Migration 0001 runs ``Base.metadata.create_all()`` against the CURRENT model,
which (after this PR) no longer has ``notes`` — so a freshly initialised DB
never has the column. To pin "0015 is the migration that drops it" we use the
same pattern as ``test_migration_0005_resource_document_id.py`` /
``test_migration_0014_*``: hand-build a "legacy" ``documents`` table that still
has ``notes``, stamp at ``0014``, and run the upgrade against that.
"""

import pytest
from alembic import command
from sqlalchemy import create_engine, inspect, text

from local_deep_research.database.alembic_runner import (
    get_alembic_config,
    stamp_database,
)


def _run_upgrade_to(engine, revision):
    config = get_alembic_config(engine)
    with engine.begin() as conn:
        config.attributes["connection"] = conn
        command.upgrade(config, revision)


def _run_downgrade_to(engine, revision):
    config = get_alembic_config(engine)
    with engine.begin() as conn:
        config.attributes["connection"] = conn
        command.downgrade(config, revision)


def _columns(engine, table):
    insp = inspect(engine)
    if not insp.has_table(table):
        return set()
    return {c["name"] for c in insp.get_columns(table)}


def _column_info(engine, table):
    insp = inspect(engine)
    if not insp.has_table(table):
        return {}
    return {c["name"]: c for c in insp.get_columns(table)}


def _index_names(engine, table):
    insp = inspect(engine)
    if not insp.has_table(table):
        return set()
    return {idx["name"] for idx in insp.get_indexes(table) if idx["name"]}


@pytest.fixture
def legacy_engine(tmp_path):
    """A DB whose ``documents`` table still has the ``notes`` column.

    Simulates an installation created before 0015. Stamped at 0014 so 0015's
    upgrade is a clean delta. Includes a named index and a seeded row so the
    batch-mode table rebuild can be checked for column/index/data preservation.
    """
    db_path = tmp_path / "legacy_pre_0015.db"
    engine = create_engine(f"sqlite:///{db_path}")

    with engine.begin() as conn:
        conn.execute(
            text(
                "CREATE TABLE documents ("
                "id INTEGER PRIMARY KEY AUTOINCREMENT, "
                "document_hash VARCHAR(64) NOT NULL, "
                "title VARCHAR(500), "
                "favorite BOOLEAN NOT NULL DEFAULT 0, "
                "tags TEXT, "
                "notes TEXT, "
                "created_at TEXT NOT NULL"
                ")"
            )
        )
        conn.execute(
            text("CREATE INDEX idx_document_hash ON documents (document_hash)")
        )
        conn.execute(
            text(
                "INSERT INTO documents "
                "(document_hash, title, favorite, tags, notes, created_at) "
                "VALUES ('hash-1', 'Doc One', 1, '[\"a\"]', "
                "'a stray note', '2026-01-01 00:00:00')"
            )
        )

    stamp_database(engine, "0014")
    yield engine
    engine.dispose()


class TestMigration0015Upgrade:
    def test_drops_notes_column(self, legacy_engine):
        engine = legacy_engine
        assert "notes" in _columns(engine, "documents")
        _run_upgrade_to(engine, "0015")
        assert "notes" not in _columns(engine, "documents")

    def test_preserves_other_columns(self, legacy_engine):
        engine = legacy_engine
        _run_upgrade_to(engine, "0015")
        cols = _columns(engine, "documents")
        for expected in {
            "id",
            "document_hash",
            "title",
            "favorite",
            "tags",
            "created_at",
        }:
            assert expected in cols

    def test_preserves_index(self, legacy_engine):
        engine = legacy_engine
        assert "idx_document_hash" in _index_names(engine, "documents")
        _run_upgrade_to(engine, "0015")
        assert "idx_document_hash" in _index_names(engine, "documents")

    def test_preserves_row_data(self, legacy_engine):
        engine = legacy_engine
        _run_upgrade_to(engine, "0015")
        with engine.begin() as conn:
            row = conn.execute(
                text(
                    "SELECT document_hash, title, favorite, tags "
                    "FROM documents WHERE document_hash = 'hash-1'"
                )
            ).fetchone()
        assert row is not None
        assert row[0] == "hash-1"
        assert row[1] == "Doc One"
        assert row[2] == 1
        assert row[3] == '["a"]'

    def test_upgrade_is_idempotent(self, legacy_engine):
        engine = legacy_engine
        _run_upgrade_to(engine, "0015")
        _run_upgrade_to(engine, "0015")  # second run is a no-op, no error
        assert "notes" not in _columns(engine, "documents")

    def test_missing_table_does_not_crash(self, tmp_path):
        """Upgrade is a no-op when ``documents`` doesn't exist yet."""
        engine = create_engine(f"sqlite:///{tmp_path}/no_documents.db")
        stamp_database(engine, "0014")
        _run_upgrade_to(engine, "0015")  # must not raise
        assert not inspect(engine).has_table("documents")
        engine.dispose()


class TestMigration0015Downgrade:
    def test_downgrade_readds_notes_nullable(self, legacy_engine):
        engine = legacy_engine
        _run_upgrade_to(engine, "0015")
        assert "notes" not in _columns(engine, "documents")
        _run_downgrade_to(engine, "0014")
        cols = _column_info(engine, "documents")
        assert "notes" in cols
        assert cols["notes"]["nullable"] is True
        assert "TEXT" in str(cols["notes"]["type"]).upper()

    def test_downgrade_then_upgrade_roundtrip(self, legacy_engine):
        engine = legacy_engine
        _run_upgrade_to(engine, "0015")
        _run_downgrade_to(engine, "0014")
        _run_upgrade_to(engine, "0015")
        assert "notes" not in _columns(engine, "documents")
        # Seeded row still present after the round-trip.
        with engine.begin() as conn:
            count = conn.execute(
                text("SELECT COUNT(*) FROM documents")
            ).scalar()
        assert count == 1