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
1530 lines
51 KiB
Python
1530 lines
51 KiB
Python
"""
|
|
Tests for the SettingsManager class.
|
|
|
|
Tests cover:
|
|
- Initialization
|
|
- Getting settings
|
|
- Setting values
|
|
- Thread safety checks
|
|
- Default settings loading
|
|
- Environment variable overrides
|
|
"""
|
|
|
|
import os
|
|
import threading
|
|
from unittest.mock import Mock, patch
|
|
|
|
|
|
class MockSetting:
|
|
"""Mock Setting model for testing."""
|
|
|
|
def __init__(
|
|
self,
|
|
key,
|
|
value,
|
|
ui_element="text",
|
|
editable=True,
|
|
type=None,
|
|
name=None,
|
|
description=None,
|
|
category=None,
|
|
options=None,
|
|
min_value=None,
|
|
max_value=None,
|
|
step=None,
|
|
visible=True,
|
|
):
|
|
self.key = key
|
|
self.value = value
|
|
self.ui_element = ui_element
|
|
self.editable = editable
|
|
self.type = type or Mock(name="APP")
|
|
self.name = name or key
|
|
self.description = description or ""
|
|
self.category = category
|
|
self.options = options
|
|
self.min_value = min_value
|
|
self.max_value = max_value
|
|
self.step = step
|
|
self.visible = visible
|
|
self.updated_at = None
|
|
|
|
|
|
class TestSettingsManagerInit:
|
|
"""Tests for SettingsManager initialization."""
|
|
|
|
def test_init_stores_db_session(self):
|
|
"""SettingsManager stores the database session."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
assert manager.db_session is mock_session
|
|
|
|
def test_init_without_db_session(self):
|
|
"""SettingsManager can be initialized without db session."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
manager = SettingsManager()
|
|
|
|
assert manager.db_session is None
|
|
|
|
def test_init_stores_thread_id(self):
|
|
"""SettingsManager stores creation thread ID."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
manager = SettingsManager()
|
|
|
|
assert manager._creation_thread_id == threading.get_ident()
|
|
|
|
|
|
class TestCheckEnvSetting:
|
|
"""Tests for check_env_setting function."""
|
|
|
|
def test_check_env_setting_found(self):
|
|
"""check_env_setting returns value when env var is set."""
|
|
from local_deep_research.settings.manager import (
|
|
check_env_setting,
|
|
)
|
|
|
|
# Set environment variable
|
|
os.environ["LDR_APP_DEBUG"] = "true"
|
|
try:
|
|
result = check_env_setting("app.debug")
|
|
assert result == "true"
|
|
finally:
|
|
del os.environ["LDR_APP_DEBUG"]
|
|
|
|
def test_check_env_setting_not_found(self):
|
|
"""check_env_setting returns None when env var not set."""
|
|
from local_deep_research.settings.manager import (
|
|
check_env_setting,
|
|
)
|
|
|
|
# Make sure env var is not set
|
|
if "LDR_APP_NONEXISTENT" in os.environ:
|
|
del os.environ["LDR_APP_NONEXISTENT"]
|
|
|
|
result = check_env_setting("app.nonexistent")
|
|
assert result is None
|
|
|
|
def test_check_env_setting_converts_dots_to_underscores(self):
|
|
"""check_env_setting converts dots to underscores in key."""
|
|
from local_deep_research.settings.manager import (
|
|
check_env_setting,
|
|
)
|
|
|
|
os.environ["LDR_LLM_MODEL_NAME"] = "test-model"
|
|
try:
|
|
result = check_env_setting("llm.model.name")
|
|
assert result == "test-model"
|
|
finally:
|
|
del os.environ["LDR_LLM_MODEL_NAME"]
|
|
|
|
|
|
class TestSettingsManagerGetSetting:
|
|
"""Tests for SettingsManager.get_setting method."""
|
|
|
|
def test_get_setting_from_db(self):
|
|
"""get_setting retrieves value from database."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.debug", value=True, ui_element="checkbox"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_setting("app.debug", check_env=False)
|
|
|
|
assert result is True
|
|
|
|
def test_get_setting_returns_default_when_not_found(self):
|
|
"""get_setting returns default when setting not in database."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.filter.return_value.all.return_value = []
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_setting("app.nonexistent", default="default_value")
|
|
|
|
assert result == "default_value"
|
|
|
|
def test_get_setting_with_env_override(self):
|
|
"""get_setting uses environment variable when available."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.test_value", value="db_value", ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
os.environ["LDR_APP_TEST_VALUE"] = "env_value"
|
|
try:
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_setting("app.test_value", check_env=True)
|
|
|
|
assert result == "env_value"
|
|
finally:
|
|
del os.environ["LDR_APP_TEST_VALUE"]
|
|
|
|
def test_get_setting_number_type(self):
|
|
"""get_setting correctly converts number types."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="llm.temperature", value=0.7, ui_element="number"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_setting("llm.temperature", check_env=False)
|
|
|
|
assert result == 0.7
|
|
assert isinstance(result, float)
|
|
|
|
|
|
class TestSettingsManagerSetSetting:
|
|
"""Tests for SettingsManager.set_setting method."""
|
|
|
|
def test_set_setting_updates_existing(self):
|
|
"""set_setting updates existing setting in database."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(key="app.debug", value=False, editable=True)
|
|
mock_query.filter.return_value.first.return_value = mock_setting
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
with patch.object(manager, "_emit_settings_changed"):
|
|
result = manager.set_setting("app.debug", True)
|
|
|
|
assert result is True
|
|
assert mock_setting.value is True
|
|
mock_session.commit.assert_called_once()
|
|
|
|
def test_set_setting_fails_without_db_session(self):
|
|
"""set_setting returns False when no db session."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
manager = SettingsManager()
|
|
result = manager.set_setting("app.debug", True)
|
|
|
|
assert result is False
|
|
|
|
def test_set_setting_fails_for_non_editable(self):
|
|
"""set_setting returns False for non-editable settings."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.version", value="1.0", editable=False
|
|
)
|
|
mock_query.filter.return_value.first.return_value = mock_setting
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.set_setting("app.version", "2.0")
|
|
|
|
assert result is False
|
|
|
|
def test_set_setting_creates_new(self):
|
|
"""set_setting creates new setting when not exists."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.filter.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
with patch.object(manager, "_emit_settings_changed"):
|
|
result = manager.set_setting("app.new_setting", "value")
|
|
|
|
assert result is True
|
|
mock_session.add.assert_called_once()
|
|
|
|
|
|
class TestSettingsManagerThreadSafety:
|
|
"""Tests for SettingsManager thread safety."""
|
|
|
|
def test_thread_safety_check_same_thread(self):
|
|
"""_check_thread_safety passes in same thread."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
# Should not raise
|
|
manager._check_thread_safety()
|
|
|
|
def test_thread_safety_check_different_thread(self):
|
|
"""_check_thread_safety raises in different thread."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
error_raised = [False]
|
|
|
|
def run_in_thread():
|
|
try:
|
|
manager._check_thread_safety()
|
|
except RuntimeError as e:
|
|
if "not thread-safe" in str(e):
|
|
error_raised[0] = True
|
|
|
|
thread = threading.Thread(target=run_in_thread)
|
|
thread.start()
|
|
thread.join()
|
|
|
|
assert error_raised[0] is True
|
|
|
|
|
|
class TestSettingsManagerGetAllSettings:
|
|
"""Tests for SettingsManager.get_all_settings method."""
|
|
|
|
def test_get_all_settings_returns_dict(self):
|
|
"""get_all_settings returns dictionary of all settings."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_settings = [
|
|
MockSetting(key="app.debug", value=True, ui_element="checkbox"),
|
|
MockSetting(key="llm.model", value="gpt-4", ui_element="text"),
|
|
]
|
|
mock_query.all.return_value = mock_settings
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_all_settings()
|
|
|
|
assert "app.debug" in result
|
|
assert "llm.model" in result
|
|
assert result["app.debug"]["value"] is True
|
|
assert result["llm.model"]["value"] == "gpt-4"
|
|
|
|
|
|
class TestSettingsManagerDeleteSetting:
|
|
"""Tests for SettingsManager.delete_setting method."""
|
|
|
|
def test_delete_setting_success(self):
|
|
"""delete_setting removes setting from database."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.filter.return_value.delete.return_value = 1
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.delete_setting("app.old_setting")
|
|
|
|
assert result is True
|
|
mock_session.commit.assert_called_once()
|
|
|
|
def test_delete_setting_not_found(self):
|
|
"""delete_setting returns False when setting not found."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.filter.return_value.delete.return_value = 0
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.delete_setting("app.nonexistent")
|
|
|
|
assert result is False
|
|
|
|
def test_delete_setting_fails_without_db(self):
|
|
"""delete_setting returns False without db session."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
manager = SettingsManager()
|
|
result = manager.delete_setting("app.setting")
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestSettingsManagerLocking:
|
|
"""Tests for settings locking functionality."""
|
|
|
|
def test_settings_locked_property(self):
|
|
"""settings_locked returns lock status."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.lock_settings", value=True, ui_element="checkbox"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
# Access property - should query and cache
|
|
assert manager.settings_locked is True
|
|
|
|
def test_set_setting_blocked_when_locked(self):
|
|
"""set_setting returns False when settings are locked."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
# Force settings to be locked
|
|
manager._SettingsManager__settings_locked = True
|
|
|
|
result = manager.set_setting("app.debug", True)
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestSettingsManagerCreateOrUpdate:
|
|
"""Tests for SettingsManager.create_or_update_setting method."""
|
|
|
|
def test_create_setting_with_dict(self):
|
|
"""create_or_update_setting accepts dictionary input."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.filter.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
with patch.object(manager, "_emit_settings_changed"):
|
|
result = manager.create_or_update_setting(
|
|
{
|
|
"key": "app.new_setting",
|
|
"value": "test_value",
|
|
"name": "New Setting",
|
|
"description": "A new setting",
|
|
}
|
|
)
|
|
|
|
assert result is not None
|
|
mock_session.add.assert_called_once()
|
|
|
|
def test_update_existing_setting(self):
|
|
"""create_or_update_setting updates existing setting."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.existing", value="old_value", editable=True
|
|
)
|
|
mock_query.filter.return_value.first.return_value = mock_setting
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
with patch.object(manager, "_emit_settings_changed"):
|
|
result = manager.create_or_update_setting(
|
|
{
|
|
"key": "app.existing",
|
|
"value": "new_value",
|
|
"name": "Existing Setting",
|
|
"description": "Updated",
|
|
}
|
|
)
|
|
|
|
assert result is not None
|
|
assert mock_setting.value == "new_value"
|
|
|
|
def test_create_or_update_fails_without_db(self):
|
|
"""create_or_update_setting returns None without db session."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
manager = SettingsManager()
|
|
result = manager.create_or_update_setting(
|
|
{"key": "app.test", "value": "value"}
|
|
)
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestSettingsManagerVersionCheck:
|
|
"""Tests for version checking functionality."""
|
|
|
|
def test_db_version_matches_package_true(self):
|
|
"""db_version_matches_package returns True when versions match."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
from local_deep_research.__version__ import __version__
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.version", value=__version__, ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.db_version_matches_package()
|
|
|
|
assert result is True
|
|
|
|
def test_db_version_matches_package_false(self):
|
|
"""db_version_matches_package returns False when versions differ."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.version", value="0.0.1", ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.db_version_matches_package()
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestGetSettingsSnapshot:
|
|
"""
|
|
Tests for get_settings_snapshot() method.
|
|
|
|
This method is required by rag_routes.py which needs both get_bool_setting()
|
|
and get_settings_snapshot() methods on SettingsManager.
|
|
"""
|
|
|
|
def test_get_settings_snapshot_exists(self):
|
|
"""Verify get_settings_snapshot method exists on SettingsManager."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
manager = SettingsManager()
|
|
assert hasattr(manager, "get_settings_snapshot"), (
|
|
"SettingsManager must have get_settings_snapshot method. "
|
|
"This is required by rag_routes.py for background thread operations."
|
|
)
|
|
assert callable(manager.get_settings_snapshot)
|
|
|
|
def test_get_settings_snapshot_returns_dict(self):
|
|
"""get_settings_snapshot returns a dictionary."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.all.return_value = []
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_settings_snapshot()
|
|
|
|
assert isinstance(result, dict)
|
|
|
|
def test_get_settings_snapshot_returns_flat_dict(self):
|
|
"""get_settings_snapshot returns flat key-value pairs, not nested metadata."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_settings = [
|
|
MockSetting(key="app.debug", value=True, ui_element="checkbox"),
|
|
MockSetting(key="app.name", value="TestApp", ui_element="text"),
|
|
MockSetting(
|
|
key="search.max_results", value=50, ui_element="number"
|
|
),
|
|
]
|
|
mock_query.all.return_value = mock_settings
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_settings_snapshot()
|
|
|
|
# Should be flat key-value pairs
|
|
assert "app.debug" in result
|
|
assert "app.name" in result
|
|
assert "search.max_results" in result
|
|
|
|
# Values should be direct values, not wrapped in {"value": ...}
|
|
assert result["app.debug"] is True
|
|
assert result["app.name"] == "TestApp"
|
|
assert result["search.max_results"] == 50
|
|
|
|
def test_get_settings_snapshot_excludes_metadata(self):
|
|
"""get_settings_snapshot excludes metadata fields like ui_element, editable."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_settings = [
|
|
MockSetting(
|
|
key="test.setting",
|
|
value="test_value",
|
|
ui_element="text",
|
|
editable=True,
|
|
description="A test setting",
|
|
),
|
|
]
|
|
mock_query.all.return_value = mock_settings
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_settings_snapshot()
|
|
|
|
# The value should be the setting value, not a metadata dict
|
|
assert result.get("test.setting") == "test_value"
|
|
# Should not contain metadata fields at the top level
|
|
assert "ui_element" not in result
|
|
assert "editable" not in result
|
|
assert "description" not in result
|
|
|
|
def test_get_settings_snapshot_supports_dict_assignment(self):
|
|
"""get_settings_snapshot result can be modified with dict assignment."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.all.return_value = []
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_settings_snapshot()
|
|
|
|
# This is how rag_routes.py uses the snapshot
|
|
result["_username"] = "testuser"
|
|
assert result["_username"] == "testuser"
|
|
|
|
def test_get_settings_snapshot_preserves_value_types(self):
|
|
"""get_settings_snapshot preserves the type of setting values."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_settings = [
|
|
MockSetting(key="bool_setting", value=True, ui_element="checkbox"),
|
|
MockSetting(key="int_setting", value=42, ui_element="number"),
|
|
MockSetting(key="str_setting", value="hello", ui_element="text"),
|
|
MockSetting(key="none_setting", value=None, ui_element="text"),
|
|
]
|
|
mock_query.all.return_value = mock_settings
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_settings_snapshot()
|
|
|
|
assert result["bool_setting"] is True
|
|
assert result["int_setting"] == 42
|
|
assert result["str_setting"] == "hello"
|
|
assert result["none_setting"] is None
|
|
|
|
def test_get_settings_snapshot_empty_database(self):
|
|
"""get_settings_snapshot works with empty database."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.all.return_value = []
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_settings_snapshot()
|
|
|
|
assert isinstance(result, dict)
|
|
# May contain default settings from files, but should not raise
|
|
|
|
|
|
class TestGetBoolSettingMethod:
|
|
"""
|
|
Tests for get_bool_setting() method.
|
|
|
|
This method is critical for rag_routes.py which calls it at lines
|
|
133, 2185, and 2309 to get boolean settings.
|
|
"""
|
|
|
|
def test_get_bool_setting_exists(self):
|
|
"""Verify get_bool_setting method exists on SettingsManager."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
manager = SettingsManager()
|
|
assert hasattr(manager, "get_bool_setting"), (
|
|
"SettingsManager must have get_bool_setting method. "
|
|
"This is required by rag_routes.py for boolean settings."
|
|
)
|
|
assert callable(manager.get_bool_setting)
|
|
|
|
def test_get_bool_setting_returns_bool_type(self):
|
|
"""get_bool_setting always returns a boolean type."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
# Return empty list to simulate setting not found
|
|
mock_query.filter.return_value.all.return_value = []
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
result_true = manager.get_bool_setting(
|
|
"nonexistent", True, check_env=False
|
|
)
|
|
result_false = manager.get_bool_setting(
|
|
"nonexistent", False, check_env=False
|
|
)
|
|
|
|
assert isinstance(result_true, bool)
|
|
assert isinstance(result_false, bool)
|
|
assert result_true is True
|
|
assert result_false is False
|
|
|
|
def test_get_bool_setting_string_true_values(self):
|
|
"""get_bool_setting converts truthy strings to True."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
# Use ui_element="text" to avoid checkbox type conversion
|
|
# which would use Python's bool() on the raw string
|
|
truthy_values = ["true", "True", "TRUE", "1", "yes", "on", "enabled"]
|
|
|
|
for value in truthy_values:
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value=value, ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"test.bool", False, check_env=False
|
|
)
|
|
|
|
assert result is True, f"Expected True for value '{value}'"
|
|
|
|
def test_get_bool_setting_string_false_values(self):
|
|
"""get_bool_setting converts falsy strings to False."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
# Use ui_element="text" to avoid checkbox type conversion
|
|
# The to_bool() function treats strings not in truthy list as False
|
|
falsy_values = ["false", "False", "FALSE", "0", "no", "off"]
|
|
|
|
for value in falsy_values:
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value=value, ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"test.bool", True, check_env=False
|
|
)
|
|
|
|
assert result is False, f"Expected False for value '{value}'"
|
|
|
|
def test_get_bool_setting_boolean_passthrough(self):
|
|
"""get_bool_setting passes through actual boolean values."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
for bool_value in [True, False]:
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value=bool_value, ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"test.bool", not bool_value, check_env=False
|
|
)
|
|
|
|
assert result is bool_value
|
|
|
|
def test_get_bool_setting_signature(self):
|
|
"""get_bool_setting has correct signature: (key, default, check_env)."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
import inspect
|
|
|
|
manager = SettingsManager()
|
|
sig = inspect.signature(manager.get_bool_setting)
|
|
params = list(sig.parameters.keys())
|
|
|
|
assert "key" in params
|
|
assert "default" in params
|
|
assert "check_env" in params
|
|
|
|
|
|
class TestSettingsManagerApiCompatibility:
|
|
"""
|
|
Tests to ensure settings.manager.SettingsManager has all
|
|
methods required by code that was previously using separate implementations.
|
|
|
|
These tests prevent regression of issue #1877.
|
|
"""
|
|
|
|
def test_has_all_required_methods_for_rag_routes(self):
|
|
"""SettingsManager has all methods required by rag_routes.py."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
manager = SettingsManager()
|
|
|
|
# Methods called in rag_routes.py
|
|
required_methods = [
|
|
"get_setting",
|
|
"get_bool_setting",
|
|
"get_settings_snapshot",
|
|
]
|
|
|
|
for method in required_methods:
|
|
assert hasattr(manager, method), (
|
|
f"SettingsManager missing required method: {method}"
|
|
)
|
|
assert callable(getattr(manager, method)), (
|
|
f"SettingsManager.{method} must be callable"
|
|
)
|
|
|
|
def test_get_setting_and_get_bool_setting_consistency(self):
|
|
"""get_bool_setting should convert get_setting result to bool."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value="true", ui_element="text"
|
|
)
|
|
# Use correct mock pattern - filter.return_value.all.return_value
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
# get_setting returns raw value
|
|
raw = manager.get_setting("test.bool", check_env=False)
|
|
# get_bool_setting returns converted boolean
|
|
converted = manager.get_bool_setting(
|
|
"test.bool", False, check_env=False
|
|
)
|
|
|
|
assert raw == "true"
|
|
assert converted is True
|
|
|
|
|
|
class TestGetBoolSettingEdgeCases:
|
|
"""
|
|
Edge case tests for get_bool_setting() method.
|
|
|
|
These tests cover whitespace handling, case sensitivity,
|
|
None values, and other edge cases.
|
|
"""
|
|
|
|
def test_get_bool_setting_whitespace_handling(self):
|
|
"""get_bool_setting handles values with leading/trailing whitespace."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
# Test whitespace around truthy values
|
|
whitespace_truthy = [" true ", "\ttrue\n", " 1 ", " yes "]
|
|
|
|
for value in whitespace_truthy:
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value=value, ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
# Note: to_bool uses .lower() which doesn't strip whitespace
|
|
# This test documents actual behavior
|
|
result = manager.get_bool_setting(
|
|
"test.bool", False, check_env=False
|
|
)
|
|
# Whitespace values may not match truthy list exactly
|
|
assert isinstance(result, bool)
|
|
|
|
def test_get_bool_setting_mixed_case(self):
|
|
"""get_bool_setting handles mixed case values."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mixed_case_values = [
|
|
"TrUe",
|
|
"TRUE",
|
|
"True",
|
|
"tRuE",
|
|
"YES",
|
|
"Yes",
|
|
"yEs",
|
|
]
|
|
|
|
for value in mixed_case_values:
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value=value, ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"test.bool", False, check_env=False
|
|
)
|
|
|
|
assert result is True, (
|
|
f"Expected True for mixed case value '{value}'"
|
|
)
|
|
|
|
def test_get_bool_setting_none_value_behavior(self):
|
|
"""get_bool_setting handles None value from database.
|
|
|
|
When a setting exists with value=None, get_typed_setting_value
|
|
returns the default value (since value is None). So
|
|
get_bool_setting returns to_bool(default, default) = default.
|
|
"""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value=None, ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
# When value is None, get_typed_setting_value returns the default
|
|
# to_bool(True, True) = True
|
|
result = manager.get_bool_setting("test.bool", True, check_env=False)
|
|
assert result is True # Returns the default when DB value is None
|
|
|
|
def test_get_bool_setting_empty_string_is_false(self):
|
|
"""get_bool_setting treats empty string as False."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(key="test.bool", value="", ui_element="text")
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting("test.bool", True, check_env=False)
|
|
|
|
# Empty string should be falsy
|
|
assert result is False
|
|
|
|
def test_get_bool_setting_numeric_zero_is_false(self):
|
|
"""get_bool_setting treats numeric 0 as False."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value=0, ui_element="number"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting("test.bool", True, check_env=False)
|
|
|
|
assert result is False
|
|
|
|
def test_get_bool_setting_numeric_nonzero_is_true(self):
|
|
"""get_bool_setting treats non-zero numbers as True."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
for num_value in [1, 42, -1, 0.5, 100]:
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="test.bool", value=num_value, ui_element="number"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"test.bool", False, check_env=False
|
|
)
|
|
|
|
assert result is True, (
|
|
f"Expected True for numeric value {num_value}"
|
|
)
|
|
|
|
|
|
class TestGetBoolSettingEnvOverride:
|
|
"""
|
|
Tests for get_bool_setting() environment variable handling.
|
|
|
|
These tests verify that environment variables properly override
|
|
database settings when check_env=True.
|
|
"""
|
|
|
|
def test_get_bool_setting_env_override_true(self):
|
|
"""get_bool_setting uses env var when check_env=True."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
# DB has False
|
|
mock_setting = MockSetting(
|
|
key="app.test_flag", value=False, ui_element="checkbox"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
# Env has "true"
|
|
os.environ["LDR_APP_TEST_FLAG"] = "true"
|
|
try:
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"app.test_flag", False, check_env=True
|
|
)
|
|
|
|
# Env should override DB
|
|
assert result is True
|
|
finally:
|
|
del os.environ["LDR_APP_TEST_FLAG"]
|
|
|
|
def test_get_bool_setting_env_override_false_checkbox_behavior(self):
|
|
"""Document behavior: env var 'false' with checkbox ui_element.
|
|
|
|
The unified SettingsManager uses parse_boolean() for checkbox type
|
|
conversion, which correctly handles "false" -> False (unlike
|
|
Python's bool("false") = True).
|
|
"""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.test_flag", value=True, ui_element="checkbox"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
# Env has "false" - parse_boolean("false") correctly returns False
|
|
os.environ["LDR_APP_TEST_FLAG"] = "false"
|
|
try:
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"app.test_flag", False, check_env=True
|
|
)
|
|
|
|
# parse_boolean("false") = False (correct behavior)
|
|
assert result is False
|
|
finally:
|
|
del os.environ["LDR_APP_TEST_FLAG"]
|
|
|
|
def test_get_bool_setting_env_override_with_text_ui_element(self):
|
|
"""Env var 'false' works correctly with text ui_element.
|
|
|
|
When ui_element="text", the setting is returned as string,
|
|
then get_bool_setting uses to_bool() which correctly handles
|
|
"false" as False.
|
|
"""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_setting = MockSetting(
|
|
key="app.test_flag", value="true", ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
# Env has "false" - with text ui_element this works correctly
|
|
os.environ["LDR_APP_TEST_FLAG"] = "false"
|
|
try:
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"app.test_flag", True, check_env=True
|
|
)
|
|
|
|
# to_bool("false") correctly returns False
|
|
assert result is False
|
|
finally:
|
|
del os.environ["LDR_APP_TEST_FLAG"]
|
|
|
|
def test_get_bool_setting_no_env_override_when_disabled(self):
|
|
"""get_bool_setting ignores env var when check_env=False."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
# DB has False
|
|
mock_setting = MockSetting(
|
|
key="app.test_flag", value=False, ui_element="checkbox"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
# Env has "true" but should be ignored
|
|
os.environ["LDR_APP_TEST_FLAG"] = "true"
|
|
try:
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"app.test_flag", True, check_env=False
|
|
)
|
|
|
|
# Should use DB value, not env
|
|
assert result is False
|
|
finally:
|
|
del os.environ["LDR_APP_TEST_FLAG"]
|
|
|
|
def test_get_bool_setting_env_numeric_string_with_text_ui(self):
|
|
"""get_bool_setting handles env var with '1' or '0' when ui_element=text."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
# Need a setting with text ui_element for string passthrough
|
|
mock_setting = MockSetting(
|
|
key="app.test_flag", value="default", ui_element="text"
|
|
)
|
|
mock_query.filter.return_value.all.return_value = [mock_setting]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
# Test "1" as truthy - to_bool("1") = True
|
|
os.environ["LDR_APP_TEST_FLAG"] = "1"
|
|
try:
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"app.test_flag", False, check_env=True
|
|
)
|
|
assert result is True
|
|
finally:
|
|
del os.environ["LDR_APP_TEST_FLAG"]
|
|
|
|
# Test "0" as falsy - to_bool("0") = False
|
|
os.environ["LDR_APP_TEST_FLAG"] = "0"
|
|
try:
|
|
manager = SettingsManager(db_session=mock_session)
|
|
result = manager.get_bool_setting(
|
|
"app.test_flag", True, check_env=True
|
|
)
|
|
assert result is False
|
|
finally:
|
|
del os.environ["LDR_APP_TEST_FLAG"]
|
|
|
|
|
|
class TestSettingsManagerErrorHandling:
|
|
"""
|
|
Tests for error handling in SettingsManager methods.
|
|
|
|
These tests verify graceful degradation when database
|
|
or other components fail.
|
|
"""
|
|
|
|
def test_get_bool_setting_db_error_returns_default(self):
|
|
"""get_bool_setting returns default when DB query fails."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_query.filter.side_effect = SQLAlchemyError("DB connection lost")
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
# Should return default on error, not raise
|
|
result = manager.get_bool_setting("test.setting", True, check_env=False)
|
|
assert result is True
|
|
|
|
result = manager.get_bool_setting(
|
|
"test.setting", False, check_env=False
|
|
)
|
|
assert result is False
|
|
|
|
def test_get_setting_without_db_session_returns_default(self):
|
|
"""get_setting returns default when no DB session available."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
# No DB session
|
|
manager = SettingsManager(db_session=None)
|
|
|
|
result = manager.get_setting("test.setting", default="fallback")
|
|
# Should return default from default_settings or the provided default
|
|
assert result is not None or result == "fallback"
|
|
|
|
def test_get_bool_setting_without_db_session(self):
|
|
"""get_bool_setting works without DB session using defaults."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
# No DB session
|
|
manager = SettingsManager(db_session=None)
|
|
|
|
# Should return the provided default
|
|
result = manager.get_bool_setting(
|
|
"nonexistent.setting", True, check_env=False
|
|
)
|
|
assert result is True
|
|
|
|
|
|
class TestSettingsManagerBackgroundThreadUsage:
|
|
"""
|
|
Tests for SettingsManager usage in background threads.
|
|
|
|
These tests verify that SettingsManager works correctly
|
|
when used from background threads (like in trigger_auto_index
|
|
and _get_rag_service_for_thread).
|
|
"""
|
|
|
|
def test_settings_manager_in_different_thread(self):
|
|
"""SettingsManager can be created and used in a different thread."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
import threading
|
|
import queue
|
|
|
|
result_queue = queue.Queue()
|
|
error_queue = queue.Queue()
|
|
|
|
def thread_func():
|
|
try:
|
|
# Create manager in this thread
|
|
manager = SettingsManager(db_session=None)
|
|
|
|
# Verify methods exist
|
|
assert hasattr(manager, "get_bool_setting")
|
|
assert callable(manager.get_bool_setting)
|
|
|
|
# Call get_bool_setting
|
|
result = manager.get_bool_setting(
|
|
"test.setting", True, check_env=False
|
|
)
|
|
result_queue.put(result)
|
|
except Exception as e:
|
|
error_queue.put(e)
|
|
|
|
thread = threading.Thread(target=thread_func)
|
|
thread.start()
|
|
thread.join(timeout=5)
|
|
|
|
# Check for errors
|
|
if not error_queue.empty():
|
|
raise error_queue.get()
|
|
|
|
# Verify result
|
|
assert not result_queue.empty()
|
|
result = result_queue.get()
|
|
assert isinstance(result, bool)
|
|
|
|
def test_concurrent_get_bool_setting_calls(self):
|
|
"""Multiple threads can call get_bool_setting concurrently."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
import threading
|
|
import queue
|
|
|
|
num_threads = 5
|
|
results = queue.Queue()
|
|
errors = queue.Queue()
|
|
|
|
def thread_func(thread_id):
|
|
try:
|
|
manager = SettingsManager(db_session=None)
|
|
for _ in range(10):
|
|
result = manager.get_bool_setting(
|
|
f"test.setting.{thread_id}",
|
|
thread_id % 2 == 0,
|
|
check_env=False,
|
|
)
|
|
results.put((thread_id, result))
|
|
except Exception as e:
|
|
errors.put((thread_id, e))
|
|
|
|
threads = []
|
|
for i in range(num_threads):
|
|
t = threading.Thread(target=thread_func, args=(i,))
|
|
threads.append(t)
|
|
t.start()
|
|
|
|
for t in threads:
|
|
t.join(timeout=10)
|
|
|
|
# Check no errors occurred
|
|
if not errors.empty():
|
|
thread_id, error = errors.get()
|
|
raise AssertionError(f"Thread {thread_id} failed: {error}")
|
|
|
|
# Verify we got results from all threads
|
|
assert results.qsize() == num_threads * 10
|
|
|
|
def test_get_settings_snapshot_in_thread(self):
|
|
"""get_settings_snapshot works correctly in background thread."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
import threading
|
|
import queue
|
|
|
|
result_queue = queue.Queue()
|
|
error_queue = queue.Queue()
|
|
|
|
def thread_func():
|
|
try:
|
|
manager = SettingsManager(db_session=None)
|
|
snapshot = manager.get_settings_snapshot()
|
|
|
|
# Verify it's a dict
|
|
assert isinstance(snapshot, dict)
|
|
|
|
# Verify we can modify it (as rag_routes does)
|
|
snapshot["_username"] = "test_user"
|
|
assert snapshot["_username"] == "test_user"
|
|
|
|
result_queue.put(snapshot)
|
|
except Exception as e:
|
|
error_queue.put(e)
|
|
|
|
thread = threading.Thread(target=thread_func)
|
|
thread.start()
|
|
thread.join(timeout=5)
|
|
|
|
if not error_queue.empty():
|
|
raise error_queue.get()
|
|
|
|
assert not result_queue.empty()
|
|
|
|
|
|
class TestGetSettingsSnapshotAdvanced:
|
|
"""
|
|
Advanced tests for get_settings_snapshot() method.
|
|
|
|
These tests cover mutation, complex values, and integration patterns.
|
|
"""
|
|
|
|
def test_snapshot_mutation_isolation(self):
|
|
"""Modifying snapshot doesn't affect subsequent calls."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_settings = [
|
|
MockSetting(key="test.key", value="original", ui_element="text"),
|
|
]
|
|
mock_query.all.return_value = mock_settings
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
# Get first snapshot
|
|
snapshot1 = manager.get_settings_snapshot()
|
|
|
|
# Modify it
|
|
snapshot1["test.key"] = "modified"
|
|
snapshot1["new.key"] = "added"
|
|
|
|
# Get second snapshot
|
|
snapshot2 = manager.get_settings_snapshot()
|
|
|
|
# Second snapshot should not be affected by modifications to first
|
|
assert snapshot2.get("test.key") == "original"
|
|
assert "new.key" not in snapshot2
|
|
|
|
def test_snapshot_with_complex_values(self):
|
|
"""Snapshot preserves complex value types like lists and dicts."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_settings = [
|
|
MockSetting(
|
|
key="list.setting",
|
|
value=["item1", "item2", "item3"],
|
|
ui_element="json",
|
|
),
|
|
MockSetting(
|
|
key="dict.setting",
|
|
value={"nested": {"key": "value"}},
|
|
ui_element="json",
|
|
),
|
|
]
|
|
mock_query.all.return_value = mock_settings
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
snapshot = manager.get_settings_snapshot()
|
|
|
|
# Verify complex types are preserved
|
|
assert snapshot["list.setting"] == ["item1", "item2", "item3"]
|
|
assert snapshot["dict.setting"] == {"nested": {"key": "value"}}
|
|
|
|
def test_snapshot_rag_service_usage_pattern(self):
|
|
"""Test the exact usage pattern from _get_rag_service_for_thread."""
|
|
from local_deep_research.settings.manager import (
|
|
SettingsManager,
|
|
)
|
|
|
|
mock_session = Mock()
|
|
mock_query = Mock()
|
|
mock_settings = [
|
|
MockSetting(
|
|
key="local_search_embedding_model",
|
|
value="all-MiniLM-L6-v2",
|
|
ui_element="text",
|
|
),
|
|
MockSetting(
|
|
key="local_search_embedding_provider",
|
|
value="sentence_transformers",
|
|
ui_element="text",
|
|
),
|
|
]
|
|
mock_query.all.return_value = mock_settings
|
|
mock_session.query.return_value = mock_query
|
|
|
|
manager = SettingsManager(db_session=mock_session)
|
|
|
|
# This is the exact pattern from rag_routes.py line 2193-2194
|
|
settings_snapshot = manager.get_settings_snapshot()
|
|
settings_snapshot["_username"] = "test_user"
|
|
|
|
# Verify the pattern works
|
|
assert settings_snapshot["_username"] == "test_user"
|
|
assert "local_search_embedding_model" in settings_snapshot
|
|
|
|
# Verify it can be passed to other functions
|
|
def mock_embedding_manager(settings_snapshot):
|
|
return settings_snapshot.get("local_search_embedding_model")
|
|
|
|
result = mock_embedding_manager(settings_snapshot)
|
|
assert result == "all-MiniLM-L6-v2"
|