learningcircuit--local-deep-research/tests/test_settings_manager.py

import os
from typing import Any

import pytest
from sqlalchemy.exc import SQLAlchemyError

from local_deep_research.settings.manager import (
    SettingsManager,
    check_env_setting,
)
from local_deep_research.database.models import Setting, SettingType


def test_check_env_setting_exists(monkeypatch):
    monkeypatch.setenv("LDR_APP_VERSION", "1.0.0")
    assert check_env_setting("app.version") == "1.0.0"


def test_check_env_setting_not_exists():
    # Ensure the environment variable is not set
    if "LDR_APP_VERSION" in os.environ:
        del os.environ["LDR_APP_VERSION"]
    assert check_env_setting("app.version") is None


@pytest.mark.parametrize(
    "ui_element, setting_value, expected",
    [
        ("text", "hello", "hello"),
        ("select", "option_a", "option_a"),
        ("password", "secret", "secret"),
        ("number", "3.14", 3.14),
        ("range", "3.14", 3.14),
        ("checkbox", "true", True),
        ("json", {"key": "value"}, {"key": "value"}),
    ],
)
def test_get_setting_from_db(
    mocker, ui_element: str, setting_value: str, expected: Any
):
    """
    Tests that we can successfully read settings from the DB.

    Args:
        mocker: The fixture to use for mocking.
        ui_element: The value of the `ui_element` parameter, which controls the return type.
        setting_value: The value to use for the setting in the DB.
        expected: The expected typed value of the setting.
    """
    # Arrange: Set up the mock database session and a sample setting
    mock_db_session = mocker.MagicMock()
    mock_setting = Setting(
        key="test.setting", value=setting_value, ui_element=ui_element
    )
    mock_db_session.query.return_value.filter.return_value.all.return_value = [
        mock_setting
    ]

    # Act: Call the get_setting method with the test key
    settings_manager = SettingsManager(db_session=mock_db_session)
    result = settings_manager.get_setting("test.setting")

    # Assert: Verify that the result matches the expected value
    assert result == expected, f"Expected {expected}, but got {result}"


@pytest.mark.parametrize(
    "ui_element, env_value, expected",
    [
        ("text", "hello", "hello"),
        ("select", "option_a", "option_a"),
        ("password", "secret", "secret"),
        ("number", "3.14", 3.14),
        ("range", "3.14", 3.14),
        ("checkbox", "true", True),
    ],
)
def test_get_setting_from_env(
    mocker, monkeypatch, ui_element: str, env_value: str, expected: Any
):
    """
    Tests that we can successfully override DB settings with environment
    variables.

    Args:
        mocker: The fixture to use for mocking.
        ui_element: The value of the `ui_element` parameter, which controls the return type.
        env_value: The value to use for the setting in the environment variable.
        expected: The expected typed value of the setting.
    """
    # Arrange: Set up the mock database session and ensure an environment variable is set
    mock_db_session = mocker.MagicMock()
    monkeypatch.setenv("LDR_TEST_SETTING", env_value)
    # Ensure DB query returns an old version to verify environment variable is prioritized
    mock_setting = Setting(
        key="test.setting", value="db_value", ui_element=ui_element
    )
    mock_db_session.query.return_value.filter.return_value.all.return_value = [
        mock_setting
    ]

    # Act: Call the get_setting method with the test key
    settings_manager = SettingsManager(db_session=mock_db_session)
    result = settings_manager.get_setting("test.setting")

    # Assert: Verify that the result matches the expected value
    assert result == expected, f"Expected {expected}, but got {result}"


def test_get_setting_default(mocker):
    mock_db_session = mocker.MagicMock()
    # Ensure DB query returns nothing
    mock_db_session.query.return_value.filter.return_value.all.return_value = []

    settings_manager = SettingsManager(db_session=mock_db_session)
    assert (
        settings_manager.get_setting(
            "non_existent_setting", default="default_value"
        )
        == "default_value"
    )


def test_get_setting_invalid_type(mocker):
    """
    Tests that when a setting's value cannot be converted to the type
    specified by the `ui_element`, it will always return the default value.

    Args:
        mocker: The fixture to use for mocking.

    """
    # Arrange: Set up a mock DB session and setting with an invalid type
    mock_db_session = mocker.MagicMock()
    mock_setting = Setting(
        key="test.invalid_type", value="not_a_number", ui_element="number"
    )
    mock_db_session.query.return_value.filter.return_value.all.return_value = [
        mock_setting
    ]

    # Act: Call get_setting with a default value and an invalid type
    settings_manager = SettingsManager(db_session=mock_db_session)
    result = settings_manager.get_setting("test.invalid_type", default=10)

    # Assert: Check that the default value is returned and not the invalid string
    assert result == 10


# The mocker fixture is automatically available
def test_set_setting_update_existing(mocker):
    mock_db_session = mocker.MagicMock()
    mock_setting = Setting(key="app.version", value="1.0.0", editable=True)
    mock_db_session.query.return_value.filter.return_value.first.return_value = mock_setting
    mocker.patch(
        "local_deep_research.settings.manager.func.now"
    )  # Patching the func.now call

    settings_manager = SettingsManager(db_session=mock_db_session)
    # Mock the _check_thread_safety method to avoid thread safety errors in tests
    mocker.patch.object(
        settings_manager, "_check_thread_safety", return_value=None
    )

    result = settings_manager.set_setting("app.version", "2.0.0")

    assert result is True
    assert mock_setting.value == "2.0.0"
    mock_db_session.commit.assert_called_once()
    mock_db_session.rollback.assert_not_called()


def test_set_setting_create_new(mocker):
    mock_db_session = mocker.MagicMock()
    mock_db_session.query.return_value.filter.return_value.first.return_value = None
    mocker.patch(
        "local_deep_research.settings.manager.func.now"
    )  # Patching the func.now call

    settings_manager = SettingsManager(db_session=mock_db_session)
    result = settings_manager.set_setting("new.setting", "new_value")

    assert result is True
    mock_db_session.add.assert_called_once()
    new_setting = mock_db_session.add.call_args[0][0]
    assert isinstance(new_setting, Setting)
    assert new_setting.key == "new.setting"
    assert new_setting.value == "new_value"
    assert (
        new_setting.type == SettingType.APP
    )  # Assuming 'app' type for this key
    mock_db_session.commit.assert_called_once()
    mock_db_session.rollback.assert_not_called()


def test_set_setting_db_error(mocker):
    mock_db_session = mocker.MagicMock()
    # Set the side_effect to an instance of SQLAlchemyError
    mock_db_session.query.return_value.filter.return_value.first.side_effect = (
        SQLAlchemyError("Simulated DB Error")
    )

    # Mock the logger to check if error is logged
    mock_logger = mocker.patch("local_deep_research.settings.manager.logger")

    settings_manager = SettingsManager(db_session=mock_db_session)
    result = settings_manager.set_setting("app.version", "2.0.0")

    # Assert that the method returned False
    assert result is False

    mock_db_session.rollback.assert_called_once()
    mock_db_session.commit.assert_not_called()
    mock_logger.exception.assert_called_once()
    # mock_logger.error.assert_called_once_with("Error setting value for app.version: Simulated DB Error")


def test_app_get_setting_from_real_db(
    setup_database_for_all_tests, monkeypatch
):
    """
    Tests get_setting via settings_service, using the real file-based database.
    `setup_database_for_all_tests` yields `SessionLocal`, which is a sessionmaker.
    """
    SessionLocal = setup_database_for_all_tests  # Get the SessionLocal class from the fixture
    session = SessionLocal()  # Create a new session instance for this test

    try:
        # Add a setting to the database using this session
        # Changed key to be specific to this test
        setting = Setting(
            key="test.app.version.get",  # Use a unique key for this test
            value="1.0.0",
            type=SettingType.APP,
            name="Version",
            description="App version",
            ui_element="text",
            visible=True,
            editable=True,
        )
        session.add(setting)
        session.commit()

        monkeypatch.delenv(
            "LDR_APP_VERSION", raising=False
        )  # Ensure no env override

        # Create settings manager with the session and get the setting
        settings_manager = SettingsManager(db_session=session)
        value = settings_manager.get_setting("test.app.version.get")
        assert value == "1.0.0"
    finally:
        session.close()  # Close the session for this test


def test_get_all_settings_from_real_file_db(
    setup_database_for_all_tests, monkeypatch
):
    """
    Tests retrieving all settings from a real *file-based* database.
    """
    # Create a new session for this test
    session = (
        setup_database_for_all_tests()
    )  # Call the sessionmaker to get a new session

    try:
        # Add some settings to the database
        setting1 = Setting(
            key="all.settings.app.version",
            value="1.0.0",
            type=SettingType.APP,
            name="Version",
            description="App version",
            ui_element="text",
            visible=True,
            editable=True,
        )
        setting2 = Setting(
            key="all.settings.llm.temperature",
            value=0.7,
            type=SettingType.LLM,
            name="Temperature",
            description="LLM temperature",
            ui_element="number",
            visible=True,
            editable=True,
        )
        session.add_all([setting1, setting2])
        session.commit()  # Commit to save them to the database

        # Ensure no conflicting environment variables are set for this test
        monkeypatch.delenv("LDR_APP_VERSION", raising=False)
        monkeypatch.delenv("LDR_LLM_TEMPERATURE", raising=False)

        settings_manager = SettingsManager(
            db_session=session
        )  # Pass the session instance
        all_settings = settings_manager.get_all_settings()

        # Assertions should check for the unique keys
        assert "all.settings.app.version" in all_settings
        assert all_settings["all.settings.app.version"]["value"] == "1.0.0"
        assert all_settings["all.settings.app.version"]["type"] == "APP"
        assert all_settings["all.settings.app.version"]["editable"] is True
        assert all_settings["all.settings.app.version"]["name"] == "Version"

        assert "all.settings.llm.temperature" in all_settings
        assert all_settings["all.settings.llm.temperature"]["value"] == 0.7
        assert all_settings["all.settings.llm.temperature"]["type"] == "LLM"
        assert all_settings["all.settings.llm.temperature"]["editable"] is True
        assert (
            all_settings["all.settings.llm.temperature"]["name"]
            == "Temperature"
        )
    finally:
        # IMPORTANT: Close the session created within the test
        session.close()


def test_get_all_settings_db_error(mock_db_session, mock_logger):
    """Tests handling a SQLAlchemyError when retrieving all settings.

    After unification, get_all_settings loads defaults first, then overlays
    DB settings.  When the DB query fails the defaults are still returned
    (graceful degradation) and the error is logged.
    """
    # Configure the mock session to raise a SQLAlchemyError
    mock_db_session.query(Setting).all.side_effect = SQLAlchemyError(
        "Simulated DB Error"
    )

    settings_manager = SettingsManager(db_session=mock_db_session)
    all_settings = settings_manager.get_all_settings()

    # Defaults are still loaded even when the DB query fails
    assert isinstance(all_settings, dict)
    assert len(all_settings) > 0
    mock_logger.exception.assert_called_once()
    # You can check the log message content if needed


def test_create_or_update_setting_does_not_app_prefix_local_search_keys(
    setup_database_for_all_tests,
):
    """Regression for #4208: keys outside the llm./search./report./app.
    buckets (e.g. local_search_*) must be written verbatim. Before the
    fix, create_or_update_setting wrapped the input dict in AppSetting,
    whose Pydantic validator silently rewrote `local_search_embedding_model`
    to `app.local_search_embedding_model` — readers never found the value,
    breaking every save on the Embeddings page.
    """
    session_local_class = setup_database_for_all_tests
    session = session_local_class()

    with session:
        settings_manager = SettingsManager(db_session=session)
        setting_dict = {
            "key": "local_search_embedding_model",
            "value": "all-mpnet-base-v2",
            "name": "Local Search Embedding Model",
            "description": "Test",
        }
        result = settings_manager.create_or_update_setting(setting_dict)
        assert result is not None
        # The key must be persisted as-is, not relocated to app.<key>
        assert result.key == "local_search_embedding_model"

        roundtrip = settings_manager.get_setting("local_search_embedding_model")
        assert roundtrip == "all-mpnet-base-v2"

        # And the buggy `app.local_search_*` row must not exist.
        bad_row = (
            session.query(Setting)
            .filter(Setting.key == "app.local_search_embedding_model")
            .first()
        )
        assert bad_row is None


def test_get_setting_with_substring_keys(setup_database_for_all_tests):
    """
    Tests that we can get the correct value for a setting, even when its key
    is a substring of the key for a different setting.

    Args:
        setup_database_for_all_tests: Fixture that sets up a real database
            for testing. (This is necessary because the bug fix this is
            testing depends on the actual behavior of the SQLAlchemy `filter()`
            function.)

    """
    # Arrange.
    session_local_class = setup_database_for_all_tests
    session = session_local_class()

    with session:
        # Add two settings with overlapping keys but different full keys
        setting1 = Setting(
            key="test.hello",
            value="world",
            ui_element="text",
            type="APP",
            name="Test Setting 1",
            visible=True,
            editable=True,
        )
        setting2 = Setting(
            key="test.hello_world",
            value="universe",
            ui_element="text",
            type="APP",
            name="Test Setting 2",
            visible=True,
            editable=True,
        )
        session.add(setting1)
        session.add(setting2)
        session.commit()

        settings_manager = SettingsManager(db_session=session)

        # Act and assert.
        # Test getting the "test.hello" setting
        result1 = settings_manager.get_setting("test.hello")
        assert result1 == "world"

        # Test getting the "test.hello_world" setting.
        result2 = settings_manager.get_setting("test.hello_world")
        assert result2 == "universe"