"""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"