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

"""
Tests for migration 0005: Add document_id column to research_resources.

Tests cover:
- Column creation with correct type and nullability
- Index creation (ix_research_resources_document_id)
- Full migration chain from empty database
- Idempotency (running migrations multiple times)
- Downgrade behavior (column and index removal)
- Data preservation during upgrade and downgrade
- Edge cases (missing table, pre-existing column, in-memory database)
"""

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

from local_deep_research.database.alembic_runner import (
    get_alembic_config,
    get_current_revision,
    get_head_revision,
    needs_migration,
    run_migrations,
)


def _run_upgrade_to(engine, revision):
    """Run migrations up to a specific 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):
    """Run downgrade to a specific revision."""
    config = get_alembic_config(engine)
    with engine.begin() as conn:
        config.attributes["connection"] = conn
        command.downgrade(config, revision)


def _get_columns(engine, table_name):
    """Get a dict of {column_name: column_info} for a table."""
    insp = inspect(engine)
    if not insp.has_table(table_name):
        return {}
    return {col["name"]: col for col in insp.get_columns(table_name)}


def _get_indexes_by_name(engine, table_name):
    """Get a dict of {index_name: [column_names]} for a table."""
    insp = inspect(engine)
    if not insp.has_table(table_name):
        return {}
    return {
        idx["name"]: idx["column_names"]
        for idx in insp.get_indexes(table_name)
        if idx["name"] is not None
    }


@pytest.fixture
def fresh_engine(tmp_path):
    """Create a fresh SQLite engine (empty database, no tables)."""
    db_path = tmp_path / "fresh_0005_test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    yield engine
    engine.dispose()


@pytest.fixture
def migrated_to_0004_engine(tmp_path):
    """Create a database migrated to revision 0004 (before document_id)."""
    db_path = tmp_path / "migrated_0004_test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    _run_upgrade_to(engine, "0004")
    yield engine
    engine.dispose()


@pytest.fixture
def fully_migrated_engine(tmp_path):
    """Create a database migrated up to revision 0005 (this file's target).

    Stops at 0005 instead of head — the downgrade tests below would
    otherwise have to roll back through migration 0010, which is
    documented as non-reversible (raises NotImplementedError). Every
    test in this file is scoped to 0005 behaviour.
    """
    db_path = tmp_path / "fully_migrated_0005_test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    _run_upgrade_to(engine, "0005")
    yield engine
    engine.dispose()


class TestMigration0005UpgradeColumn:
    """Tests that verify document_id column creation on upgrade."""

    def test_document_id_column_exists(self, fully_migrated_engine):
        """document_id column should exist on research_resources after migration."""
        columns = _get_columns(fully_migrated_engine, "research_resources")
        assert "document_id" in columns

    def test_document_id_column_is_nullable(self, fully_migrated_engine):
        """document_id should be nullable (existing rows have no value)."""
        columns = _get_columns(fully_migrated_engine, "research_resources")
        assert columns["document_id"]["nullable"] is True

    def test_document_id_column_type_is_varchar(self, fully_migrated_engine):
        """document_id should be VARCHAR(36) to hold UUIDs."""
        columns = _get_columns(fully_migrated_engine, "research_resources")
        col_type = str(columns["document_id"]["type"])
        assert "VARCHAR" in col_type or "CHAR" in col_type

    def test_document_id_index_exists(self, fully_migrated_engine):
        """Index ix_research_resources_document_id should exist."""
        indexes = _get_indexes_by_name(
            fully_migrated_engine, "research_resources"
        )
        assert "ix_research_resources_document_id" in indexes
        assert indexes["ix_research_resources_document_id"] == ["document_id"]

    def test_document_id_defaults_to_null(self, fully_migrated_engine):
        """Inserting a row without document_id should default to NULL."""
        engine = fully_migrated_engine
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO research_resources "
                    "(research_id, title, url, created_at) "
                    "VALUES ('test-rh-1', 'Test Resource', "
                    "'https://example.com', '2026-01-01')"
                )
            )

        with engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT document_id FROM research_resources "
                    "WHERE research_id = 'test-rh-1'"
                )
            ).fetchone()
            assert result[0] is None

    def test_document_id_can_store_uuid(self, fully_migrated_engine):
        """document_id should accept a UUID string value."""
        engine = fully_migrated_engine
        test_uuid = "abcdef01-2345-6789-abcd-ef0123456789"

        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO research_resources "
                    "(research_id, title, url, created_at, document_id) "
                    "VALUES ('test-rh-2', 'Linked Resource', "
                    "'https://example.com', '2026-01-01', :doc_id)"
                ),
                {"doc_id": test_uuid},
            )

        with engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT document_id FROM research_resources "
                    "WHERE research_id = 'test-rh-2'"
                )
            ).fetchone()
            assert result[0] == test_uuid


class TestMigration0005UpgradeFromPrior:
    """Tests that verify the upgrade path for databases missing document_id.

    Migration 0001 uses Base.metadata.create_all() which includes document_id
    from the current model. To test the real-world scenario (database created
    before document_id was added to the model), we manually create the table
    WITHOUT document_id and stamp at 0004.
    """

    @pytest.fixture
    def legacy_engine(self, tmp_path):
        """Create a database with research_resources missing document_id.

        This simulates a database created before commit 2033f977e added
        document_id to the ResearchResource model.
        """
        db_path = tmp_path / "legacy_no_docid.db"
        engine = create_engine(f"sqlite:///{db_path}")

        with engine.begin() as conn:
            # Create research_history (needed for FK)
            conn.execute(
                text(
                    "CREATE TABLE research_history ("
                    "id VARCHAR(36) PRIMARY KEY, "
                    "query TEXT NOT NULL, "
                    "mode TEXT NOT NULL, "
                    "status TEXT NOT NULL, "
                    "created_at TEXT NOT NULL"
                    ")"
                )
            )
            # Create research_resources WITHOUT document_id
            conn.execute(
                text(
                    "CREATE TABLE research_resources ("
                    "id INTEGER PRIMARY KEY AUTOINCREMENT, "
                    "research_id VARCHAR(36) NOT NULL "
                    "  REFERENCES research_history(id) ON DELETE CASCADE, "
                    "title TEXT, "
                    "url TEXT, "
                    "content_preview TEXT, "
                    "source_type TEXT, "
                    "metadata JSON, "
                    "created_at VARCHAR NOT NULL"
                    ")"
                )
            )
            # Stamp at 0004 so migration 0005 runs
            conn.execute(
                text(
                    "CREATE TABLE alembic_version "
                    "(version_num VARCHAR(32) NOT NULL)"
                )
            )
            conn.execute(text("INSERT INTO alembic_version VALUES ('0004')"))

        yield engine
        engine.dispose()

    def test_upgrade_adds_column_to_legacy_table(self, legacy_engine):
        """Upgrading a database missing document_id should add the column."""
        columns = _get_columns(legacy_engine, "research_resources")
        assert "document_id" not in columns

        _run_upgrade_to(legacy_engine, "0005")

        columns = _get_columns(legacy_engine, "research_resources")
        assert "document_id" in columns

    def test_upgrade_adds_index_to_legacy_table(self, legacy_engine):
        """Upgrading should create the document_id index."""
        _run_upgrade_to(legacy_engine, "0005")

        indexes = _get_indexes_by_name(legacy_engine, "research_resources")
        assert "ix_research_resources_document_id" in indexes

    def test_revision_is_0005_after_upgrade(self, legacy_engine):
        """Current revision should be 0005 after targeted upgrade."""
        _run_upgrade_to(legacy_engine, "0005")
        assert get_current_revision(legacy_engine) == "0005"

    def test_existing_data_preserved_after_upgrade(self, legacy_engine):
        """Rows inserted before upgrade should survive with NULL document_id."""
        with legacy_engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO research_history "
                    "(id, query, mode, status, created_at) "
                    "VALUES ('rh-legacy', 'old query', 'quick', "
                    "'completed', '2025-06-01')"
                )
            )
            conn.execute(
                text(
                    "INSERT INTO research_resources "
                    "(research_id, title, url, created_at) "
                    "VALUES ('rh-legacy', 'Old Resource', "
                    "'https://old.com', '2025-06-01')"
                )
            )

        _run_upgrade_to(legacy_engine, "0005")

        with legacy_engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT title, url, document_id "
                    "FROM research_resources "
                    "WHERE research_id = 'rh-legacy'"
                )
            ).fetchone()
            assert result[0] == "Old Resource"
            assert result[1] == "https://old.com"
            assert result[2] is None

    def test_idempotent_on_fresh_db(self, migrated_to_0004_engine):
        """On a fresh DB (0001 already created document_id), 0005 is a no-op."""
        engine = migrated_to_0004_engine

        # 0001's create_all already added document_id from current model
        columns_before = _get_columns(engine, "research_resources")
        assert "document_id" in columns_before

        _run_upgrade_to(engine, "0005")

        columns_after = _get_columns(engine, "research_resources")
        assert "document_id" in columns_after
        assert get_current_revision(engine) == "0005"


class TestMigration0005FromFreshDatabase:
    """Tests that verify the full migration chain on a fresh database."""

    def test_fresh_db_full_migration_creates_column(self, fresh_engine):
        """Running all migrations on empty DB should create document_id."""
        run_migrations(fresh_engine)

        columns = _get_columns(fresh_engine, "research_resources")
        assert "document_id" in columns

    def test_head_revision_is_real_id(self):
        """get_head_revision() returns a real 4-digit revision id."""
        head = get_head_revision()
        assert head is not None and head.isdigit() and len(head) == 4

    def test_current_revision_is_head_after_migrate(self, fresh_engine):
        """After full migration, current revision should match head."""
        run_migrations(fresh_engine)
        assert get_current_revision(fresh_engine) == get_head_revision()

    def test_needs_migration_false_after_full_upgrade(self, fresh_engine):
        """After full migration, needs_migration() should return False."""
        run_migrations(fresh_engine)
        assert needs_migration(fresh_engine) is False


class TestMigration0005Idempotency:
    """Tests that verify safe re-runs of migration."""

    def test_run_migrations_twice_no_error(self, fresh_engine):
        """Calling run_migrations() twice should not raise."""
        run_migrations(fresh_engine)
        run_migrations(fresh_engine)

    def test_column_unchanged_after_double_migration(self, fresh_engine):
        """Column should be identical after running migrations twice."""
        run_migrations(fresh_engine)
        columns_first = _get_columns(fresh_engine, "research_resources")

        run_migrations(fresh_engine)
        columns_second = _get_columns(fresh_engine, "research_resources")

        assert ("document_id" in columns_first) == (
            "document_id" in columns_second
        )

    def test_pre_existing_column_not_duplicated(self, tmp_path):
        """If document_id already exists (fresh DB), migration is a no-op."""
        db_path = tmp_path / "pre_existing_col.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            # Full migration creates the column via 0001's create_all
            run_migrations(engine)

            columns_before = _get_columns(engine, "research_resources")
            assert "document_id" in columns_before

            # Downgrade to 0004 and re-upgrade to test the migration path
            # when the column was already added by create_all
            # (This tests the idempotency guard)
            col_count_before = len(columns_before)

            # Running migrations again should be a no-op
            run_migrations(engine)

            columns_after = _get_columns(engine, "research_resources")
            assert len(columns_after) == col_count_before
        finally:
            engine.dispose()


class TestMigration0005Downgrade:
    """Tests for rollback behavior."""

    def test_downgrade_to_0004_removes_column(self, fully_migrated_engine):
        """Downgrade from 0005 to 0004 should remove document_id."""
        _run_downgrade_to(fully_migrated_engine, "0004")

        columns = _get_columns(fully_migrated_engine, "research_resources")
        assert "document_id" not in columns

    def test_downgrade_to_0004_removes_index(self, fully_migrated_engine):
        """Downgrade should remove ix_research_resources_document_id."""
        _run_downgrade_to(fully_migrated_engine, "0004")

        indexes = _get_indexes_by_name(
            fully_migrated_engine, "research_resources"
        )
        assert "ix_research_resources_document_id" not in indexes

    def test_downgrade_preserves_table(self, fully_migrated_engine):
        """research_resources table should still exist after downgrade."""
        _run_downgrade_to(fully_migrated_engine, "0004")

        insp = inspect(fully_migrated_engine)
        assert insp.has_table("research_resources")

    def test_downgrade_preserves_other_columns(self, fully_migrated_engine):
        """Other columns (title, url, etc.) should survive downgrade."""
        _run_downgrade_to(fully_migrated_engine, "0004")

        columns = _get_columns(fully_migrated_engine, "research_resources")
        for col in ("id", "research_id", "title", "url", "created_at"):
            assert col in columns, f"Column {col} lost during downgrade"

    def test_downgrade_then_upgrade_roundtrip(self, fully_migrated_engine):
        """Downgrade to 0004 then upgrade back to 0005 should restore column."""
        engine = fully_migrated_engine

        _run_downgrade_to(engine, "0004")
        assert get_current_revision(engine) == "0004"

        _run_upgrade_to(engine, "0005")
        assert get_current_revision(engine) == "0005"

        columns = _get_columns(engine, "research_resources")
        assert "document_id" in columns

        indexes = _get_indexes_by_name(engine, "research_resources")
        assert "ix_research_resources_document_id" in indexes


class TestMigration0005DataPreservation:
    """Tests that verify migration is non-destructive."""

    def test_existing_rows_preserved_after_upgrade(
        self, migrated_to_0004_engine
    ):
        """Data in research_resources should survive the migration."""
        engine = migrated_to_0004_engine

        # Insert a research_history row first (FK constraint)
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO research_history "
                    "(id, query, mode, status, created_at) "
                    "VALUES ('rh-preserve', 'test query', 'quick', "
                    "'completed', '2026-01-01')"
                )
            )
            conn.execute(
                text(
                    "INSERT INTO research_resources "
                    "(research_id, title, url, created_at) "
                    "VALUES ('rh-preserve', 'Preserved Resource', "
                    "'https://example.com', '2026-01-01')"
                )
            )

        _run_upgrade_to(engine, "0005")

        with engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT title, url, document_id "
                    "FROM research_resources "
                    "WHERE research_id = 'rh-preserve'"
                )
            ).fetchone()
            assert result is not None
            assert result[0] == "Preserved Resource"
            assert result[1] == "https://example.com"
            assert result[2] is None  # New column defaults to NULL

    def test_data_preserved_after_downgrade(self, fully_migrated_engine):
        """Non-document_id data should survive downgrade."""
        engine = fully_migrated_engine

        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO research_history "
                    "(id, query, mode, status, created_at) "
                    "VALUES ('rh-down', 'downgrade query', 'detailed', "
                    "'completed', '2026-01-01')"
                )
            )
            conn.execute(
                text(
                    "INSERT INTO research_resources "
                    "(research_id, title, url, created_at, document_id) "
                    "VALUES ('rh-down', 'Will Lose DocID', "
                    "'https://example.com', '2026-01-01', 'some-uuid')"
                )
            )

        _run_downgrade_to(engine, "0004")

        with engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT title, url FROM research_resources "
                    "WHERE research_id = 'rh-down'"
                )
            ).fetchone()
            assert result is not None
            assert result[0] == "Will Lose DocID"
            assert result[1] == "https://example.com"


class TestMigration0005EdgeCases:
    """Tests for edge cases and robustness."""

    def test_missing_research_resources_table(self, tmp_path):
        """Migration should be a no-op when research_resources doesn't exist."""
        db_path = tmp_path / "no_resources_table.db"
        engine = create_engine(f"sqlite:///{db_path}")

        try:
            with engine.begin() as conn:
                conn.execute(
                    text(
                        "CREATE TABLE alembic_version "
                        "(version_num VARCHAR(32) NOT NULL)"
                    )
                )
                conn.execute(
                    text("INSERT INTO alembic_version VALUES ('0004')")
                )

            _run_upgrade_to(engine, "0005")
            assert get_current_revision(engine) == "0005"

            insp = inspect(engine)
            assert not insp.has_table("research_resources")
        finally:
            engine.dispose()

    def test_in_memory_database(self):
        """Migration should work on in-memory SQLite database."""
        engine = create_engine("sqlite:///:memory:")

        try:
            run_migrations(engine)

            assert get_current_revision(engine) == get_head_revision()

            columns = _get_columns(engine, "research_resources")
            assert "document_id" in columns

            indexes = _get_indexes_by_name(engine, "research_resources")
            assert "ix_research_resources_document_id" in indexes
        finally:
            engine.dispose()

    def test_document_id_queryable_after_upgrade(self, migrated_to_0004_engine):
        """Queries filtering on document_id should work after upgrade."""
        engine = migrated_to_0004_engine

        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO research_history "
                    "(id, query, mode, status, created_at) "
                    "VALUES ('rh-query', 'query test', 'quick', "
                    "'completed', '2026-01-01')"
                )
            )
            conn.execute(
                text(
                    "INSERT INTO research_resources "
                    "(research_id, title, url, created_at) "
                    "VALUES ('rh-query', 'Resource A', "
                    "'https://a.com', '2026-01-01')"
                )
            )

        _run_upgrade_to(engine, "0005")

        # Set document_id on the row
        with engine.begin() as conn:
            conn.execute(
                text(
                    "UPDATE research_resources SET document_id = 'doc-uuid-1' "
                    "WHERE research_id = 'rh-query'"
                )
            )

        # Query using the new column (mimics library_service.py join condition)
        with engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT title FROM research_resources "
                    "WHERE document_id = 'doc-uuid-1'"
                )
            ).fetchall()
            assert len(result) == 1
            assert result[0][0] == "Resource A"

            # NULL document_id query
            result = conn.execute(
                text(
                    "SELECT COUNT(*) FROM research_resources "
                    "WHERE document_id IS NULL"
                )
            ).scalar()
            assert result == 0  # We updated the only row