"""Advanced tests for programmatic API settings functionality.""" import threading from local_deep_research.api.settings_utils import ( InMemorySettingsManager, create_settings_snapshot, extract_setting_value, ) class TestSettingsEdgeCases: """Test edge cases and error handling for settings.""" def test_deeply_nested_settings_override(self): """Test overriding deeply nested settings.""" snapshot = create_settings_snapshot( overrides={ "search.engines.arxiv.enabled": True, "search.engines.wikipedia.max_results": 20, "llm.anthropic.model": "claude-3-opus-20240229", } ) # Check nested settings were created properly assert "search.engines.arxiv.enabled" in snapshot assert snapshot["search.engines.arxiv.enabled"]["value"] is True assert "search.engines.wikipedia.max_results" in snapshot assert snapshot["search.engines.wikipedia.max_results"]["value"] == 20 assert "llm.anthropic.model" in snapshot assert ( snapshot["llm.anthropic.model"]["value"] == "claude-3-opus-20240229" ) def test_invalid_settings_handling(self): """Test handling of invalid settings.""" manager = InMemorySettingsManager() # Test various invalid operations assert manager.get_setting(None, "default") == "default" assert manager.get_setting("", "default") == "default" assert manager.set_setting(None, "value") is False assert manager.set_setting("", "value") is False assert manager.delete_setting(None) is False assert manager.delete_setting("") is False def test_settings_type_preservation(self): """Test that setting types are preserved correctly.""" snapshot = create_settings_snapshot( overrides={ "llm.temperature": 0.7, # float "search.max_results": 10, # int "search.safe_search": True, # bool "llm.provider": "openai", # string "custom.list": ["a", "b", "c"], # list "custom.dict": {"key": "value"}, # dict } ) # Check types are preserved assert isinstance(snapshot["llm.temperature"]["value"], float) assert isinstance(snapshot["search.max_results"]["value"], int) assert isinstance(snapshot["search.safe_search"]["value"], bool) assert isinstance(snapshot["llm.provider"]["value"], str) assert isinstance(snapshot["custom.list"]["value"], list) assert isinstance(snapshot["custom.dict"]["value"], dict) def test_settings_immutability(self): """Test that returned settings are immutable copies.""" manager = InMemorySettingsManager() settings1 = manager.get_all_settings() settings2 = manager.get_all_settings() # Modify settings1 if "llm.provider" in settings1: settings1["llm.provider"]["value"] = "modified" # settings2 should not be affected assert settings2["llm.provider"]["value"] != "modified" def test_empty_and_none_values(self): """Test handling of empty and None values.""" snapshot = create_settings_snapshot( overrides={ "empty.string": "", "none.value": None, "zero.value": 0, "false.value": False, "empty.list": [], "empty.dict": {}, } ) assert snapshot["empty.string"]["value"] == "" assert snapshot["none.value"]["value"] is None assert snapshot["zero.value"]["value"] == 0 assert snapshot["false.value"]["value"] is False assert snapshot["empty.list"]["value"] == [] assert snapshot["empty.dict"]["value"] == {} class TestSettingsThreadSafety: """Test thread safety of settings management.""" def test_concurrent_settings_access(self): """Test concurrent access to settings doesn't cause issues.""" # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: replace with deterministic concurrent invariant test). manager = InMemorySettingsManager() results = [] errors = [] def access_settings(thread_id): try: # Each thread tries to read and write settings for i in range(10): # Read _ = manager.get_setting("llm.temperature") # Write (if setting exists) new_value = 0.1 * thread_id + 0.01 * i manager.set_setting("llm.temperature", new_value) # Read again read_value = manager.get_setting("llm.temperature") results.append((thread_id, read_value)) except Exception as e: errors.append((thread_id, str(e))) # Create multiple threads threads = [] for i in range(5): t = threading.Thread(target=access_settings, args=(i,)) threads.append(t) t.start() # Wait for all threads for t in threads: t.join() # Should have no errors assert len(errors) == 0 # Should have results from all threads assert len(results) > 0 def test_settings_snapshot_isolation(self): """Test that settings snapshots are isolated between threads.""" snapshots = {} def create_snapshot(thread_id): # Each thread creates its own snapshot with unique settings snapshot = create_settings_snapshot( provider=f"provider_{thread_id}", temperature=0.1 * thread_id, overrides={f"thread.{thread_id}.custom": f"value_{thread_id}"}, ) snapshots[thread_id] = snapshot # Create snapshots in parallel threads = [] for i in range(5): t = threading.Thread(target=create_snapshot, args=(i,)) threads.append(t) t.start() for t in threads: t.join() # Verify each snapshot has its unique values for thread_id, snapshot in snapshots.items(): assert snapshot["llm.provider"]["value"] == f"provider_{thread_id}" assert snapshot["llm.temperature"]["value"] == 0.1 * thread_id assert ( snapshot[f"thread.{thread_id}.custom"]["value"] == f"value_{thread_id}" ) class TestSettingsIntegration: """Test integration between settings and other components.""" def test_settings_propagation_through_api(self): """Test that settings are properly propagated through the API.""" # Create custom settings settings_snapshot = create_settings_snapshot( provider="anthropic", temperature=0.3, overrides={ "llm.anthropic.api_key": "test-key", "llm.anthropic.model": "claude-3-opus-20240229", "search.tool": "searxng", "search.max_results": 15, "search.region": "us-en", }, ) # Test that settings can be extracted correctly assert ( extract_setting_value(settings_snapshot, "llm.provider") == "anthropic" ) assert ( extract_setting_value(settings_snapshot, "llm.temperature") == 0.3 ) assert ( extract_setting_value(settings_snapshot, "llm.anthropic.api_key") == "test-key" ) assert ( extract_setting_value(settings_snapshot, "search.tool") == "searxng" ) assert ( extract_setting_value(settings_snapshot, "search.max_results") == 15 ) # Test that settings structure is correct for use by components assert "llm.provider" in settings_snapshot assert isinstance(settings_snapshot["llm.provider"], dict) assert "value" in settings_snapshot["llm.provider"] # Test settings isolation snapshot2 = create_settings_snapshot(provider="openai", temperature=0.9) # Original snapshot should not be affected assert ( extract_setting_value(settings_snapshot, "llm.provider") == "anthropic" ) assert ( extract_setting_value(settings_snapshot, "llm.temperature") == 0.3 ) assert extract_setting_value(snapshot2, "llm.provider") == "openai" assert extract_setting_value(snapshot2, "llm.temperature") == 0.9 def test_extract_setting_value_edge_cases(self): """Test extract_setting_value with various edge cases.""" # Test with None snapshot assert extract_setting_value(None, "any.key", "default") == "default" # Test with empty snapshot assert extract_setting_value({}, "any.key", "default") == "default" # Test with nested structures complex_snapshot = { "simple": "direct_value", "nested": {"value": "nested_value", "extra": "data"}, "deep": {"value": {"nested": {"value": "deep_value"}}}, "list": {"value": [1, 2, 3]}, "null": {"value": None}, } assert ( extract_setting_value(complex_snapshot, "simple") == "direct_value" ) assert ( extract_setting_value(complex_snapshot, "nested") == "nested_value" ) assert extract_setting_value(complex_snapshot, "deep") == { "nested": {"value": "deep_value"} } assert extract_setting_value(complex_snapshot, "list") == [1, 2, 3] assert extract_setting_value(complex_snapshot, "null") is None class TestComplexSettingsScenarios: """Test complex real-world settings scenarios.""" def test_multiple_provider_configuration(self): """Test configuring multiple LLM providers in one snapshot.""" snapshot = create_settings_snapshot( provider="openai", # Primary provider api_key="openai-key", overrides={ # Configure multiple providers "llm.anthropic.api_key": "anthropic-key", "llm.anthropic.model": "claude-3-opus-20240229", "llm.ollama.base_url": "http://localhost:11434", "llm.ollama.model": "llama2", # Provider-specific settings "llm.openai.model": "gpt-4", "llm.openai.max_tokens": 4000, "llm.anthropic.max_tokens": 3000, }, ) # Verify all providers are configured assert snapshot["llm.provider"]["value"] == "openai" assert snapshot["llm.openai.api_key"]["value"] == "openai-key" assert snapshot["llm.openai.model"]["value"] == "gpt-4" assert snapshot["llm.openai.max_tokens"]["value"] == 4000 assert snapshot["llm.anthropic.api_key"]["value"] == "anthropic-key" assert ( snapshot["llm.anthropic.model"]["value"] == "claude-3-opus-20240229" ) assert snapshot["llm.anthropic.max_tokens"]["value"] == 3000 assert ( snapshot["llm.ollama.base_url"]["value"] == "http://localhost:11434" ) assert snapshot["llm.ollama.model"]["value"] == "llama2" def test_search_engine_configuration_matrix(self): """Test configuring multiple search engines with different settings.""" snapshot = create_settings_snapshot( overrides={ # Enable/disable specific engines "search.engines.searxng.enabled": True, "search.engines.duckduckgo.enabled": True, "search.engines.wikipedia.enabled": True, "search.engines.arxiv.enabled": False, # Engine-specific settings "search.engines.searxng.base_url": "https://searx.example.com", "search.engines.searxng.timeout": 10, "search.engines.duckduckgo.region": "us-en", "search.engines.duckduckgo.safe_search": True, "search.engines.wikipedia.language": "en", "search.engines.wikipedia.max_chars": 1000, # Global search settings "search.max_results": 20, "search.max_filtered_results": 10, "search.snippets_only": False, } ) # Verify engine configurations assert snapshot["search.engines.searxng.enabled"]["value"] is True assert ( snapshot["search.engines.searxng.base_url"]["value"] == "https://searx.example.com" ) assert ( snapshot["search.engines.duckduckgo.safe_search"]["value"] is True ) assert snapshot["search.engines.arxiv.enabled"]["value"] is False # Verify global settings assert snapshot["search.max_results"]["value"] == 20 assert snapshot["search.snippets_only"]["value"] is False def test_research_mode_specific_settings(self): """Test settings for different research modes.""" # Quick mode settings quick_snapshot = create_settings_snapshot( overrides={ "research.quick.max_iterations": 1, "research.quick.max_search_queries": 3, "research.quick.enable_deep_analysis": False, "llm.temperature": 0.7, # Higher for creativity } ) # Detailed mode settings detailed_snapshot = create_settings_snapshot( overrides={ "research.detailed.max_iterations": 5, "research.detailed.max_search_queries": 10, "research.detailed.enable_deep_analysis": True, "llm.temperature": 0.3, # Lower for accuracy "search.max_results": 30, # More results for detailed research } ) # Verify different configurations assert quick_snapshot["research.quick.max_iterations"]["value"] == 1 assert quick_snapshot["llm.temperature"]["value"] == 0.7 assert ( detailed_snapshot["research.detailed.max_iterations"]["value"] == 5 ) assert detailed_snapshot["llm.temperature"]["value"] == 0.3 assert detailed_snapshot["search.max_results"]["value"] == 30 def test_environment_variable_override_priority(self, monkeypatch): """Test that env vars have correct priority over defaults and overrides.""" # Set environment variables monkeypatch.setenv("LDR_LLM_PROVIDER", "env_provider") monkeypatch.setenv("LDR_LLM_TEMPERATURE", "0.9") # Create manager (picks up env vars) manager = InMemorySettingsManager() # Env vars should override defaults assert manager.get_setting("llm.provider") == "env_provider" assert manager.get_setting("llm.temperature") == 0.9 # But explicit overrides should take precedence over env vars snapshot = create_settings_snapshot( provider="explicit_provider", temperature=0.1 ) assert snapshot["llm.provider"]["value"] == "explicit_provider" assert snapshot["llm.temperature"]["value"] == 0.1 class TestSettingsPersistence: """Test settings persistence and isolation.""" def test_in_memory_settings_not_persisted(self): """Test that in-memory settings are not persisted between instances.""" # Create manager and modify settings manager1 = InMemorySettingsManager() original_temp = manager1.get_setting("llm.temperature") manager1.set_setting("llm.temperature", 0.999) # Create new manager manager2 = InMemorySettingsManager() # Should have original default, not modified value assert manager2.get_setting("llm.temperature") == original_temp assert manager2.get_setting("llm.temperature") != 0.999 def test_settings_snapshot_independence(self): """Test that settings snapshots are independent of each other.""" # Create base snapshot base_snapshot = create_settings_snapshot(provider="openai") # Create another snapshot with same base snapshot1 = create_settings_snapshot( base_settings=base_snapshot, temperature=0.5 ) # Create third snapshot with same base snapshot2 = create_settings_snapshot( base_settings=base_snapshot, temperature=0.9 ) # Verify independence assert base_snapshot["llm.provider"]["value"] == "openai" assert ( "llm.temperature" not in base_snapshot or base_snapshot["llm.temperature"]["value"] != 0.5 ) assert snapshot1["llm.temperature"]["value"] == 0.5 assert snapshot2["llm.temperature"]["value"] == 0.9 # Modifying one shouldn't affect others snapshot1["llm.provider"]["value"] = "modified" assert base_snapshot["llm.provider"]["value"] == "openai" assert snapshot2["llm.provider"]["value"] == "openai"