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
1380 lines
54 KiB
Python
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."
|
|
)
|