Files
learningcircuit--local-deep…/tests/web/routes/test_settings_routes_deep_coverage2.py
wehub-resource-sync 7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:08:55 +08:00

1380 lines
54 KiB
Python

"""
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 <full-settings-dict>`` 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."
)