"""End-to-end tests for API settings in research workflows.""" from unittest.mock import patch, MagicMock from local_deep_research.api import quick_summary, detailed_research from local_deep_research.api.settings_utils import create_settings_snapshot class TestE2EResearchWithSettings: """Test end-to-end research workflows with various settings.""" @patch("local_deep_research.api.research_functions.get_llm") @patch("local_deep_research.api.research_functions.get_search") def test_quick_summary_full_flow(self, mock_get_search, mock_get_llm): """Test quick_summary with full settings propagation. Patches happen on `api.research_functions` because the names are imported into that module at import time — patching the original module (`config.llm_config.get_llm`) would not affect already-bound references. """ # Mock LLM mock_llm = MagicMock() mock_llm.invoke.return_value = MagicMock(content="Test summary") mock_get_llm.return_value = mock_llm # Mock search engine mock_search = MagicMock() mock_search.search.return_value = { "results": [ { "title": "Result 1", "url": "http://example.com/1", "snippet": "Snippet 1", }, { "title": "Result 2", "url": "http://example.com/2", "snippet": "Snippet 2", }, ] } mock_get_search.return_value = mock_search # Run quick summary with custom settings result = quick_summary( "What is quantum computing?", provider="anthropic", api_key="test-key", temperature=0.5, max_search_results=10, settings_override={ "llm.anthropic.model": "claude-3-opus-20240229", "search.tool": "duckduckgo", "search.region": "us-en", }, ) # Verify LLM was configured correctly mock_get_llm.assert_called() # Verify search was configured correctly. get_search() takes the # tool name as the first positional arg; everything else is in # settings_snapshot rather than dedicated kwargs. mock_get_search.assert_called() search_args, search_kwargs = mock_get_search.call_args assert search_args[0] == "duckduckgo" snapshot = search_kwargs["settings_snapshot"] assert snapshot["search.region"]["value"] == "us-en" assert snapshot["search.max_results"]["value"] == 10 # Verify result structure assert "summary" in result assert "findings" in result assert "iterations" in result @patch("local_deep_research.api.research_functions.get_llm") @patch("local_deep_research.api.research_functions.get_search") def test_detailed_research_full_flow(self, mock_get_search, mock_get_llm): """Test detailed_research with comprehensive settings. detailed_research accepts only `settings_snapshot` for configuration (unlike quick_summary which has provider/api_key shortcuts), so the test builds a snapshot via create_settings_snapshot and threads it in directly. Patch targets follow `api.research_functions` because that is where get_llm / get_search are imported and bound. """ # Mock LLM with different responses mock_llm = MagicMock() mock_llm.invoke.side_effect = [ MagicMock(content="Initial analysis"), MagicMock(content="Deeper analysis"), MagicMock(content="Final synthesis"), ] mock_get_llm.return_value = mock_llm mock_search = MagicMock() mock_get_search.return_value = mock_search snapshot = create_settings_snapshot( provider="openai", overrides={ "search.iterations": 3, "search.max_results": 20, "search.engines.arxiv.enabled": True, "llm.max_tokens": 4000, }, ) # Run detailed research with custom settings result = detailed_research( "Explain the applications of quantum computing in cryptography", settings_snapshot=snapshot, ) # The settings_snapshot values should be threaded through to # get_search. mock_get_search.assert_called() passed_snapshot = mock_get_search.call_args[1]["settings_snapshot"] assert passed_snapshot["search.max_results"]["value"] == 20 assert passed_snapshot["search.iterations"]["value"] == 3 # Verify the structured-report shape is returned (detailed_research # returns a report dict, not a summary dict). assert isinstance(result, dict) def test_settings_isolation_between_calls(self): """Test that settings don't leak between API calls.""" with patch( "local_deep_research.api.research_functions._init_search_system" ) as mock_init: mock_system = MagicMock() mock_system.analyze_topic.return_value = { "current_knowledge": "Summary", "findings": [], "iterations": 1, "questions": {}, "formatted_findings": "Findings", "all_links_of_system": [], } mock_init.return_value = mock_system # First call with one set of settings _ = quick_summary("Query 1", provider="openai", temperature=0.9) # Second call with different settings _ = quick_summary("Query 2", provider="anthropic", temperature=0.1) # Verify each call got its own settings call1_settings = mock_init.call_args_list[0][1]["settings_snapshot"] call2_settings = mock_init.call_args_list[1][1]["settings_snapshot"] assert call1_settings["llm.provider"]["value"] == "openai" assert call1_settings["llm.temperature"]["value"] == 0.9 assert call2_settings["llm.provider"]["value"] == "anthropic" assert call2_settings["llm.temperature"]["value"] == 0.1 class TestMultiProviderScenarios: """Test scenarios with multiple LLM providers.""" def test_multi_model_research(self): """Test research using multiple models for comparison.""" models = [ ("openai", "gpt-4", 0.3), ("anthropic", "claude-3-opus-20240229", 0.3), ("openai", "gpt-3.5-turbo", 0.5), ] results = [] with patch( "local_deep_research.api.research_functions._init_search_system" ) as mock_init: mock_system = MagicMock() mock_system.analyze_topic.return_value = { "current_knowledge": "Model-specific summary", "findings": [], "iterations": 1, "questions": {}, "formatted_findings": "Findings", "all_links_of_system": [], } mock_init.return_value = mock_system for provider, model, temp in models: result = quick_summary( "Compare quantum and classical computing", provider=provider, temperature=temp, settings_override={ f"llm.{provider}.model": model, }, ) results.append((provider, model, result)) # Verify each call used different settings assert mock_init.call_count == len(models) for i, (provider, model, _) in enumerate(models): call_settings = mock_init.call_args_list[i][1][ "settings_snapshot" ] assert call_settings["llm.provider"]["value"] == provider assert call_settings[f"llm.{provider}.model"]["value"] == model class TestSearchEngineIntegration: """Test integration with various search engines through settings.""" @patch("local_deep_research.api.research_functions.get_llm") @patch("local_deep_research.api.research_functions.get_search") def test_search_engine_specific_settings( self, mock_get_search, mock_get_llm ): """Test that search engine specific settings are applied. Each call's positional first argument identifies the chosen tool; engine-specific settings should appear in `settings_snapshot`. Patches target `api.research_functions` since that is where the names are bound. """ search_configs = [ { "engine": "searxng", "settings": { "search.tool": "searxng", "search.engines.searxng.base_url": "https://searx.example.com", "search.engines.searxng.timeout": 15, }, }, { "engine": "duckduckgo", "settings": { "search.tool": "duckduckgo", "search.engines.duckduckgo.region": "uk-en", "search.engines.duckduckgo.safe_search": False, }, }, { "engine": "wikipedia", "settings": { "search.tool": "wikipedia", "search.engines.wikipedia.language": "es", "search.engines.wikipedia.max_chars": 2000, }, }, ] mock_get_llm.return_value = MagicMock() mock_get_search.return_value = MagicMock() for config in search_configs: _ = quick_summary( f"Test with {config['engine']}", provider="openai", api_key="test-key", settings_override=config["settings"], ) # Verify the chosen search engine appears as the first positional # argument (the get_search public signature). last_args, last_kwargs = mock_get_search.call_args assert last_args[0] == config["engine"] # Engine-specific overrides must propagate through settings_snapshot. snapshot = last_kwargs["settings_snapshot"] for key, value in config["settings"].items(): assert key in snapshot, ( f"missing snapshot key {key} for {config['engine']}" ) assert snapshot[key]["value"] == value class TestPerformanceSettings: """Test performance-related settings.""" def test_concurrent_research_settings(self): """Test settings for concurrent research operations.""" # Settings optimized for concurrent operations concurrent_settings = create_settings_snapshot( overrides={ "research.concurrent.enabled": True, "research.concurrent.max_workers": 5, "llm.request_timeout": 30, "llm.max_retries": 2, "search.request_timeout": 10, "search.concurrent_searches": 3, } ) # Verify concurrent settings assert ( concurrent_settings["research.concurrent.enabled"]["value"] is True ) assert ( concurrent_settings["research.concurrent.max_workers"]["value"] == 5 ) assert concurrent_settings["llm.request_timeout"]["value"] == 30 def test_rate_limiting_settings(self): """Test rate limiting settings for API calls.""" rate_limit_settings = create_settings_snapshot( overrides={ "llm.rate_limit.enabled": True, "llm.rate_limit.requests_per_minute": 60, "llm.rate_limit.tokens_per_minute": 90000, "llm.rate_limit.concurrent_requests": 5, "search.rate_limit.requests_per_second": 10, } ) # Verify rate limiting configuration assert rate_limit_settings["llm.rate_limit.enabled"]["value"] is True assert ( rate_limit_settings["llm.rate_limit.requests_per_minute"]["value"] == 60 ) assert ( rate_limit_settings["search.rate_limit.requests_per_second"][ "value" ] == 10 ) class TestDebugAndMonitoringSettings: """Test debug and monitoring related settings.""" def test_debug_mode_settings(self): """Test settings for debug mode.""" debug_settings = create_settings_snapshot( overrides={ "debug.enabled": True, "debug.log_level": "DEBUG", "debug.log_api_calls": True, "debug.log_search_queries": True, "debug.save_intermediate_results": True, "debug.pretty_print_responses": True, } ) # Verify debug settings assert debug_settings["debug.enabled"]["value"] is True assert debug_settings["debug.log_level"]["value"] == "DEBUG" assert debug_settings["debug.log_api_calls"]["value"] is True def test_monitoring_settings(self): """Test settings for monitoring and metrics.""" monitoring_settings = create_settings_snapshot( overrides={ "monitoring.enabled": True, "monitoring.metrics.api_latency": True, "monitoring.metrics.token_usage": True, "monitoring.metrics.search_performance": True, "monitoring.export.format": "prometheus", "monitoring.export.endpoint": "http://metrics.example.com", } ) # Verify monitoring configuration assert monitoring_settings["monitoring.enabled"]["value"] is True assert ( monitoring_settings["monitoring.metrics.token_usage"]["value"] is True ) assert ( monitoring_settings["monitoring.export.format"]["value"] == "prometheus" )