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

"""Tests for migration 0006: Journal Quality System.

This migration consolidates what were originally five separate revisions
(0006-0010) into a single atomic change. It creates the ``papers`` and
``paper_appearances`` tables, adds the ``name_lower`` / ``score_source`` /
``quality_model`` columns + indexes to ``journals``, and adds the
``ix_research_resources_research_id`` FK index.

Tests cover:
- New table creation (``papers``, ``paper_appearances``) with the
  correct columns, uniques, and FK cascade actions.
- New column creation on ``journals`` with ``name_lower`` backfill
  (Python ``str.lower`` on existing rows).
- Named index creation and idempotency.
- Upgrade → downgrade → upgrade roundtrip (schema restored, no
  leftover objects).
- Timestamp columns use ``UtcDateTime`` with ``utcnow()`` server
  defaults (enforced by the pre-commit hook but worth a runtime
  check in case the hook is bypassed).
"""

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

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


# --------------------------------------------------------------------------- #
# Helpers                                                                      #
# --------------------------------------------------------------------------- #


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 _table_exists(engine, name):
    return inspect(engine).has_table(name)


def _get_columns(engine, table_name):
    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):
    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
    }


def _get_unique_column_sets(engine, table_name):
    insp = inspect(engine)
    if not insp.has_table(table_name):
        return []
    return [
        set(u["column_names"]) for u in insp.get_unique_constraints(table_name)
    ]


def _get_fk_ondelete(engine, table_name):
    """Return {column_name: ondelete_action} for all FKs in the table."""
    insp = inspect(engine)
    if not insp.has_table(table_name):
        return {}
    out = {}
    for fk in insp.get_foreign_keys(table_name):
        cols = fk.get("constrained_columns") or []
        if not cols:
            continue
        ondelete = (fk.get("options") or {}).get("ondelete")
        out[cols[0]] = ondelete
    return out


# --------------------------------------------------------------------------- #
# Fixtures                                                                     #
# --------------------------------------------------------------------------- #


@pytest.fixture
def fresh_engine(tmp_path):
    """Brand-new SQLite database — no migrations applied yet."""
    db_path = tmp_path / "fresh_0006_test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    yield engine
    engine.dispose()


@pytest.fixture
def migrated_to_0005_engine(tmp_path):
    """Database stamped at 0005 (just before the journal-quality schema)."""
    db_path = tmp_path / "migrated_0005_test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    _run_upgrade_to(engine, "0005")
    yield engine
    engine.dispose()


@pytest.fixture
def fully_migrated_engine(tmp_path):
    """Database stamped at head (includes 0006)."""
    db_path = tmp_path / "fully_migrated_0006_test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    run_migrations(engine)
    yield engine
    engine.dispose()


# --------------------------------------------------------------------------- #
# Tests — new tables                                                           #
# --------------------------------------------------------------------------- #


class TestMigration0006PapersTable:
    """Creation of the ``papers`` table with the right shape."""

    def test_papers_table_exists(self, fully_migrated_engine):
        assert _table_exists(fully_migrated_engine, "papers")

    def test_papers_has_expected_columns(self, fully_migrated_engine):
        cols = _get_columns(fully_migrated_engine, "papers")
        expected = {
            "id",
            "doi",
            "arxiv_id",
            "pmid",
            "journal_id",
            "container_title",
            "year",
            "metadata",
            "created_at",
            "updated_at",
        }
        assert expected.issubset(set(cols))

    def test_papers_has_no_journal_quality_column(self, fully_migrated_engine):
        """Quality is resolved live (journals.quality + bundled ref DB)
        so there is no frozen per-Paper ``journal_quality`` column —
        re-adding it would re-introduce the staleness footgun the
        migration design explicitly avoids.
        """
        cols = _get_columns(fully_migrated_engine, "papers")
        assert "journal_quality" not in cols

    def test_papers_year_is_indexed(self, fully_migrated_engine):
        """Year is promoted out of the metadata JSON blob to a
        first-class integer column so the dashboard can filter/group
        by year without paying for json_extract on every row.
        """
        indexes = _get_indexes_by_name(fully_migrated_engine, "papers")
        assert "idx_papers_year" in indexes
        assert indexes["idx_papers_year"] == ["year"]

    def test_papers_year_nullable(self, fully_migrated_engine):
        """Year is nullable — many sources lack a publication year."""
        cols = _get_columns(fully_migrated_engine, "papers")
        assert cols["year"]["nullable"] is True

    def test_container_title_nullable(self, fully_migrated_engine):
        """container_title is nullable — populated by the write path
        only when the filter scored the result, never required.
        """
        cols = _get_columns(fully_migrated_engine, "papers")
        assert cols["container_title"]["nullable"] is True

    def test_identifier_columns_are_unique(self, fully_migrated_engine):
        """doi / arxiv_id / pmid each carry a single-column UNIQUE
        constraint — that's the mechanism dedup relies on.
        """
        unique_sets = _get_unique_column_sets(fully_migrated_engine, "papers")
        assert {"doi"} in unique_sets
        assert {"arxiv_id"} in unique_sets
        assert {"pmid"} in unique_sets

    def test_identifier_columns_are_nullable(self, fully_migrated_engine):
        """Papers without identifiers are still inserted — the UNIQUE
        constraint must tolerate multiple NULLs (standard SQLite behavior).
        """
        cols = _get_columns(fully_migrated_engine, "papers")
        assert cols["doi"]["nullable"] is True
        assert cols["arxiv_id"]["nullable"] is True
        assert cols["pmid"]["nullable"] is True

    def test_journal_id_has_fk_with_set_null(self, fully_migrated_engine):
        """journal_id → journals.id with ON DELETE SET NULL.

        Deleting a journal leaves orphan Paper rows rather than cascading
        the delete — paper provenance is worth preserving even if the
        upstream journal is removed.
        """
        fks = _get_fk_ondelete(fully_migrated_engine, "papers")
        assert fks.get("journal_id", "").upper() == "SET NULL"

    def test_idx_papers_journal_exists(self, fully_migrated_engine):
        indexes = _get_indexes_by_name(fully_migrated_engine, "papers")
        assert "idx_papers_journal" in indexes
        assert indexes["idx_papers_journal"] == ["journal_id"]

    def test_idx_papers_container_title_exists(self, fully_migrated_engine):
        """Dashboard GROUP BY container_title needs an index to be cheap."""
        indexes = _get_indexes_by_name(fully_migrated_engine, "papers")
        assert "idx_papers_container_title" in indexes
        assert indexes["idx_papers_container_title"] == ["container_title"]


class TestMigration0006PaperAppearancesTable:
    """Creation of the ``paper_appearances`` join table."""

    def test_paper_appearances_table_exists(self, fully_migrated_engine):
        assert _table_exists(fully_migrated_engine, "paper_appearances")

    def test_paper_appearances_has_expected_columns(
        self, fully_migrated_engine
    ):
        cols = _get_columns(fully_migrated_engine, "paper_appearances")
        expected = {
            "id",
            "paper_id",
            "resource_id",
            "source_engine",
            "created_at",
        }
        assert expected.issubset(set(cols))

    def test_resource_id_is_unique(self, fully_migrated_engine):
        """One resource appears in exactly one paper_appearance row —
        enforced at the schema level so dedup bugs can't double-count.
        """
        unique_sets = _get_unique_column_sets(
            fully_migrated_engine, "paper_appearances"
        )
        assert {"resource_id"} in unique_sets

    def test_fks_use_cascade_ondelete(self, fully_migrated_engine):
        """Both FKs on paper_appearances cascade: deleting a paper or the
        research resource must also drop the join row so we don't accrue
        dangling rows pointing at missing parents.
        """
        fks = _get_fk_ondelete(fully_migrated_engine, "paper_appearances")
        assert fks.get("paper_id", "").upper() == "CASCADE"
        assert fks.get("resource_id", "").upper() == "CASCADE"

    def test_paper_id_is_indexed(self, fully_migrated_engine):
        """An index on paper_id must exist for the papers→appearances
        join hot path. The exact index name depends on who wins the
        race between alembic's named index and any ORM bootstrap that
        also declares a backing index, so we check by column coverage
        rather than by name.
        """
        indexes = _get_indexes_by_name(
            fully_migrated_engine, "paper_appearances"
        )
        assert any(cols == ["paper_id"] for cols in indexes.values()), (
            f"no index covering paper_id found; got {indexes}"
        )

    def test_paper_appearances_paper_id_indexed_after_migration(
        self, fully_migrated_engine
    ):
        """The migration path must produce the named index
        ``ix_paper_appearances_paper_id``. create_all() produces the
        same name (from ``index=True`` on citation.py:159), so both
        paths agree. A missing named index here means alembic-only
        installs are running full scans on the paper→appearance join.
        """
        indexes = _get_indexes_by_name(
            fully_migrated_engine, "paper_appearances"
        )
        assert "ix_paper_appearances_paper_id" in indexes, (
            f"named index ix_paper_appearances_paper_id missing from "
            f"migration path; got {list(indexes.keys())}"
        )
        assert indexes["ix_paper_appearances_paper_id"] == ["paper_id"]


# --------------------------------------------------------------------------- #
# Tests — journals columns + name_lower backfill                               #
# --------------------------------------------------------------------------- #


class TestMigration0006JournalsColumns:
    """Column additions and the ``name_lower`` backfill."""

    def test_new_columns_present(self, fully_migrated_engine):
        cols = _get_columns(fully_migrated_engine, "journals")
        assert "name_lower" in cols
        assert "score_source" in cols
        assert "quality_model" in cols

    def test_indexes_present(self, fully_migrated_engine):
        """0006 creates ``ix_journals_quality_model``. It does NOT
        create a non-unique index on ``name_lower``: the
        ``uq_journals_name_lower`` UNIQUE constraint already provides
        the backing index and a second B-tree on the same column
        would be pure write-amplification.
        """
        indexes = _get_indexes_by_name(fully_migrated_engine, "journals")
        assert "ix_journals_quality_model" in indexes
        assert "ix_journals_name_lower" not in indexes, (
            "ix_journals_name_lower is redundant with uq_journals_name_lower"
            " — must not be created by 0006."
        )
        assert "ix_journals_name" not in indexes, (
            "ix_journals_name is redundant with UNIQUE on Journal.name"
            " — must not be created via model.create_all() either."
        )

    def test_name_lower_backfill_preserves_diacritics(
        self, migrated_to_0005_engine
    ):
        """Python ``str.lower`` on an existing row should produce the
        same normalized form the runtime insert path emits — diacritics
        must survive unchanged.
        """
        engine = migrated_to_0005_engine
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO journals (name, quality, quality_analysis_time) "
                    "VALUES ('Café Scientifique', 5, 0)"
                )
            )

        _run_upgrade_to(engine, "0006")

        with engine.connect() as conn:
            row = conn.execute(
                text(
                    "SELECT name_lower FROM journals "
                    "WHERE name = 'Café Scientifique'"
                )
            ).fetchone()

        assert row is not None
        assert row[0] == "café scientifique"

    def test_name_lower_unique_constraint_exists(self, fully_migrated_engine):
        """UNIQUE on ``name_lower`` is the defence against two rows with
        different-cased ``name`` values splitting the journal cache —
        e.g. ``"Nature Medicine"`` vs ``"NATURE MEDICINE"`` both passing
        the ``name`` UNIQUE check while agreeing on ``name_lower``."""
        unique_sets = _get_unique_column_sets(fully_migrated_engine, "journals")
        assert {"name_lower"} in unique_sets

    def test_migration_dedupes_name_lower_collisions_before_unique(
        self, migrated_to_0005_engine
    ):
        """If pre-0006 rows collide on ``name_lower`` (possible because
        0005's schema had no UNIQUE constraint), the migration must
        dedupe before adding the new UNIQUE constraint, otherwise the
        ALTER TABLE would fail. The surviving row is the HIGHEST-
        quality one — the best LLM verdict wins — with ties broken by
        lowest id.
        """
        engine = migrated_to_0005_engine
        with engine.begin() as conn:
            # Pre-0006 schema has no name_lower column yet, so collisions
            # emerge after backfill. Insert two rows with different-
            # cased ``name`` values — both pass the ``name`` UNIQUE but
            # will produce the same name_lower.
            conn.execute(
                text(
                    "INSERT INTO journals (name, quality, quality_analysis_time) "
                    "VALUES ('Nature Medicine', 8, 1000)"
                )
            )
            conn.execute(
                text(
                    "INSERT INTO journals (name, quality, quality_analysis_time) "
                    "VALUES ('NATURE MEDICINE', 9, 2000)"
                )
            )

        # Run 0006 — backfill + dedupe + UNIQUE.
        _run_upgrade_to(engine, "0006")

        with engine.connect() as conn:
            rows = conn.execute(
                text(
                    "SELECT id, name, name_lower, quality FROM journals "
                    "WHERE name_lower = 'nature medicine' ORDER BY id"
                )
            ).fetchall()

        assert len(rows) == 1, (
            f"Expected dedupe to leave exactly 1 row, got {len(rows)}: {rows}"
        )
        # Highest quality wins — that's the second insert (quality=9).
        assert rows[0][1] == "NATURE MEDICINE"
        assert rows[0][3] == 9

    def test_dedupe_prefers_highest_quality_across_nfkc_variants(
        self, migrated_to_0005_engine
    ):
        """Case-fold dedupe must ALSO collapse NFKC compatibility
        variants (e.g. "Physics Letters™" vs "Physics Letters TM"),
        and the surviving row must be the highest-quality one.
        """
        engine = migrated_to_0005_engine
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO journals (name, quality, quality_analysis_time) "
                    "VALUES ('Physics Letters\u2122', 5, 1000)"
                )
            )
            conn.execute(
                text(
                    "INSERT INTO journals (name, quality, quality_analysis_time) "
                    "VALUES ('physics lettersTM', 8, 2000)"
                )
            )

        _run_upgrade_to(engine, "0006")

        with engine.connect() as conn:
            rows = conn.execute(
                text(
                    "SELECT name, name_lower, quality FROM journals "
                    "WHERE name_lower = 'physics letterstm'"
                )
            ).fetchall()

        assert len(rows) == 1, f"Expected 1 row after NFKC dedupe, got {rows}"
        assert rows[0][2] == 8

    def test_backfill_nfkc_roundtrip(self, migrated_to_0005_engine):
        """Backfilled ``name_lower`` must match scoring.normalize_name
        output. U+2122 (™) NFKC-decomposes to "TM"; bare .lower() would
        leave it intact. This test locks NFKC semantics against silent
        regression to bare lowercase.
        """
        engine = migrated_to_0005_engine
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO journals (name, quality, quality_analysis_time) "
                    "VALUES ('Physics Letters\u2122', 7, 1000)"
                )
            )

        _run_upgrade_to(engine, "0006")

        with engine.connect() as conn:
            row = conn.execute(
                text("SELECT name, name_lower FROM journals WHERE quality = 7")
            ).fetchone()

        # Expected: NFKC(Physics Letters™) → "Physics LettersTM" → lower → "physics letterstm"
        assert row[1] == "physics letterstm", (
            f"Expected NFKC-normalized name_lower, got {row[1]!r}"
        )

    # NOTE: no test for name=NULL rows because the 0005 schema has
    # NOT NULL on journals.name (Journal model, journal.py:37). The
    # ``row.name is not None`` skip in the migration's dedupe is
    # defensive against a scenario the schema disallows; we cannot
    # seed it in a regression test.


# --------------------------------------------------------------------------- #
# Tests — research_resources index                                             #
# --------------------------------------------------------------------------- #


class TestMigration0006ResearchResourcesIndex:
    def test_research_id_index_exists(self, fully_migrated_engine):
        indexes = _get_indexes_by_name(
            fully_migrated_engine, "research_resources"
        )
        assert "ix_research_resources_research_id" in indexes
        assert indexes["ix_research_resources_research_id"] == ["research_id"]

    def test_research_resources_index_present_via_create_all(
        self, fresh_engine
    ):
        """The model MUST declare the index (via __table_args__) so a
        fresh ``Base.metadata.create_all()`` path produces the same
        ``ix_research_resources_research_id`` as the migration path.
        Without this, create_all-style installs (dev setups, test
        fixtures bypassing migrations) have no research_id index and
        run full-table scans on 20+ call sites.
        """
        from local_deep_research.database.models.base import Base

        Base.metadata.create_all(fresh_engine)
        indexes = _get_indexes_by_name(fresh_engine, "research_resources")
        assert "ix_research_resources_research_id" in indexes, (
            f"named index missing after create_all; got {list(indexes.keys())}"
        )
        assert indexes["ix_research_resources_research_id"] == ["research_id"]


# --------------------------------------------------------------------------- #
# Tests — idempotency + roundtrip                                              #
# --------------------------------------------------------------------------- #


class TestMigration0006Idempotency:
    """Running the migration twice must be a no-op (no errors, same
    schema). The migration uses ``_table_exists`` / ``_column_exists`` /
    ``_index_exists`` guards for exactly this reason.
    """

    def test_double_migrate_no_error(self, fresh_engine):
        run_migrations(fresh_engine)
        # Second run should succeed silently.
        run_migrations(fresh_engine)
        head = get_head_revision()
        assert head is not None and head.isdigit() and len(head) == 4

    def test_rerun_recreates_dropped_papers_indexes(
        self, migrated_to_0005_engine
    ):
        """Regression: if ``papers`` exists but one of its named indexes
        was dropped (partial migration, manual intervention, etc.), a
        rerun of 0006 must recreate the missing index rather than
        silently skip it. Prior behaviour gated all three
        ``idx_papers_*`` creations on ``if not _table_exists('papers')``,
        so the rerun path could never converge.
        """
        engine = migrated_to_0005_engine
        _run_upgrade_to(engine, "0006")

        # Simulate drift: drop one of the named indexes.
        with engine.begin() as conn:
            conn.execute(text("DROP INDEX idx_papers_year"))
        assert "idx_papers_year" not in _get_indexes_by_name(engine, "papers")

        # Stamp back to 0005 without touching the schema, then re-upgrade.
        # The table still exists, so the rerun must hit the index-guard path.
        config = get_alembic_config(engine)
        with engine.begin() as conn:
            config.attributes["connection"] = conn
            command.stamp(config, "0005")
        _run_upgrade_to(engine, "0006")

        indexes = _get_indexes_by_name(engine, "papers")
        assert "idx_papers_year" in indexes
        assert indexes["idx_papers_year"] == ["year"]
        # The other two indexes must also still be present (not re-created
        # twice, not dropped).
        assert "idx_papers_journal" in indexes
        assert "idx_papers_container_title" in indexes


class TestMigration0006Roundtrip:
    """Upgrade 0005 → 0006 → downgrade → upgrade again.

    Downgrade must remove every object the upgrade created; the second
    upgrade must recreate them without error. This is the cheapest way
    to catch missing drops in ``downgrade()``.
    """

    def test_downgrade_then_upgrade_restores_schema(
        self, migrated_to_0005_engine
    ):
        engine = migrated_to_0005_engine

        _run_upgrade_to(engine, "0006")
        assert _table_exists(engine, "papers")
        assert _table_exists(engine, "paper_appearances")
        assert "name_lower" in _get_columns(engine, "journals")

        _run_downgrade_to(engine, "0005")
        assert not _table_exists(engine, "papers")
        assert not _table_exists(engine, "paper_appearances")
        assert "name_lower" not in _get_columns(engine, "journals")
        assert "score_source" not in _get_columns(engine, "journals")
        assert "quality_model" not in _get_columns(engine, "journals")
        # Index on research_resources is dropped too.
        rr_indexes = _get_indexes_by_name(engine, "research_resources")
        assert "ix_research_resources_research_id" not in rr_indexes

        _run_upgrade_to(engine, "0006")
        assert _table_exists(engine, "papers")
        assert _table_exists(engine, "paper_appearances")
        assert "name_lower" in _get_columns(engine, "journals")

    def test_downgrade_preserves_journals_data(self, migrated_to_0005_engine):
        """The 0005 baseline columns (id, name, quality,
        quality_analysis_time) must survive a downgrade. Only the three
        columns added by 0006 (name_lower, score_source, quality_model)
        are dropped. The downgrade docstring promises this; this test
        enforces it.
        """
        engine = migrated_to_0005_engine
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO journals (name, quality, quality_analysis_time) "
                    "VALUES ('Nature', 10, 1234567890)"
                )
            )

        _run_upgrade_to(engine, "0006")
        _run_downgrade_to(engine, "0005")

        with engine.connect() as conn:
            row = conn.execute(
                text(
                    "SELECT name, quality, quality_analysis_time "
                    "FROM journals WHERE name = 'Nature'"
                )
            ).fetchone()

        assert row is not None, "journals row lost on downgrade"
        assert row[0] == "Nature"
        assert row[1] == 10
        assert row[2] == 1234567890