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

"""Tests for migration 0018: remove the 'mcp'/'agentic' search strategy.

Pins the upgrade semantics:
- ``search.search_strategy`` rows naming a removed strategy (mcp / agentic)
  are rewritten to ``langgraph-agent``; concrete choices are preserved.
- The orphaned ``mcp.servers`` setting row is deleted.
- ``news_subscriptions.search_strategy`` is NULLed for removed strategies
  (falsy means "use the user's default strategy" in the scheduler).
- ``queued_researches.settings_snapshot`` JSON has both the top-level
  ``submission.strategy`` and the embedded ``search.search_strategy``
  rewritten, for the nested and the legacy flat snapshot structure.
- ``benchmark_runs`` / ``benchmark_configs`` ``search_config.search_strategy``
  is rewritten.
- The settings ``value`` column stores JSON text (``"mcp"`` with quotes); the
  rewritten value must be stored as ``"langgraph-agent"`` (single JSON
  encoding) — a raw-bytes assertion guards against double-encoding.
- Idempotency: re-running the upgrade is a no-op.
"""

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,
)

REMOVED = ["mcp", "agentic"]
REPLACEMENT = "langgraph-agent"


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 _seed_setting(engine, key, value):
    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]
    return json.loads(raw) if isinstance(raw, str) else raw


def _read_setting_raw(engine, key):
    """Return the stored bytes of ``settings.value`` without decoding."""
    with engine.begin() as conn:
        row = conn.execute(
            text("SELECT value FROM settings WHERE key = :key"),
            {"key": key},
        ).fetchone()
    return None if row is None else row[0]


@pytest.fixture
def migrated_to_0017_engine(tmp_path):
    """Database fully migrated through 0017 (the revision before 0018)."""
    db_path = tmp_path / "test_0018.db"
    engine = create_engine(f"sqlite:///{db_path}")
    _run_upgrade_to(engine, "0017")
    yield engine
    engine.dispose()


class TestMigration0018Settings:
    @pytest.mark.parametrize("removed_value", REMOVED)
    def test_strategy_rewritten_to_langgraph_agent(
        self, migrated_to_0017_engine, removed_value
    ):
        engine = migrated_to_0017_engine
        _seed_setting(engine, "search.search_strategy", removed_value)

        _run_upgrade_to(engine, "0018")

        assert _read_setting(engine, "search.search_strategy") == REPLACEMENT

    @pytest.mark.parametrize("removed_value", REMOVED)
    def test_rewritten_value_stored_single_encoded(
        self, migrated_to_0017_engine, removed_value
    ):
        """The stored bytes must be ``"langgraph-agent"`` (single JSON
        encoding), not a double-encoded ``"\\"langgraph-agent\\""``."""
        engine = migrated_to_0017_engine
        _seed_setting(engine, "search.search_strategy", removed_value)

        _run_upgrade_to(engine, "0018")

        assert _read_setting_raw(
            engine, "search.search_strategy"
        ) == json.dumps(REPLACEMENT)

    @pytest.mark.parametrize(
        "concrete", ["source-based", "focused-iteration", "topic-organization"]
    )
    def test_concrete_strategy_preserved(
        self, migrated_to_0017_engine, concrete
    ):
        engine = migrated_to_0017_engine
        _seed_setting(engine, "search.search_strategy", concrete)

        _run_upgrade_to(engine, "0018")

        assert _read_setting(engine, "search.search_strategy") == concrete

    def test_mcp_servers_setting_deleted(self, migrated_to_0017_engine):
        engine = migrated_to_0017_engine
        _seed_setting(engine, "mcp.servers", [{"command": "npx"}])

        _run_upgrade_to(engine, "0018")

        assert _read_setting(engine, "mcp.servers") is None

    def test_upgrade_is_idempotent(self, migrated_to_0017_engine):
        engine = migrated_to_0017_engine
        _seed_setting(engine, "search.search_strategy", "mcp")

        _run_upgrade_to(engine, "0018")
        _run_upgrade_to(engine, "0018")

        assert _read_setting(engine, "search.search_strategy") == REPLACEMENT

    def test_missing_rows_are_a_clean_noop(self, migrated_to_0017_engine):
        engine = migrated_to_0017_engine

        _run_upgrade_to(engine, "0018")

        assert _read_setting(engine, "search.search_strategy") is None


class TestMigration0018NewsSubscriptions:
    def _seed_subscription(self, engine, sub_id, search_strategy):
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO news_subscriptions "
                    "(id, subscription_type, query_or_topic, search_strategy) "
                    "VALUES (:id, 'search', 'test query', :ss)"
                ),
                {"id": sub_id, "ss": search_strategy},
            )

    def _read_subscription_strategy(self, engine, sub_id):
        with engine.begin() as conn:
            return conn.execute(
                text(
                    "SELECT search_strategy FROM news_subscriptions "
                    "WHERE id = :id"
                ),
                {"id": sub_id},
            ).fetchone()[0]

    def test_removed_strategy_nulled_concrete_preserved(
        self, migrated_to_0017_engine
    ):
        engine = migrated_to_0017_engine
        self._seed_subscription(engine, "sub-mcp", "mcp")
        self._seed_subscription(engine, "sub-agentic", "agentic")
        self._seed_subscription(engine, "sub-lg", "langgraph-agent")
        self._seed_subscription(engine, "sub-null", None)

        _run_upgrade_to(engine, "0018")

        assert self._read_subscription_strategy(engine, "sub-mcp") is None
        assert self._read_subscription_strategy(engine, "sub-agentic") is None
        assert (
            self._read_subscription_strategy(engine, "sub-lg")
            == "langgraph-agent"
        )
        assert self._read_subscription_strategy(engine, "sub-null") is None


class TestMigration0018QueuedResearches:
    def _seed_queued(self, engine, research_id, snapshot):
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO queued_researches "
                    "(username, research_id, query, mode, settings_snapshot, "
                    "position) "
                    "VALUES ('tester', :rid, 'q', 'quick', :snap, 1)"
                ),
                {"rid": research_id, "snap": json.dumps(snapshot)},
            )

    def _read_snapshot(self, engine, research_id):
        with engine.begin() as conn:
            raw = conn.execute(
                text(
                    "SELECT settings_snapshot FROM queued_researches "
                    "WHERE research_id = :rid"
                ),
                {"rid": research_id},
            ).fetchone()[0]
        return json.loads(raw) if isinstance(raw, str) else raw

    def test_nested_submission_strategy_rewritten(
        self, migrated_to_0017_engine
    ):
        engine = migrated_to_0017_engine
        self._seed_queued(
            engine,
            "rid-nested",
            {
                "submission": {"strategy": "mcp", "model": "m"},
                "settings_snapshot": {
                    "search.search_strategy": {
                        "value": "agentic",
                        "type": "SEARCH",
                    }
                },
            },
        )

        _run_upgrade_to(engine, "0018")

        snap = self._read_snapshot(engine, "rid-nested")
        assert snap["submission"]["strategy"] == "langgraph-agent"
        assert (
            snap["settings_snapshot"]["search.search_strategy"]["value"]
            == "langgraph-agent"
        )

    def test_legacy_flat_snapshot_rewritten(self, migrated_to_0017_engine):
        engine = migrated_to_0017_engine
        self._seed_queued(
            engine, "rid-flat", {"strategy": "agentic", "model": "m"}
        )

        _run_upgrade_to(engine, "0018")

        snap = self._read_snapshot(engine, "rid-flat")
        assert snap["strategy"] == "langgraph-agent"

    def test_concrete_strategy_snapshot_untouched(
        self, migrated_to_0017_engine
    ):
        engine = migrated_to_0017_engine
        original = {
            "submission": {"strategy": "source-based"},
            "settings_snapshot": {"search.search_strategy": "source-based"},
        }
        self._seed_queued(engine, "rid-src", original)

        _run_upgrade_to(engine, "0018")

        assert self._read_snapshot(engine, "rid-src") == original


class TestMigration0018Benchmarks:
    def _seed_config(self, engine, name, search_strategy):
        search_config = json.dumps(
            {"search_strategy": search_strategy, "iterations": 2}
        )
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO benchmark_configs "
                    "(name, config_hash, search_config, evaluation_config, "
                    "datasets_config, created_at, updated_at, is_default, "
                    "is_public, usage_count) "
                    "VALUES (:name, 'abcd1234', :sc, '{}', '{}', "
                    "CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, 0, 0, 0)"
                ),
                {"name": name, "sc": search_config},
            )

    def _read_config_strategy(self, engine, name):
        with engine.begin() as conn:
            raw = conn.execute(
                text(
                    "SELECT search_config FROM benchmark_configs "
                    "WHERE name = :name"
                ),
                {"name": name},
            ).fetchone()[0]
        config = json.loads(raw) if isinstance(raw, str) else raw
        return config["search_strategy"]

    def test_benchmark_config_strategy_rewritten(self, migrated_to_0017_engine):
        engine = migrated_to_0017_engine
        self._seed_config(engine, "cfg-mcp", "mcp")
        self._seed_config(engine, "cfg-src", "source-based")

        _run_upgrade_to(engine, "0018")

        assert (
            self._read_config_strategy(engine, "cfg-mcp") == "langgraph-agent"
        )
        assert self._read_config_strategy(engine, "cfg-src") == "source-based"

    def _seed_run(self, engine, run_name, search_strategy):
        search_config = json.dumps(
            {"search_strategy": search_strategy, "iterations": 2}
        )
        with engine.begin() as conn:
            conn.execute(
                text(
                    "INSERT INTO benchmark_runs "
                    "(run_name, config_hash, query_hash_list, search_config, "
                    "evaluation_config, datasets_config, status, created_at, "
                    "updated_at, total_examples, completed_examples, "
                    "failed_examples) "
                    "VALUES (:name, 'h0', '[]', :sc, '{}', '{}', 'PENDING', "
                    "'2026-01-01 00:00:00', '2026-01-01 00:00:00', 0, 0, 0)"
                ),
                {"name": run_name, "sc": search_config},
            )

    def _read_run_strategy(self, engine, run_name):
        with engine.begin() as conn:
            raw = conn.execute(
                text(
                    "SELECT search_config FROM benchmark_runs "
                    "WHERE run_name = :name"
                ),
                {"name": run_name},
            ).fetchone()[0]
        config = json.loads(raw) if isinstance(raw, str) else raw
        return config["search_strategy"]

    def test_benchmark_run_strategy_rewritten(self, migrated_to_0017_engine):
        # upgrade() rewrites search_config in BOTH benchmark_runs and
        # benchmark_configs; this pins the benchmark_runs branch (the
        # benchmark_configs path is covered above), guarding against a
        # regression that breaks only one table binding.
        engine = migrated_to_0017_engine
        self._seed_run(engine, "run-mcp", "mcp")
        self._seed_run(engine, "run-src", "source-based")

        _run_upgrade_to(engine, "0018")

        assert self._read_run_strategy(engine, "run-mcp") == "langgraph-agent"
        assert self._read_run_strategy(engine, "run-src") == "source-based"


class TestMigration0018HeadAlignment:
    def test_0018_chains_correctly_to_0017(self):
        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_0018 = script.get_revision("0018")
        assert rev_0018.down_revision == "0017"

    # NOTE: the head-alignment guard (assert head == latest) always lives in
    # the newest migration's test file — currently
    # test_migration_0020_add_zotero_tables.py. 0018 is no longer the head,
    # so asserting it here would break every future migration.