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

"""Tests for migration 0009: default search.fetch.mode 'full' → 'summary_focus_query'.

Pins the upgrade/downgrade semantics:
- Rows with the legacy default ``"full"`` get flipped.
- Rows users explicitly chose (``summary_focus``, ``summary_focus_query``,
  ``disabled``) are left untouched.
- Other settings keys with value ``"full"`` are not affected.
- Idempotency: a second upgrade is a no-op once values are migrated.

The on-disk encoding is JSON-text (``"full"`` with the surrounding
quotes), so the test inserts through SQLAlchemy's JSON column type to
match production storage exactly.
"""

import json

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

from local_deep_research.database.alembic_runner import (
    get_alembic_config,
)


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 _seed_setting(engine, key, value):
    """Insert a setting matching production's JSON-text storage.

    SQLAlchemy's JSON column writes ``json.dumps(value)``; we mirror that
    explicitly so raw SQL produces the same on-disk bytes the migration
    expects to match in its WHERE clause.
    """
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO settings "
                "(key, value, type, name, ui_element, visible, editable) "
                "VALUES (:key, :value, 'search', :name, 'select', 1, 1)"
            ),
            {"key": key, "value": json.dumps(value), "name": key},
        )


def _read_setting(engine, key):
    with engine.begin() as conn:
        row = conn.execute(
            text("SELECT value FROM settings WHERE key = :key"),
            {"key": key},
        ).fetchone()
    if row is None:
        return None
    raw = row[0]
    # JSON column round-trips through json.loads on read, but raw text()
    # bypasses the column type — so decode manually.
    return json.loads(raw) if isinstance(raw, str) else raw


@pytest.fixture
def migrated_to_0008_engine(tmp_path):
    """Database fully migrated through 0008 (the revision before 0009)."""
    db_path = tmp_path / "test_0009.db"
    engine = create_engine(f"sqlite:///{db_path}")
    _run_upgrade_to(engine, "0008")
    yield engine
    engine.dispose()


class TestMigration0009Upgrade:
    def test_full_value_is_migrated_to_summary_focus_query(
        self, migrated_to_0008_engine
    ):
        engine = migrated_to_0008_engine

        with engine.begin() as conn:
            conn.execute(
                text("DELETE FROM settings WHERE key = 'search.fetch.mode'")
            )
        _seed_setting(engine, "search.fetch.mode", "full")

        _run_upgrade_to(engine, "0009")

        assert (
            _read_setting(engine, "search.fetch.mode") == "summary_focus_query"
        )

    @pytest.mark.parametrize(
        "explicit_value",
        ["summary_focus", "summary_focus_query", "disabled"],
    )
    def test_explicit_non_full_choices_are_preserved(
        self, migrated_to_0008_engine, explicit_value
    ):
        engine = migrated_to_0008_engine

        with engine.begin() as conn:
            conn.execute(
                text("DELETE FROM settings WHERE key = 'search.fetch.mode'")
            )
        _seed_setting(engine, "search.fetch.mode", explicit_value)

        _run_upgrade_to(engine, "0009")

        assert _read_setting(engine, "search.fetch.mode") == explicit_value

    def test_other_keys_with_full_value_are_not_touched(
        self, migrated_to_0008_engine
    ):
        engine = migrated_to_0008_engine

        # An unrelated key that happens to hold 'full' must not be flipped.
        _seed_setting(engine, "test.unrelated.mode", "full")

        _run_upgrade_to(engine, "0009")

        assert _read_setting(engine, "test.unrelated.mode") == "full"

    def test_upgrade_is_idempotent(self, migrated_to_0008_engine):
        """Running the upgrade a second time is a no-op (already at head)."""
        engine = migrated_to_0008_engine

        with engine.begin() as conn:
            conn.execute(
                text("DELETE FROM settings WHERE key = 'search.fetch.mode'")
            )
        _seed_setting(engine, "search.fetch.mode", "full")

        _run_upgrade_to(engine, "0009")
        # Second invocation: nothing left to do; alembic short-circuits.
        _run_upgrade_to(engine, "0009")

        assert (
            _read_setting(engine, "search.fetch.mode") == "summary_focus_query"
        )

    def test_no_settings_row_does_not_error(self, migrated_to_0008_engine):
        """If the row never existed, the migration is a clean no-op."""
        engine = migrated_to_0008_engine

        with engine.begin() as conn:
            conn.execute(
                text("DELETE FROM settings WHERE key = 'search.fetch.mode'")
            )

        _run_upgrade_to(engine, "0009")

        assert _read_setting(engine, "search.fetch.mode") is None


class TestMigration0009Downgrade:
    def test_downgrade_reverts_summary_focus_query_to_full(
        self, migrated_to_0008_engine
    ):
        engine = migrated_to_0008_engine

        with engine.begin() as conn:
            conn.execute(
                text("DELETE FROM settings WHERE key = 'search.fetch.mode'")
            )
        _seed_setting(engine, "search.fetch.mode", "full")

        _run_upgrade_to(engine, "0009")
        _run_downgrade_to(engine, "0008")

        assert _read_setting(engine, "search.fetch.mode") == "full"

    @pytest.mark.parametrize("preserved_value", ["summary_focus", "disabled"])
    def test_downgrade_preserves_other_explicit_choices(
        self, migrated_to_0008_engine, preserved_value
    ):
        engine = migrated_to_0008_engine

        with engine.begin() as conn:
            conn.execute(
                text("DELETE FROM settings WHERE key = 'search.fetch.mode'")
            )
        _seed_setting(engine, "search.fetch.mode", preserved_value)

        _run_upgrade_to(engine, "0009")
        _run_downgrade_to(engine, "0008")

        assert _read_setting(engine, "search.fetch.mode") == preserved_value


class TestMigration0009HeadAlignment:
    def test_0009_chains_correctly_to_0008(self):
        """0009 (default_fetch_mode_summary) chains directly off 0008.

        Originally this asserted ``get_head_revision() == "0009"``,
        but a later migration added 0010 (chat tables, including the
        partial unique chat-in-progress index) on top, so head moved
        past 0009. The substantive invariant the original test was
        protecting — that 0009 is correctly anchored in the chain —
        survives by checking down_revision instead.
        """
        from alembic.config import Config
        from alembic.script import ScriptDirectory

        from local_deep_research.database.alembic_runner import (
            get_migrations_dir,
        )

        config = Config()
        config.set_main_option("script_location", str(get_migrations_dir()))
        script = ScriptDirectory.from_config(config)
        rev_0009 = script.get_revision("0009")
        assert rev_0009.down_revision == "0008"