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

"""
Tests for migration 0004: Migrate legacy app.* settings.

Tests cover:
- Deprecated settings deletion (app.enable_fact_checking, app.output_dir)
- Re-scoping app.* keys to general.*/search.*/llm.*
- Corrected key mappings (search_engine, openai_endpoint_url, lmstudio_url)
- No overwrite when new key already exists
- Idempotency (no error when old keys don't exist)
- Downgrade is a no-op
"""

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

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


def _run_upgrade_to(engine, revision):
    """Run migrations up to a specific 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):
    """Run downgrade to a specific revision."""
    config = get_alembic_config(engine)
    with engine.begin() as conn:
        config.attributes["connection"] = conn
        command.downgrade(config, revision)


def _insert_setting(conn, key, value, setting_type="app", name=None):
    """Insert a test setting row.

    The value column is JSON type. Pass the raw value to store;
    it will be stored as-is (no extra JSON encoding by raw SQL).
    """
    conn.execute(
        text(
            "INSERT INTO settings (key, value, type, name, ui_element, visible, editable) "
            "VALUES (:key, :value, :type, :name, 'text', 1, 1)"
        ),
        {
            "key": key,
            "value": value,
            "type": setting_type,
            "name": name or key,
        },
    )


def _get_setting(conn, key):
    """Get a setting row by key, or None."""
    return conn.execute(
        text("SELECT key, value, type, name FROM settings WHERE key = :key"),
        {"key": key},
    ).fetchone()


def _count_settings(conn, key_prefix):
    """Count settings with a key prefix."""
    return conn.execute(
        text("SELECT COUNT(*) FROM settings WHERE key LIKE :prefix"),
        {"prefix": f"{key_prefix}%"},
    ).scalar()


@pytest.fixture
def migrated_to_0003_engine(tmp_path):
    """Create a database migrated to revision 0003 (before app settings migration)."""
    db_path = tmp_path / "test_0004.db"
    engine = create_engine(f"sqlite:///{db_path}")
    _run_upgrade_to(engine, "0003")
    yield engine
    engine.dispose()


@pytest.fixture
def fresh_engine(tmp_path):
    """Create a fresh SQLite engine (empty database)."""
    db_path = tmp_path / "fresh_0004_test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    yield engine
    engine.dispose()


class TestMigration0004DeprecatedSettings:
    """Tests for deletion of deprecated settings."""

    def test_deletes_enable_fact_checking(self, migrated_to_0003_engine):
        """app.enable_fact_checking should be deleted."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.enable_fact_checking", "true")

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            assert _get_setting(conn, "app.enable_fact_checking") is None

    def test_deletes_output_dir(self, migrated_to_0003_engine):
        """app.output_dir should be deleted."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.output_dir", '"/tmp/output"')

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            assert _get_setting(conn, "app.output_dir") is None

    def test_no_error_when_deprecated_keys_absent(
        self, migrated_to_0003_engine
    ):
        """Migration should succeed even if deprecated keys don't exist."""
        engine = migrated_to_0003_engine
        # Don't insert any deprecated keys
        _run_upgrade_to(engine, "0004")
        assert get_current_revision(engine) == "0004"


class TestMigration0004GeneralSettings:
    """Tests for re-scoping app.* to general.*."""

    def test_migrates_knowledge_accumulation(self, migrated_to_0003_engine):
        """app.knowledge_accumulation should move to general.knowledge_accumulation."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.knowledge_accumulation", "true")

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            new = _get_setting(conn, "general.knowledge_accumulation")
            assert new is not None
            assert new[2] == "app"  # type set to app
            assert _get_setting(conn, "app.knowledge_accumulation") is None

    def test_migrates_knowledge_accumulation_context_limit(
        self, migrated_to_0003_engine
    ):
        """app.knowledge_accumulation_context_limit should move to general.*."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(
                conn, "app.knowledge_accumulation_context_limit", "5000"
            )

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            new = _get_setting(
                conn, "general.knowledge_accumulation_context_limit"
            )
            assert new is not None
            assert (
                _get_setting(conn, "app.knowledge_accumulation_context_limit")
                is None
            )


class TestMigration0004SearchSettings:
    """Tests for re-scoping app.* to search.*."""

    def test_migrates_questions_per_iteration(self, migrated_to_0003_engine):
        """app.questions_per_iteration should move to search.questions_per_iteration."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.questions_per_iteration", "3")

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            new = _get_setting(conn, "search.questions_per_iteration")
            assert new is not None
            assert new[2] == "search"
            assert _get_setting(conn, "app.questions_per_iteration") is None

    def test_corrects_search_engine_key(self, migrated_to_0003_engine):
        """app.search_engine should map to search.engine.DEFAULT_SEARCH_ENGINE (not search.search_engine)."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.search_engine", '"duckduckgo"')

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            new = _get_setting(conn, "search.engine.DEFAULT_SEARCH_ENGINE")
            assert new is not None
            # Verify old naive key was NOT created
            assert _get_setting(conn, "search.search_engine") is None
            assert _get_setting(conn, "app.search_engine") is None

    def test_migrates_all_search_keys(self, migrated_to_0003_engine):
        """All search keys should be migrated correctly."""
        engine = migrated_to_0003_engine
        search_keys = [
            ("app.iterations", "search.iterations"),
            ("app.max_results", "search.max_results"),
            ("app.region", "search.region"),
            ("app.safe_search", "search.safe_search"),
            ("app.search_language", "search.search_language"),
            ("app.snippets_only", "search.snippets_only"),
        ]

        with engine.begin() as conn:
            for old_key, _ in search_keys:
                _insert_setting(conn, old_key, f'"{old_key}_val"')

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            for old_key, new_key in search_keys:
                new = _get_setting(conn, new_key)
                assert new is not None, f"Expected {new_key} to exist"
                assert new[2] == "search"
                assert _get_setting(conn, old_key) is None


class TestMigration0004LlmSettings:
    """Tests for re-scoping app.* to llm.*."""

    def test_migrates_basic_llm_keys(self, migrated_to_0003_engine):
        """app.model, app.provider, etc. should move to llm.*."""
        engine = migrated_to_0003_engine
        basic_llm_keys = [
            ("app.model", "llm.model"),
            ("app.provider", "llm.provider"),
            ("app.temperature", "llm.temperature"),
            ("app.max_tokens", "llm.max_tokens"),
            ("app.llamacpp_model_path", "llm.llamacpp_model_path"),
        ]

        with engine.begin() as conn:
            for old_key, _ in basic_llm_keys:
                _insert_setting(conn, old_key, f'"{old_key}_val"')

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            for old_key, new_key in basic_llm_keys:
                new = _get_setting(conn, new_key)
                assert new is not None, f"Expected {new_key} to exist"
                assert new[2] == "llm"
                assert _get_setting(conn, old_key) is None

    def test_corrects_openai_endpoint_url_key(self, migrated_to_0003_engine):
        """app.openai_endpoint_url should map to llm.openai_endpoint.url (not llm.openai_endpoint_url)."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(
                conn, "app.openai_endpoint_url", '"http://localhost:1234"'
            )

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            new = _get_setting(conn, "llm.openai_endpoint.url")
            assert new is not None
            assert _get_setting(conn, "llm.openai_endpoint_url") is None
            assert _get_setting(conn, "app.openai_endpoint_url") is None

    def test_corrects_lmstudio_url_key(self, migrated_to_0003_engine):
        """app.lmstudio_url should map to llm.lmstudio.url (not llm.lmstudio_url)."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.lmstudio_url", '"http://localhost:1234"')

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            new = _get_setting(conn, "llm.lmstudio.url")
            assert new is not None
            assert _get_setting(conn, "llm.lmstudio_url") is None
            assert _get_setting(conn, "app.lmstudio_url") is None


class TestMigration0004NoOverwrite:
    """Tests that existing new-key settings are preserved."""

    def test_preserves_existing_new_key(self, migrated_to_0003_engine):
        """If the new key already exists, migration should preserve it and still delete old key."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.model", "old_model")
            _insert_setting(conn, "llm.model", "user_current_model", "llm")

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            new = _get_setting(conn, "llm.model")
            assert new is not None
            assert new[1] == "user_current_model"  # preserved
            assert _get_setting(conn, "app.model") is None  # still deleted

    def test_preserves_existing_search_engine_key(
        self, migrated_to_0003_engine
    ):
        """If search.engine.DEFAULT_SEARCH_ENGINE exists, don't overwrite."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.search_engine", "old_engine")
            _insert_setting(
                conn,
                "search.engine.DEFAULT_SEARCH_ENGINE",
                "current_engine",
                "search",
            )

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            new = _get_setting(conn, "search.engine.DEFAULT_SEARCH_ENGINE")
            assert new[1] == "current_engine"  # preserved
            assert _get_setting(conn, "app.search_engine") is None


class TestMigration0004Idempotency:
    """Tests that migration handles missing keys gracefully."""

    def test_no_error_when_no_app_keys_exist(self, migrated_to_0003_engine):
        """Migration should be a no-op when no app.* keys exist."""
        engine = migrated_to_0003_engine
        _run_upgrade_to(engine, "0004")
        assert get_current_revision(engine) == "0004"

    def test_no_error_with_partial_app_keys(self, migrated_to_0003_engine):
        """Migration should handle only some app.* keys existing."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.model", '"gpt-4"')
            # Don't insert any other app.* keys

        _run_upgrade_to(engine, "0004")

        with engine.connect() as conn:
            assert _get_setting(conn, "llm.model") is not None
            assert _count_settings(conn, "app.") == 0

    def test_full_migration_from_fresh_db(self, fresh_engine):
        """Full migration chain on empty DB should work (no settings to migrate)."""
        run_migrations(fresh_engine)
        assert get_current_revision(fresh_engine) == get_head_revision()


class TestMigration0004Downgrade:
    """Tests for downgrade behavior."""

    def test_downgrade_does_not_crash(self, migrated_to_0003_engine):
        """Downgrade from 0004 to 0003 should not raise."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.model", '"gpt-4"')

        _run_upgrade_to(engine, "0004")
        _run_downgrade_to(engine, "0003")

        assert get_current_revision(engine) == "0003"

    def test_downgrade_does_not_recreate_old_keys(
        self, migrated_to_0003_engine
    ):
        """Downgrade should not recreate the old app.* keys (they're stale)."""
        engine = migrated_to_0003_engine

        with engine.begin() as conn:
            _insert_setting(conn, "app.model", '"gpt-4"')
            _insert_setting(conn, "app.provider", '"openai"')

        _run_upgrade_to(engine, "0004")
        _run_downgrade_to(engine, "0003")

        with engine.connect() as conn:
            # Old keys should NOT be restored
            assert _get_setting(conn, "app.model") is None
            assert _get_setting(conn, "app.provider") is None
            # New keys should still exist (downgrade is no-op)
            assert _get_setting(conn, "llm.model") is not None
            assert _get_setting(conn, "llm.provider") is not None

    def test_downgrade_then_upgrade_roundtrip(self, migrated_to_0003_engine):
        """Downgrade then re-upgrade should work without errors."""
        engine = migrated_to_0003_engine

        _run_upgrade_to(engine, "0004")
        _run_downgrade_to(engine, "0003")
        _run_upgrade_to(engine, "0004")

        assert get_current_revision(engine) == "0004"