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