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

"""Tests for migration 0013: remove the auto/parallel meta search engines.

Pins the upgrade semantics:
- ``search.tool`` rows naming a removed engine (auto / meta / parallel /
  parallel_scientific) are rewritten to ``searxng``; concrete engine
  choices are preserved.
- Orphaned ``search.engine.auto.*`` / ``search.engine.web.parallel.*``
  setting rows are deleted; sibling keys (e.g. ``use_in_auto_search``
  flags under other engines) survive.
- ``news_subscriptions.search_engine`` is NULLed for removed engines
  (falsy means "use the user's default search tool" in the scheduler).
- ``queued_researches.settings_snapshot`` JSON has its
  ``submission.search_engine`` (and embedded ``search.tool``) rewritten,
  for both the new nested and the legacy flat snapshot structure.
- ``benchmark_runs`` / ``benchmark_configs`` ``search_config.search_tool``
  is rewritten.
- Idempotency: re-running the upgrade is a no-op.

The settings ``value`` column stores JSON text (``"auto"`` with quotes);
seeding mirrors that encoding so the WHERE clauses match production.
"""

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 = ["auto", "meta", "parallel", "parallel_scientific"]


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


@pytest.fixture
def migrated_to_0012_engine(tmp_path):
    """Database fully migrated through 0012 (the revision before 0013)."""
    db_path = tmp_path / "test_0013.db"
    engine = create_engine(f"sqlite:///{db_path}")
    _run_upgrade_to(engine, "0012")
    yield engine
    engine.dispose()


class TestMigration0013Settings:
    @pytest.mark.parametrize("removed_value", REMOVED)
    def test_search_tool_rewritten_to_searxng(
        self, migrated_to_0012_engine, removed_value
    ):
        engine = migrated_to_0012_engine
        _seed_setting(engine, "search.tool", removed_value)

        _run_upgrade_to(engine, "0013")

        assert _read_setting(engine, "search.tool") == "searxng"

    @pytest.mark.parametrize("concrete", ["searxng", "wikipedia", "arxiv"])
    def test_concrete_engine_choice_preserved(
        self, migrated_to_0012_engine, concrete
    ):
        engine = migrated_to_0012_engine
        _seed_setting(engine, "search.tool", concrete)

        _run_upgrade_to(engine, "0013")

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

    def test_orphan_engine_settings_deleted(self, migrated_to_0012_engine):
        engine = migrated_to_0012_engine
        _seed_setting(engine, "search.engine.auto.display_name", "Auto")
        _seed_setting(engine, "search.engine.web.parallel.reliability", 0.5)
        # Sibling keys under other engines must survive — including the
        # use_in_auto_search flags whose names merely contain "auto".
        _seed_setting(
            engine, "search.engine.web.searxng.use_in_auto_search", True
        )

        _run_upgrade_to(engine, "0013")

        assert _read_setting(engine, "search.engine.auto.display_name") is None
        assert (
            _read_setting(engine, "search.engine.web.parallel.reliability")
            is None
        )
        assert (
            _read_setting(
                engine, "search.engine.web.searxng.use_in_auto_search"
            )
            is True
        )

    def test_upgrade_is_idempotent(self, migrated_to_0012_engine):
        engine = migrated_to_0012_engine
        _seed_setting(engine, "search.tool", "auto")

        _run_upgrade_to(engine, "0013")
        _run_upgrade_to(engine, "0013")

        assert _read_setting(engine, "search.tool") == "searxng"

    def test_missing_rows_are_a_clean_noop(self, migrated_to_0012_engine):
        engine = migrated_to_0012_engine

        _run_upgrade_to(engine, "0013")

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


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

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

    def test_removed_engine_nulled_concrete_preserved(
        self, migrated_to_0012_engine
    ):
        engine = migrated_to_0012_engine
        self._seed_subscription(engine, "sub-auto", "auto")
        self._seed_subscription(engine, "sub-parallel", "parallel")
        self._seed_subscription(engine, "sub-wiki", "wikipedia")
        self._seed_subscription(engine, "sub-null", None)

        _run_upgrade_to(engine, "0013")

        assert self._read_subscription_engine(engine, "sub-auto") is None
        assert self._read_subscription_engine(engine, "sub-parallel") is None
        assert self._read_subscription_engine(engine, "sub-wiki") == "wikipedia"
        assert self._read_subscription_engine(engine, "sub-null") is None


class TestMigration0013QueuedResearches:
    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_engine_rewritten(self, migrated_to_0012_engine):
        engine = migrated_to_0012_engine
        self._seed_queued(
            engine,
            "rid-nested",
            {
                "submission": {"search_engine": "auto", "model": "m"},
                "settings_snapshot": {
                    "search.tool": {"value": "auto", "type": "SEARCH"}
                },
            },
        )

        _run_upgrade_to(engine, "0013")

        snap = self._read_snapshot(engine, "rid-nested")
        assert snap["submission"]["search_engine"] == "searxng"
        assert snap["settings_snapshot"]["search.tool"]["value"] == "searxng"

    def test_legacy_flat_snapshot_rewritten(self, migrated_to_0012_engine):
        engine = migrated_to_0012_engine
        self._seed_queued(
            engine, "rid-flat", {"search_engine": "parallel", "model": "m"}
        )

        _run_upgrade_to(engine, "0013")

        snap = self._read_snapshot(engine, "rid-flat")
        assert snap["search_engine"] == "searxng"

    def test_concrete_engine_snapshot_untouched(self, migrated_to_0012_engine):
        engine = migrated_to_0012_engine
        original = {
            "submission": {"search_engine": "wikipedia"},
            "settings_snapshot": {"search.tool": "wikipedia"},
        }
        self._seed_queued(engine, "rid-wiki", original)

        _run_upgrade_to(engine, "0013")

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


class TestMigration0013Benchmarks:
    def _seed_config(self, engine, name, search_tool):
        search_config = json.dumps(
            {"search_tool": search_tool, "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_tool(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_tool"]

    def test_benchmark_config_search_tool_rewritten(
        self, migrated_to_0012_engine
    ):
        engine = migrated_to_0012_engine
        self._seed_config(engine, "cfg-auto", "auto")
        self._seed_config(engine, "cfg-searxng", "searxng")

        _run_upgrade_to(engine, "0013")

        assert self._read_config_tool(engine, "cfg-auto") == "searxng"
        assert self._read_config_tool(engine, "cfg-searxng") == "searxng"


class TestMigration0013HeadAlignment:
    def test_0013_chains_correctly_to_0012(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_0013 = script.get_revision("0013")
        assert rev_0013.down_revision == "0012"