""" Additional deep coverage tests for settings_routes.py targeting ~170 missing statements. Focuses on branches NOT yet covered by existing test files: - api_delete_setting: blocked key (403), not-editable (403), delete returns False (500) - api_update_setting: set_setting returns False (500), create returns None (500), warning-affecting key triggers calculate_warnings, is_blocked_setting (403) - api_toggle_search_favorite: remove path (is_favorite=True), add path, set_setting fails - api_update_search_favorites: set_setting fails (500), no favorites key (400) - api_get_all_settings: category filter path - api_get_db_setting: not found (404), type with .value attribute - save_settings: blocked keys (redirect), outer exception - save_all_settings: multiple settings message, warning-affecting key response - fix_corrupted_settings: duplicate settings found, report.* with no default, exception path - api_get_available_models: Anthropic path, auto-discovery with url_setting, cache save error - reset_to_defaults: exception path - api_get_data_location: encrypted database path - get_bulk_settings: per-setting exception path """ from contextlib import contextmanager from unittest.mock import MagicMock, Mock, patch import pytest from ._settings_route_helpers import _create_test_app # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- MODULE = "local_deep_research.web.routes.settings_routes" DECORATOR_MODULE = "local_deep_research.web.utils.route_decorators" AUTH_DB_MANAGER = "local_deep_research.web.auth.decorators.db_manager" SETTINGS_PREFIX = "/settings" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _mock_auth(): """Return a MagicMock that satisfies login_required db_manager check.""" return MagicMock(is_user_connected=MagicMock(return_value=True)) def _make_db_ctx(mock_session): """Build a mock context-manager for get_user_db_session.""" ctx = MagicMock() ctx.__enter__ = MagicMock(return_value=mock_session) ctx.__exit__ = MagicMock(return_value=None) return ctx def _make_setting( key="test.key", value="val", ui_element="text", name="Test Key", editable=True, setting_type="app", options=None, min_value=None, max_value=None, ): """Build a mock Setting ORM object.""" s = MagicMock() s.key = key s.value = value s.ui_element = ui_element s.name = name s.description = "desc" s.category = "general" s.type = setting_type s.editable = editable s.visible = True s.options = options s.min_value = min_value s.max_value = max_value s.step = None s.updated_at = None return s @contextmanager def _authenticated_client(app, mock_settings=None): """Provide a test client with mocked auth and DB session.""" mock_db = Mock() mock_db.connections = {"testuser": True} mock_db.has_encryption = False mock_setting_obj = Mock() mock_setting_obj.key = "llm.temperature" mock_setting_obj.value = "0.7" mock_setting_obj.editable = True mock_setting_obj.ui_element = "number" _mock_query = Mock() _mock_query.all.return_value = mock_settings or [mock_setting_obj] _mock_query.first.return_value = None _mock_query.filter_by.return_value = _mock_query _mock_query.filter.return_value = _mock_query _mock_query.distinct.return_value = _mock_query _mock_query.group_by.return_value = _mock_query _mock_query.having.return_value = _mock_query _mock_query.order_by.return_value = _mock_query _mock_query.delete.return_value = 0 _mock_db_session = Mock() _mock_db_session.query.return_value = _mock_query @contextmanager def _fake_session(*args, **kwargs): yield _mock_db_session patches = [ patch("local_deep_research.web.auth.decorators.db_manager", mock_db), patch( f"{DECORATOR_MODULE}.get_user_db_session", side_effect=_fake_session ), patch(f"{MODULE}.get_user_db_session", side_effect=_fake_session), patch(f"{MODULE}.settings_limit", lambda f: f), ] try: for p in patches: p.start() with app.test_client() as client: with client.session_transaction() as sess: sess["username"] = "testuser" sess["session_id"] = "test-session-id" yield client, _mock_db_session finally: for p in reversed(patches): p.stop() # --------------------------------------------------------------------------- # Fixture # --------------------------------------------------------------------------- @pytest.fixture() def app(): return _create_test_app() def _authed_get(app, path, **kwargs): """Issue an authenticated GET request.""" with app.test_client() as c: with c.session_transaction() as sess: sess["username"] = "testuser" return c.get(path, **kwargs) def _authed_post(app, path, **kwargs): """Issue an authenticated POST request.""" with app.test_client() as c: with c.session_transaction() as sess: sess["username"] = "testuser" return c.post(path, **kwargs) # --------------------------------------------------------------------------- # api_delete_setting - blocked key (403) # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # api_delete_setting - not editable (403) # --------------------------------------------------------------------------- class TestApiDeleteSettingNotEditable: """api_delete_setting returns 403 when setting is not editable.""" def test_non_editable_setting_returns_403(self): """DELETE on a non-editable setting returns 403.""" app = _create_test_app() locked = _make_setting(key="app.locked", editable=False) with _authenticated_client(app, mock_settings=[locked]) as ( client, mock_session, ): mock_session.query.return_value.filter.return_value.first.return_value = locked resp = client.delete(f"{SETTINGS_PREFIX}/api/app.locked") assert resp.status_code == 403 data = resp.get_json() assert "not editable" in data["error"].lower() # --------------------------------------------------------------------------- # api_delete_setting - delete returns False (500) # --------------------------------------------------------------------------- class TestApiDeleteSettingFails: """api_delete_setting returns 500 when delete_setting returns False.""" def test_delete_returns_false_gives_500(self): """When settings_manager.delete_setting returns False, return 500.""" app = _create_test_app() setting = _make_setting(key="llm.model", editable=True) mock_sm = MagicMock() mock_sm.delete_setting.return_value = False with _authenticated_client(app, mock_settings=[setting]) as ( client, mock_session, ): mock_session.query.return_value.filter.return_value.first.return_value = setting with patch( f"{DECORATOR_MODULE}.SettingsManager", return_value=mock_sm ): resp = client.delete(f"{SETTINGS_PREFIX}/api/llm.model") assert resp.status_code == 500 data = resp.get_json() assert "failed" in data["error"].lower() # --------------------------------------------------------------------------- # api_update_setting - set_setting returns False (500) # --------------------------------------------------------------------------- class TestApiUpdateSettingSetFails: """api_update_setting returns 500 when set_setting returns False.""" def test_set_setting_false_returns_500(self): """When set_setting returns False for existing setting, return 500.""" app = _create_test_app() setting = _make_setting( key="llm.model", value="gpt-4", ui_element="text", editable=True ) setting.type = "llm" with _authenticated_client(app, mock_settings=[setting]) as ( client, mock_session, ): mock_session.query.return_value.filter.return_value.first.return_value = setting with patch( f"{MODULE}.coerce_setting_for_write", return_value="gpt-3.5" ): with patch( f"{MODULE}.validate_setting", return_value=(True, None) ): with patch(f"{MODULE}.set_setting", return_value=False): resp = client.put( f"{SETTINGS_PREFIX}/api/llm.model", json={"value": "gpt-3.5"}, content_type="application/json", ) assert resp.status_code == 500 data = resp.get_json() assert "failed" in data["error"].lower() # --------------------------------------------------------------------------- # api_update_setting - create returns None (500) # --------------------------------------------------------------------------- class TestApiUpdateSettingCreateFails: """api_update_setting returns 500 when create_or_update_setting returns None.""" def test_create_returns_none_gives_500(self): """When create_or_update_setting returns None for new setting, return 500.""" app = _create_test_app() with _authenticated_client(app) as (client, mock_session): mock_session.query.return_value.filter.return_value.first.return_value = None with patch(f"{MODULE}.create_or_update_setting", return_value=None): resp = client.put( f"{SETTINGS_PREFIX}/api/llm.new_setting", json={"value": "val"}, content_type="application/json", ) assert resp.status_code == 500 data = resp.get_json() assert "failed" in data["error"].lower() # --------------------------------------------------------------------------- # api_update_setting - warning-affecting key triggers calculate_warnings # --------------------------------------------------------------------------- class TestApiUpdateSettingWarningKey: """api_update_setting includes warnings for warning-affecting keys.""" def test_warning_key_includes_warnings_in_response(self): """Updating llm.provider triggers calculate_warnings in the response.""" app = _create_test_app() setting = _make_setting( key="llm.provider", value="openai", ui_element="text", editable=True ) setting.type = "llm" mock_warnings = [{"type": "info", "message": "Provider changed"}] with _authenticated_client(app, mock_settings=[setting]) as ( client, mock_session, ): mock_session.query.return_value.filter.return_value.first.return_value = setting with patch( f"{MODULE}.coerce_setting_for_write", return_value="anthropic" ): with patch( f"{MODULE}.validate_setting", return_value=(True, None) ): with patch(f"{MODULE}.set_setting", return_value=True): with patch( f"{MODULE}.calculate_warnings", return_value=mock_warnings, ): resp = client.put( f"{SETTINGS_PREFIX}/api/llm.provider", json={"value": "anthropic"}, content_type="application/json", ) assert resp.status_code == 200 data = resp.get_json() assert "warnings" in data assert data["warnings"] == mock_warnings # --------------------------------------------------------------------------- # api_update_setting - blocked key (403) # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # api_toggle_search_favorite - remove path (was a favorite) # --------------------------------------------------------------------------- class TestApiToggleFavoriteRemove: """api_toggle_search_favorite removes engine when already a favorite.""" def test_toggle_removes_existing_favorite(self): """Engine already in favorites gets removed.""" app = _create_test_app() mock_sm = MagicMock() mock_sm.get_setting.return_value = ["google", "bing"] mock_sm.set_setting.return_value = True with _authenticated_client(app) as (client, mock_session): with patch( f"{DECORATOR_MODULE}.SettingsManager", return_value=mock_sm ): resp = client.post( f"{SETTINGS_PREFIX}/api/search-favorites/toggle", json={"engine_id": "google"}, content_type="application/json", ) assert resp.status_code == 200 data = resp.get_json() assert data["is_favorite"] is False assert "google" not in data["favorites"] # --------------------------------------------------------------------------- # api_toggle_search_favorite - add path (not yet a favorite) # --------------------------------------------------------------------------- class TestApiToggleFavoriteAdd: """api_toggle_search_favorite adds engine when not yet a favorite.""" def test_toggle_adds_new_favorite(self): """Engine not in favorites gets added.""" app = _create_test_app() mock_sm = MagicMock() mock_sm.get_setting.return_value = ["bing"] mock_sm.set_setting.return_value = True with _authenticated_client(app) as (client, mock_session): with patch( f"{DECORATOR_MODULE}.SettingsManager", return_value=mock_sm ): resp = client.post( f"{SETTINGS_PREFIX}/api/search-favorites/toggle", json={"engine_id": "google"}, content_type="application/json", ) assert resp.status_code == 200 data = resp.get_json() assert data["is_favorite"] is True assert "google" in data["favorites"] # --------------------------------------------------------------------------- # api_toggle_search_favorite - set_setting fails (500) # --------------------------------------------------------------------------- class TestApiToggleFavoriteFails: """api_toggle_search_favorite returns 500 when set_setting fails.""" def test_set_setting_failure_returns_500(self): """When set_setting fails, return 500.""" app = _create_test_app() mock_sm = MagicMock() mock_sm.get_setting.return_value = [] mock_sm.set_setting.return_value = False with _authenticated_client(app) as (client, mock_session): with patch( f"{DECORATOR_MODULE}.SettingsManager", return_value=mock_sm ): resp = client.post( f"{SETTINGS_PREFIX}/api/search-favorites/toggle", json={"engine_id": "google"}, content_type="application/json", ) assert resp.status_code == 500 data = resp.get_json() assert "error" in data # --------------------------------------------------------------------------- # api_update_search_favorites - no favorites key in body (400) # --------------------------------------------------------------------------- class TestApiUpdateSearchFavoritesNoFavorites: """api_update_search_favorites returns 400 when favorites key is missing.""" def test_missing_favorites_key_returns_400(self): """Body without 'favorites' key returns 400.""" app = _create_test_app() with _authenticated_client(app) as (client, _): resp = client.put( f"{SETTINGS_PREFIX}/api/search-favorites", json={"other": "data"}, content_type="application/json", ) assert resp.status_code == 400 data = resp.get_json() assert "error" in data # --------------------------------------------------------------------------- # api_update_search_favorites - set_setting fails (500) # --------------------------------------------------------------------------- class TestApiUpdateSearchFavoritesFails: """api_update_search_favorites returns 500 when set_setting fails.""" def test_set_setting_failure_returns_500(self): """When set_setting returns False, return 500.""" app = _create_test_app() mock_sm = MagicMock() mock_sm.set_setting.return_value = False with _authenticated_client(app) as (client, mock_session): with patch( f"{DECORATOR_MODULE}.SettingsManager", return_value=mock_sm ): resp = client.put( f"{SETTINGS_PREFIX}/api/search-favorites", json={"favorites": ["google"]}, content_type="application/json", ) assert resp.status_code == 500 data = resp.get_json() assert "error" in data # --------------------------------------------------------------------------- # api_get_all_settings - category filter # --------------------------------------------------------------------------- class TestApiGetAllSettingsWithCategory: """api_get_all_settings with ?category= query parameter.""" def test_category_filter_returns_matching_settings(self): """Settings with matching category are returned.""" app = _create_test_app() llm_setting = _make_setting( key="llm.model", value="gpt-4", setting_type="llm" ) llm_setting.category = "llm_general" search_setting = _make_setting( key="search.tool", value="searxng", setting_type="search" ) search_setting.category = "search_general" mock_sm = MagicMock() mock_sm.get_all_settings.return_value = { "llm.model": "gpt-4", "search.tool": "searxng", } with _authenticated_client( app, mock_settings=[llm_setting, search_setting] ) as ( client, mock_session, ): mock_session.query.return_value.all.return_value = [ llm_setting, search_setting, ] with patch( f"{DECORATOR_MODULE}.SettingsManager", return_value=mock_sm ): resp = client.get(f"{SETTINGS_PREFIX}/api?category=llm_general") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" assert "llm.model" in data["settings"] assert "search.tool" not in data["settings"] # --------------------------------------------------------------------------- # api_get_db_setting - not found (404) # --------------------------------------------------------------------------- class TestApiGetDbSettingNotFound: """api_get_db_setting returns 404 when setting not found.""" def test_not_found_returns_404(self): """When setting key is absent from DB, return 404.""" app = _create_test_app() with _authenticated_client(app) as (client, mock_session): mock_session.query.return_value.filter.return_value.first.return_value = None resp = client.get(f"{SETTINGS_PREFIX}/api/nonexistent.key") assert resp.status_code == 404 data = resp.get_json() assert "not found" in data["error"].lower() # --------------------------------------------------------------------------- # api_get_db_setting - type with .value attribute # --------------------------------------------------------------------------- class TestApiGetDbSettingTypeValue: """api_get_db_setting serializes setting type using .value attribute.""" def test_enum_type_is_serialized_via_value(self): """When setting.type has a .value attribute, it is used in the response.""" app = _create_test_app() setting = _make_setting( key="llm.temperature", value="0.7", ui_element="number", editable=True, ) # Give type an enum-like .value attribute enum_type = MagicMock() enum_type.value = "llm" setting.type = enum_type with _authenticated_client(app, mock_settings=[setting]) as ( client, mock_session, ): mock_session.query.return_value.filter.return_value.first.return_value = setting resp = client.get(f"{SETTINGS_PREFIX}/api/llm.temperature") assert resp.status_code == 200 data = resp.get_json() assert data["type"] == "llm" # --------------------------------------------------------------------------- # save_settings - blocked keys # --------------------------------------------------------------------------- class TestSaveSettingsBlockedKeys: """save_settings (POST form fallback) blocks security-sensitive keys.""" def test_blocked_key_redirects_with_error(self): """Form POST with blocked key triggers flash and redirect.""" app = _create_test_app() with _authenticated_client(app) as (client, _): resp = client.post( f"{SETTINGS_PREFIX}/save_settings", data={"engine.module_path": "/evil/path"}, ) # The endpoint redirects after blocking assert resp.status_code == 302 # --------------------------------------------------------------------------- # save_settings - outer exception # --------------------------------------------------------------------------- class TestSaveSettingsOuterException: """save_settings outer exception triggers flash + redirect.""" def test_outer_exception_returns_500(self): """Exception during SettingsManager init in decorator returns 500.""" app = _create_test_app() with _authenticated_client(app) as (client, mock_session): mock_session.query.side_effect = RuntimeError("unexpected") resp = client.post( f"{SETTINGS_PREFIX}/save_settings", data={"llm.model": "gpt-4"}, ) assert resp.status_code == 500 # --------------------------------------------------------------------------- # save_all_settings - multiple settings message # --------------------------------------------------------------------------- class TestSaveAllSettingsMultipleMessage: """save_all_settings uses generic message when multiple settings updated.""" def test_multiple_settings_shows_generic_message(self): """When 2+ settings are updated, message shows count.""" app = _create_test_app() s1 = _make_setting( key="llm.model", value="gpt-4", ui_element="text", editable=True ) s1.type = "llm" s2 = _make_setting( key="search.tool", value="searxng", ui_element="text", editable=True ) s2.type = "search" with _authenticated_client(app, mock_settings=[s1, s2]) as ( client, mock_session, ): mock_session.query.return_value.all.side_effect = [ [s1, s2], # initial fetch [s1, s2], # second query for all_settings ] with patch( f"{MODULE}.coerce_setting_for_write", return_value="new_val" ): with patch( f"{MODULE}.validate_setting", return_value=(True, None) ): with patch(f"{MODULE}.set_setting", return_value=True): resp = client.post( f"{SETTINGS_PREFIX}/save_all_settings", json={ "llm.model": "gpt-3.5", "search.tool": "google", }, content_type="application/json", ) assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" # Multiple settings: message should contain count info assert "2" in data["message"] or "updated" in data["message"].lower() class TestSaveAllSettingsSecretNoop: """POST /save_all_settings must treat ""/sentinel for a SECRET setting as a no-op (never overwrite the stored value). "Secret" is the same predicate the GET redactor uses: ui_element=='password' OR a sensitive key suffix — so a redacted GET round-trip can't destroy the secret.""" def _post_and_assert_noop(self, setting, submitted_value): from local_deep_research.security.data_sanitizer import DataSanitizer # noqa: F401 app = _create_test_app() with _authenticated_client(app, mock_settings=[setting]) as ( client, mock_session, ): mock_session.query.return_value.all.side_effect = [ [setting], # all_db_settings (carries ui_element) [setting], # echo query ] with patch(f"{MODULE}.set_setting", return_value=True) as set_mock: resp = client.post( f"{SETTINGS_PREFIX}/save_all_settings", json={setting.key: submitted_value}, content_type="application/json", ) assert resp.status_code == 200 assert all( call.args[0] != setting.key for call in set_mock.call_args_list ), f"{setting.key} was written despite no-op guard" def test_empty_password_is_noop(self): s = _make_setting( key="llm.openai.api_key", ui_element="password", editable=True ) s.type = "llm" self._post_and_assert_noop(s, "") def test_sentinel_password_is_noop(self): from local_deep_research.security.data_sanitizer import DataSanitizer s = _make_setting( key="llm.openai.api_key", ui_element="password", editable=True ) s.type = "llm" self._post_and_assert_noop(s, DataSanitizer.REDACTION_TEXT) def test_sensitive_suffix_non_password_sentinel_is_noop(self): """Closes the read/write asymmetry: a secret stored with a non-password ui_element but a sensitive '.api_key' suffix is redacted on GET, so a save round-trip submits the sentinel — the guard must skip it too (it now shares the redactor's predicate).""" from local_deep_research.security.data_sanitizer import DataSanitizer s = _make_setting( key="llm.custom.api_key", ui_element="text", editable=True ) s.type = "llm" self._post_and_assert_noop(s, DataSanitizer.REDACTION_TEXT) class TestSaveAllSettingsRedactsResponse: """The POST /save_all_settings response echoes the full settings dict; password values must be redacted in it so the endpoint never ships plaintext API keys back to the browser (matching GET /settings/api).""" def test_password_value_redacted_in_response(self): from local_deep_research.security.data_sanitizer import DataSanitizer app = _create_test_app() secret = _make_setting( key="llm.openai.api_key", value="sk-super-secret", ui_element="password", editable=True, ) secret.type = "llm" plain = _make_setting( key="llm.model", value="gpt-4", ui_element="text", editable=True ) plain.type = "llm" with _authenticated_client(app, mock_settings=[secret, plain]) as ( client, mock_session, ): mock_session.query.return_value.all.side_effect = [ [secret, plain], # initial fetch [secret, plain], # second query for all_settings echo ] with patch( f"{MODULE}.coerce_setting_for_write", return_value="gpt-3.5" ): with patch( f"{MODULE}.validate_setting", return_value=(True, None) ): with patch(f"{MODULE}.set_setting", return_value=True): resp = client.post( f"{SETTINGS_PREFIX}/save_all_settings", json={"llm.model": "gpt-3.5"}, content_type="application/json", ) assert resp.status_code == 200 data = resp.get_json() echoed = data["settings"]["llm.openai.api_key"] assert echoed["value"] == DataSanitizer.REDACTION_TEXT # The plaintext secret must never appear anywhere in the response. assert "sk-super-secret" not in resp.get_data(as_text=True) # Non-secret settings pass through unredacted. assert data["settings"]["llm.model"]["value"] == "gpt-4" # --------------------------------------------------------------------------- # save_all_settings - warning-affecting key # --------------------------------------------------------------------------- class TestSaveAllSettingsWarningKey: """save_all_settings includes warnings when warning-affecting key changed.""" def test_warning_affecting_key_includes_warnings(self): """Updating llm.provider triggers calculate_warnings in response.""" app = _create_test_app() setting = _make_setting( key="llm.provider", value="openai", ui_element="text", editable=True ) setting.type = "llm" mock_warnings = [{"type": "info", "message": "Check LLM config"}] with _authenticated_client(app, mock_settings=[setting]) as ( client, mock_session, ): mock_session.query.return_value.all.side_effect = [ [setting], [setting], ] with patch( f"{MODULE}.coerce_setting_for_write", return_value="anthropic" ): with patch( f"{MODULE}.validate_setting", return_value=(True, None) ): with patch(f"{MODULE}.set_setting", return_value=True): with patch( f"{MODULE}.calculate_warnings", return_value=mock_warnings, ): resp = client.post( f"{SETTINGS_PREFIX}/save_all_settings", json={"llm.provider": "anthropic"}, content_type="application/json", ) assert resp.status_code == 200 data = resp.get_json() assert "warnings" in data assert data["warnings"] == mock_warnings # --------------------------------------------------------------------------- # fix_corrupted_settings - duplicate settings removed # --------------------------------------------------------------------------- class TestFixCorruptedSettingsDuplicates: """fix_corrupted_settings removes duplicate settings.""" def test_duplicate_settings_are_removed(self): """Duplicate keys trigger deletion of all but the most recent.""" app = _create_test_app() dupe1 = _make_setting(key="llm.model", value="gpt-4") dupe2 = _make_setting(key="llm.model", value="gpt-3.5") with _authenticated_client(app, mock_settings=[]) as ( client, mock_session, ): # First query: duplicate keys query dup_query = MagicMock() dup_query.group_by.return_value.having.return_value.all.return_value = [ ("llm.model",) ] # Second query: settings for the duplicate key dupe_query = MagicMock() dupe_query.filter.return_value.order_by.return_value.all.return_value = [ dupe1, dupe2, ] # Third query: all settings for corruption check all_query = MagicMock() all_query.all.return_value = [] call_count = [0] def _query_side_effect(model_class): call_count[0] += 1 if call_count[0] == 1: return dup_query if call_count[0] == 2: return dupe_query return all_query mock_session.query.side_effect = _query_side_effect resp = client.post(f"{SETTINGS_PREFIX}/fix_corrupted_settings") assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "success" mock_session.delete.assert_called_once_with(dupe2) # --------------------------------------------------------------------------- # fix_corrupted_settings - report.* with no known default → set to {} # --------------------------------------------------------------------------- class TestFixCorruptedSettingsReportNoDefault: """fix_corrupted_settings sets unknown report.* to empty dict.""" def test_corrupted_unknown_report_key_set_to_empty_dict(self): """Corrupted report.unknown_key gets set to {} since no default exists.""" app = _create_test_app() setting = _make_setting(key="report.unknown_format", value=None) with _authenticated_client(app, mock_settings=[setting]) as ( client, mock_session, ): mock_session.query.return_value.group_by.return_value.having.return_value.all.return_value = [] mock_session.query.return_value.all.return_value = [setting] resp = client.post(f"{SETTINGS_PREFIX}/fix_corrupted_settings") assert resp.status_code == 200 data = resp.get_json() assert "report.unknown_format" in data["fixed_settings"] # --------------------------------------------------------------------------- # fix_corrupted_settings - exception path (500) # --------------------------------------------------------------------------- class TestFixCorruptedSettingsException: """fix_corrupted_settings returns 500 on unexpected exception.""" def test_exception_returns_500(self): """When an unexpected error occurs, return 500 with error status.""" app = _create_test_app() with _authenticated_client(app) as (client, mock_session): # Make the group_by chain raise an exception mock_session.query.return_value.group_by.side_effect = RuntimeError( "db error" ) resp = client.post(f"{SETTINGS_PREFIX}/fix_corrupted_settings") assert resp.status_code == 500 data = resp.get_json() assert data["status"] == "error" # --------------------------------------------------------------------------- # api_get_available_models - Anthropic key path # --------------------------------------------------------------------------- class TestApiGetAvailableModelsAnthropic: """api_get_available_models fetches Anthropic models when key is set.""" def test_anthropic_key_triggers_model_fetch(self): """Anthropic (cloud) models flow through the auto-discovered-provider loop, which calls AnthropicProvider.list_models_for_api (anthropic SDK). The previous hardcoded route branch was removed once that method listed models correctly.""" from local_deep_research.llm.providers.auto_discovery import ( ProviderInfo, ) from local_deep_research.llm.providers.implementations.anthropic import ( AnthropicProvider, ) app = _create_test_app() mock_model = MagicMock() mock_model.id = "claude-3-opus-20240229" mock_model.display_name = "Claude 3 Opus" mock_models_resp = MagicMock() mock_models_resp.data = [mock_model] mock_anthropic_client = MagicMock() mock_anthropic_client.models.list.return_value = mock_models_resp mock_cache_query = MagicMock() mock_cache_query.filter.return_value.all.return_value = [] def _setting_side_effect(key, default=""): if key == "llm.anthropic.api_key": return "sk-ant-test" return default with _authenticated_client(app) as (client, mock_session): mock_session.query.return_value = mock_cache_query mock_session.query.return_value.delete.return_value = 0 with ( patch( "local_deep_research.llm.providers.get_discovered_provider_options", return_value=[], ), patch( f"{MODULE}._get_setting_from_session", side_effect=_setting_side_effect, ), patch(f"{MODULE}._model_list_local_only", return_value=False), patch(f"{MODULE}.safe_get") as mock_safe_get, patch( "local_deep_research.llm.providers.discover_providers", return_value={"ANTHROPIC": ProviderInfo(AnthropicProvider)}, ), patch( "anthropic.Anthropic", return_value=mock_anthropic_client ), ): mock_ollama_resp = MagicMock() mock_ollama_resp.status_code = 200 mock_ollama_resp.text = '{"models": []}' mock_ollama_resp.json.return_value = {"models": []} mock_safe_get.return_value = mock_ollama_resp resp = client.get( f"{SETTINGS_PREFIX}/api/available-models?force_refresh=true" ) assert resp.status_code == 200 data = resp.get_json() assert "providers" in data # The model fetched via the anthropic SDK made it into the response — # proving the discovered-provider loop now lists Anthropic correctly. assert "claude-3-opus-20240229" in str(data) mock_anthropic_client.models.list.assert_called_once() # --------------------------------------------------------------------------- # api_get_available_models - auto-discovery with url_setting # --------------------------------------------------------------------------- class TestApiGetAvailableModelsAutoDiscoveryUrl: """api_get_available_models uses url_setting from provider class if present.""" def test_auto_discovered_provider_with_url_setting(self): """Provider with url_setting attribute fetches the base_url from session.""" app = _create_test_app() mock_provider_class = MagicMock() mock_provider_class.api_key_setting = "llm.custom.api_key" mock_provider_class.url_setting = "llm.custom.url" mock_provider_class.list_models_for_api.return_value = [ {"value": "custom-model", "label": "Custom Model (Custom)"} ] mock_provider_info = MagicMock() mock_provider_info.provider_name = "Custom Provider" mock_provider_info.provider_class = mock_provider_class mock_cache_query = MagicMock() mock_cache_query.filter.return_value.all.return_value = [] def _setting_side_effect(key, default=""): if key == "llm.custom.api_key": return "custom-key" if key == "llm.custom.url": return "http://custom.example.com" return default with _authenticated_client(app) as (client, mock_session): mock_session.query.return_value = mock_cache_query mock_session.query.return_value.delete.return_value = 0 with patch( "local_deep_research.llm.providers.get_discovered_provider_options", return_value=[], ): with patch( f"{MODULE}._get_setting_from_session", side_effect=_setting_side_effect, ): with patch(f"{MODULE}.safe_get") as mock_safe_get: mock_ollama_resp = MagicMock() mock_ollama_resp.status_code = 200 mock_ollama_resp.text = '{"models": []}' mock_ollama_resp.json.return_value = {"models": []} mock_safe_get.return_value = mock_ollama_resp with patch( "local_deep_research.llm.providers.discover_providers", return_value={"custom": mock_provider_info}, ): resp = client.get( f"{SETTINGS_PREFIX}/api/available-models?force_refresh=true" ) assert resp.status_code == 200 resp.get_json() # The url_setting branch should have been exercised mock_provider_class.list_models_for_api.assert_called_once_with( "custom-key", "http://custom.example.com" ) # --------------------------------------------------------------------------- # api_get_available_models - cache save error (continues gracefully) # --------------------------------------------------------------------------- class TestApiGetAvailableModelsCacheSaveError: """api_get_available_models continues when saving to cache fails.""" def test_cache_save_error_still_returns_200(self): """Even if saving models to cache raises, the response is still 200.""" app = _create_test_app() mock_cache_query = MagicMock() mock_cache_query.filter.return_value.all.return_value = [] with _authenticated_client(app) as (client, mock_session): mock_session.query.return_value = mock_cache_query # Make commit raise so the cache save fails mock_session.commit.side_effect = RuntimeError("db locked") with patch( "local_deep_research.llm.providers.get_discovered_provider_options", return_value=[], ): with patch( f"{MODULE}._get_setting_from_session", return_value="" ): with patch(f"{MODULE}.safe_get") as mock_safe_get: mock_ollama_resp = MagicMock() mock_ollama_resp.status_code = 200 mock_ollama_resp.text = '{"models": []}' mock_ollama_resp.json.return_value = {"models": []} mock_safe_get.return_value = mock_ollama_resp with patch( "local_deep_research.llm.providers.discover_providers", return_value={}, ): resp = client.get( f"{SETTINGS_PREFIX}/api/available-models?force_refresh=true" ) # Should still return 200 — cache save error is logged but not fatal assert resp.status_code == 200 data = resp.get_json() assert "providers" in data # --------------------------------------------------------------------------- # reset_to_defaults - exception path (500) # --------------------------------------------------------------------------- class TestResetToDefaultsException: """reset_to_defaults returns 500 when an exception occurs.""" def test_exception_returns_500(self): """When SettingsManager.load_from_defaults_file raises, return 500.""" app = _create_test_app() mock_sm = MagicMock() mock_sm.load_from_defaults_file.side_effect = RuntimeError( "file not found" ) with _authenticated_client(app) as (client, _): with patch( f"{DECORATOR_MODULE}.SettingsManager", return_value=mock_sm ): resp = client.post(f"{SETTINGS_PREFIX}/reset_to_defaults") assert resp.status_code == 500 data = resp.get_json() assert data["status"] == "error" assert "failed" in data["message"].lower() # --------------------------------------------------------------------------- # api_get_data_location - encryption enabled path # --------------------------------------------------------------------------- class TestApiGetDataLocationEncrypted: """api_get_data_location reports encryption status when db is encrypted.""" def test_encrypted_database_path(self): """When has_encryption is True, response shows encrypted=True.""" app = _create_test_app() mock_dbm = MagicMock() mock_dbm.has_encryption = True with _authenticated_client(app) as (client, _): with patch(f"{MODULE}.db_manager", mock_dbm): with patch( f"{MODULE}.get_data_directory", return_value="/data" ): with patch( f"{MODULE}.get_encrypted_database_path", return_value="/data/db.enc", ): with patch( "local_deep_research.web.utils.route_decorators.SettingsManager" ) as mock_sm_cls: mock_sm_instance = MagicMock() mock_sm_instance.get_setting.return_value = None mock_sm_cls.return_value = mock_sm_instance with patch( "local_deep_research.database.sqlcipher_utils.get_sqlcipher_settings", return_value={"cipher": "AES-256"}, ): with patch( f"{MODULE}.platform" ) as mock_platform: mock_platform.system.return_value = "Linux" resp = client.get( f"{SETTINGS_PREFIX}/api/data-location" ) assert resp.status_code == 200 data = resp.get_json() assert data["security_notice"]["encrypted"] is True # --------------------------------------------------------------------------- # get_bulk_settings - per-setting exception # --------------------------------------------------------------------------- class TestGetBulkSettingsPerSettingError: """get_bulk_settings handles per-setting exception gracefully.""" def test_per_setting_error_included_in_response(self): """When _get_setting_from_session raises for one key, error is noted.""" app = _create_test_app() def _setting_side_effect(key, *args, **kwargs): if key == "llm.model": raise RuntimeError("db error") return "some_value" with _authenticated_client(app) as (client, _): with patch( f"{MODULE}._get_setting_from_session", side_effect=_setting_side_effect, ): resp = client.get( f"{SETTINGS_PREFIX}/api/bulk?keys[]=llm.model&keys[]=search.tool" ) assert resp.status_code == 200 data = resp.get_json() assert data["success"] is True # The failing key should still be present but with error info assert "llm.model" in data["settings"] assert data["settings"]["llm.model"]["exists"] is False assert "error" in data["settings"]["llm.model"] # --------------------------------------------------------------------------- # _get_setting_from_session - guard against key=None # --------------------------------------------------------------------------- class TestGetSettingFromSessionNoneKey: """_get_setting_from_session must short-circuit when key is None. Regression for issue #3800: providers like LM Studio and Llama.cpp declare ``api_key_setting = None``. Without the guard, the helper would delegate to ``SettingsManager.get_setting(None, ...)``, which treats None as "return all settings" — leaking every other provider's API key into the auto-discovery loop's ``api_key`` argument. """ def test_none_key_returns_default_without_db_call(self): """With key=None, return default and never touch SettingsManager.""" from local_deep_research.web.routes.settings_routes import ( _get_setting_from_session, ) app = _create_test_app() with _authenticated_client(app) as (client, _): with client.session_transaction() as sess: sess["username"] = "testuser" with app.test_request_context("/"): from flask import session as flask_session flask_session["username"] = "testuser" with patch(f"{MODULE}.get_settings_manager") as mock_get_sm: result = _get_setting_from_session(None, "fallback") assert result == "fallback" mock_get_sm.assert_not_called() # --------------------------------------------------------------------------- # api_get_available_models - api_key_setting=None must not poison api_key # --------------------------------------------------------------------------- class TestApiGetAvailableModelsApiKeySettingNone: """Auto-discovered providers with api_key_setting=None get api_key="". Regression for issue #3800: LMStudioProvider and LlamaCppProvider declare ``api_key_setting = None``. The route must not pass a dict of all settings to ``list_models_for_api`` — that would build ``Authorization: Bearer `` and leak every cloud provider's API key to the local LM Studio/llama-server endpoint. """ def test_none_api_key_setting_passes_empty_string_not_dict(self): """When api_key_setting=None, list_models_for_api gets api_key=''. This test mocks ``get_settings_manager`` (one layer below the helper) rather than ``_get_setting_from_session`` itself. That way the production helper actually runs and the ``key is None`` guard is exercised. The mocked manager simulates the original bug — ``get_setting(None, ...)`` returns a settings dict — so if the guard were removed, the dict would propagate and the final ``isinstance(api_key, str)`` assertion would fail. """ app = _create_test_app() mock_provider_class = MagicMock() mock_provider_class.api_key_setting = None mock_provider_class.url_setting = "llm.lmstudio.url" mock_provider_class.list_models_for_api.return_value = [ {"value": "local-model", "label": "Local Model"} ] mock_provider_info = MagicMock() mock_provider_info.provider_name = "LM Studio" mock_provider_info.provider_class = mock_provider_class mock_cache_query = MagicMock() mock_cache_query.filter.return_value.all.return_value = [] # Build a mock SettingsManager that simulates the buggy # ``SettingsManager.get_setting(None, ...) → full settings dict`` # behavior. With the helper guard in place, this branch is never # reached; without the guard, the dict would leak through. buggy_dict = { "llm.openai.api_key": "sk-leaked-openai-key", "llm.anthropic.api_key": "sk-ant-leaked-key", } def _sm_get_setting(key, default=None, *_args, **_kwargs): if key is None: return buggy_dict if key == "llm.lmstudio.url": return "http://localhost:1234/v1" return default if default is not None else "" mock_sm = MagicMock() mock_sm.get_setting.side_effect = _sm_get_setting with _authenticated_client(app) as (client, mock_session): mock_session.query.return_value = mock_cache_query mock_session.query.return_value.delete.return_value = 0 with patch( "local_deep_research.llm.providers.get_discovered_provider_options", return_value=[], ): with patch( f"{MODULE}.get_settings_manager", return_value=mock_sm ): with patch(f"{MODULE}.safe_get") as mock_safe_get: mock_ollama_resp = MagicMock() mock_ollama_resp.status_code = 200 mock_ollama_resp.text = '{"models": []}' mock_ollama_resp.json.return_value = {"models": []} mock_safe_get.return_value = mock_ollama_resp with patch( "local_deep_research.llm.providers.discover_providers", return_value={"lmstudio": mock_provider_info}, ): resp = client.get( f"{SETTINGS_PREFIX}/api/available-models?force_refresh=true" ) assert resp.status_code == 200 # Critical assertions: api_key reaching the provider is a string, # not the buggy settings dict that would leak other providers' keys. mock_provider_class.list_models_for_api.assert_called_once() call_args = mock_provider_class.list_models_for_api.call_args passed_api_key = ( call_args.args[0] if call_args.args else call_args.kwargs.get("api_key") ) assert isinstance(passed_api_key, str), ( f"api_key must be a string, got {type(passed_api_key).__name__}" ) assert not isinstance(passed_api_key, dict) assert passed_api_key != buggy_dict # Confirm the production helper ran (was not silently mocked away): # for the URL setting it must have reached the mock manager. assert any( call.args and call.args[0] == "llm.lmstudio.url" for call in mock_sm.get_setting.call_args_list ), ( "Expected the production helper to call " "SettingsManager.get_setting('llm.lmstudio.url', ...); " "the mock manager was never invoked, so the test isn't " "exercising the production code path." ) # Conversely, the helper must NOT have called get_setting(None, ...) # — the guard short-circuits that branch before delegating. assert not any( call.args and call.args[0] is None for call in mock_sm.get_setting.call_args_list ), ( "The helper guard at _get_setting_from_session must short-circuit " "key=None to default; instead it delegated to the manager." )