""" Integration tests for LLM provider configuration and execution. Tests the full flow from API key configuration to research execution. """ import pytest from unittest.mock import Mock, patch from langchain_core.language_models import BaseChatModel from local_deep_research.settings import SettingsManager class TestLLMProviderIntegration: """Test complete LLM provider integration scenarios.""" @pytest.fixture def settings_dict(self): """Create a dictionary of settings.""" return { "llm.provider": "openai_endpoint", "llm.model": "claude-3-sonnet", "llm.temperature": 0.7, "llm.openai_endpoint.api_key": "sk-openrouter-test-key", "llm.openai_endpoint.url": "https://openrouter.ai/api/v1", "llm.openai_endpoint.model": "claude-3-sonnet", "llm.supports_max_tokens": True, "llm.max_tokens": 4096, "llm.context_window_unrestricted": True, "llm.openai.api_key": None, "llm.openai.api_base": None, "llm.openai.organization": None, "llm.anthropic.api_key": None, "llm.streaming": False, "llm.max_retries": None, "llm.request_timeout": None, "llm.ollama.url": "http://localhost:11434", "llm.lmstudio.url": "http://localhost:1234", "llm.llamacpp.url": "http://localhost:8080/v1", "app.lock_settings": False, "rate_limiting.llm_enabled": False, "search.tool": "searxng", "search.iterations": 5, "search.questions_per_iteration": 3, "search.max_results": 10, "search.enable_direct_summary": True, "search.enable_search_engine": True, "search.additional_results": 3, "search.timeout": 15, "search.region": "en-US", "search.proxy": None, "search.validate_sources": True, "search.enable_think_tags": False, "search.require_all_sources": False, "search.max_backoff_time": 300, "search.time_period": "all", "search.safe_search": True, "search.snippets_only": True, "search.search_language": "en", "search.max_filtered_results": 10, "search.smart_search.use_query_improvement": True, "search.smart_search.use_document_relevance": True, "search.smart_search.use_answer_extraction": True, "search.smart_search.use_semantic_cache": True, "search.smart_search.use_result_reranking": True, "search.smart_search.use_query_suggestion": True, "llm.local_context_window_size": 4096, "llm.context_window_size": 128000, } @pytest.fixture def mock_session(self, settings_dict): """Create a mock database session with settings.""" session = Mock() # Store settings that can be modified current_settings = settings_dict.copy() # Helper to get settings def get_setting_mock(key, default=None, check_env=True): return current_settings.get(key, default) # Helper to set settings def set_setting_mock(key, value, commit=True): current_settings[key] = value return True # Helper to get all settings def get_all_settings_mock(): # Return in the format expected by the API result = {} for key, value in current_settings.items(): result[key] = { "value": value, "type": "STRING", "name": key, "description": f"Setting for {key}", "category": key.split(".")[0], "ui_element": "text", "visible": True, "editable": True, } return result # Store these helpers on the session for patching session._get_setting = get_setting_mock session._set_setting = set_setting_mock session._get_all_settings = get_all_settings_mock session._current_settings = current_settings return session def test_research_with_configured_openrouter(self, mock_session): """Test running research with configured OpenRouter API.""" from local_deep_research.api.research_functions import quick_summary # Create settings manager with patched methods with patch.object( SettingsManager, "get_all_settings", side_effect=mock_session._get_all_settings, ): settings_manager = SettingsManager(None) settings_snapshot = settings_manager.get_all_settings() # The fixture sets search.tool="searxng", but searxng is only an # available engine once an instance URL is configured (it isn't # here) and the research path instantiates the engine. Point it at # the always-available "library" engine instead. "library" is # PRIVATE, so pin the egress scope to "unprotected" (the escape # hatch; "both" is retired per ADR-0007) — otherwise the default # adaptive scope resolves to PRIVATE_ONLY, forces local LLM, and # denies the OpenRouter provider this test configures. settings_snapshot["search.tool"]["value"] = "library" settings_snapshot["policy.egress_scope"] = {"value": "unprotected"} # Verify settings are correct - get_all_settings returns nested dicts assert ( settings_snapshot["llm.provider"]["value"] == "openai_endpoint" ) assert ( settings_snapshot["llm.openai_endpoint.api_key"]["value"] == "sk-openrouter-test-key" ) # Patch ChatOpenAI in both locations: the auto-discovered provider # path (openai_base) and the hardcoded fallback (llm_config). # Which path runs depends on whether other tests triggered # provider auto-registration in this process. with patch( "local_deep_research.llm.providers.openai_base.ChatOpenAI" ) as mock_openai: with ( patch( "local_deep_research.llm.providers.implementations.openai.ChatOpenAI", mock_openai, ), patch( "local_deep_research.api.research_functions.AdvancedSearchSystem" ) as mock_search_system, ): # Setup mock LLM (spec=BaseChatModel so isinstance check passes) mock_llm_instance = Mock(spec=BaseChatModel) mock_response = Mock() mock_response.content = "Based on my research: Test topic is important because..." mock_llm_instance.invoke.return_value = mock_response mock_openai.return_value = mock_llm_instance # Setup mock search system mock_system_instance = Mock() mock_system_instance.analyze_topic.return_value = { "current_knowledge": "Based on my research: Test topic is important because...", "sources": ["https://example.com/test-topic"], "all_links_of_system": [ "https://example.com/test-topic" ], "findings": [], "iterations": 2, "questions": {}, } mock_search_system.return_value = mock_system_instance # Run research result = quick_summary( query="Explain test topic", iterations=2, questions_per_iteration=3, settings_snapshot=settings_snapshot, ) # Verify OpenRouter was configured correctly assert mock_openai.called call_args = mock_openai.call_args assert ( call_args.kwargs["api_key"] == "sk-openrouter-test-key" ) # The URL kwarg name differs by code path: # registry path uses "base_url", direct path uses "openai_api_base" url_value = call_args.kwargs.get( "base_url", call_args.kwargs.get("openai_api_base") ) assert url_value == "https://openrouter.ai/api/v1" assert call_args.kwargs["model"] == "claude-3-sonnet" # Verify research completed assert "summary" in result assert "sources" in result def test_switching_providers_dynamically(self, mock_session): """Test switching between providers dynamically.""" from local_deep_research.config.llm_config import get_llm with patch.object( SettingsManager, "get_setting", side_effect=mock_session._get_setting, ): with patch.object( SettingsManager, "set_setting", side_effect=mock_session._set_setting, ): with patch.object( SettingsManager, "get_all_settings", side_effect=mock_session._get_all_settings, ): settings_manager = SettingsManager(None) # Test 1: OpenRouter configuration # Patch both the auto-discovered provider path and the # hardcoded fallback in llm_config (which path runs # depends on whether other tests registered providers). settings_manager.set_setting( "llm.provider", "openai_endpoint" ) settings_manager.set_setting( "llm.openai_endpoint.api_key", "sk-openrouter-key" ) settings_snapshot = settings_manager.get_all_settings() with patch( "local_deep_research.llm.providers.openai_base.ChatOpenAI" ) as mock_openai: with patch( "local_deep_research.llm.providers.implementations.openai.ChatOpenAI", mock_openai, ): mock_openai.return_value = Mock(spec=BaseChatModel) get_llm(settings_snapshot=settings_snapshot) assert ( mock_openai.call_args.kwargs["api_key"] == "sk-openrouter-key" ) # Test 2: Switch to OpenAI settings_manager.set_setting("llm.provider", "openai") settings_manager.set_setting( "llm.openai.api_key", "sk-openai-key" ) settings_snapshot = settings_manager.get_all_settings() with patch( "local_deep_research.llm.providers.implementations.openai.ChatOpenAI" ) as mock_openai: with patch( "local_deep_research.llm.providers.implementations.openai.ChatOpenAI", mock_openai, ): mock_openai.return_value = Mock(spec=BaseChatModel) get_llm(settings_snapshot=settings_snapshot) assert ( mock_openai.call_args.kwargs["api_key"] == "sk-openai-key" ) # Test 3: Switch to Anthropic settings_manager.set_setting("llm.provider", "anthropic") settings_manager.set_setting( "llm.anthropic.api_key", "sk-anthropic-key" ) settings_snapshot = settings_manager.get_all_settings() with patch( "local_deep_research.llm.providers.implementations.anthropic.ChatAnthropic" ) as mock_anthropic: with patch( "local_deep_research.llm.providers.implementations.anthropic.ChatAnthropic", mock_anthropic, ): mock_anthropic.return_value = Mock( spec=BaseChatModel ) get_llm(settings_snapshot=settings_snapshot) assert ( mock_anthropic.call_args.kwargs[ "anthropic_api_key" ] == "sk-anthropic-key" ) def test_api_key_validation_before_research(self, mock_session): """Test that API key presence is validated before starting research.""" from local_deep_research.llm.providers.implementations.custom_openai_endpoint import ( CustomOpenAIEndpointProvider, ) with patch.object( SettingsManager, "set_setting", side_effect=mock_session._set_setting, ): settings_manager = SettingsManager(None) # With API key settings_manager.set_setting( "llm.openai_endpoint.api_key", "sk-test-key" ) settings_with_key = { "llm.openai_endpoint.api_key": { "value": "sk-test-key", "type": "str", } } assert ( CustomOpenAIEndpointProvider.is_available( settings_snapshot=settings_with_key ) is True ) # Without API key settings_manager.set_setting("llm.openai_endpoint.api_key", None) settings_without_key = { "llm.openai_endpoint.api_key": {"value": None, "type": "str"} } assert ( CustomOpenAIEndpointProvider.is_available( settings_snapshot=settings_without_key ) is False ) @pytest.mark.requires_llm def test_custom_model_configuration(self, mock_session): """Test configuring custom models with API endpoints.""" with patch.object( SettingsManager, "get_setting", side_effect=mock_session._get_setting, ): with patch.object( SettingsManager, "set_setting", side_effect=mock_session._set_setting, ): with patch.object( SettingsManager, "get_all_settings", side_effect=mock_session._get_all_settings, ): settings_manager = SettingsManager(None) # Configure for a specific model on OpenRouter settings_manager.set_setting( "llm.provider", "openai_endpoint" ) settings_manager.set_setting( "llm.model", "anthropic/claude-3-opus" ) settings_manager.set_setting( "llm.openai_endpoint.api_key", "sk-openrouter-key" ) settings_manager.set_setting( "llm.openai_endpoint.url", "https://openrouter.ai/api/v1", ) settings_manager.set_setting("llm.temperature", 0.3) settings_snapshot = settings_manager.get_all_settings() with patch( "local_deep_research.llm.providers.implementations.openai.ChatOpenAI" ) as mock_openai: mock_openai.return_value = Mock(spec=BaseChatModel) from local_deep_research.config.llm_config import ( get_llm, ) get_llm(settings_snapshot=settings_snapshot) # Verify model configuration call_args = mock_openai.call_args assert ( call_args.kwargs["model"] == "anthropic/claude-3-opus" ) assert call_args.kwargs["temperature"] == 0.3 assert ( call_args.kwargs["api_key"] == "sk-openrouter-key" ) assert ( call_args.kwargs["openai_api_base"] == "https://openrouter.ai/api/v1" ) @pytest.mark.integration @pytest.mark.skip( reason=( "BenchmarkService.create_benchmark_run inserts into the encrypted " "user database, which the unit-test fixture does not stand up. " "Run as part of the integration suite with a real DB session." ) ) def test_benchmark_with_api_configured_llm(self, mock_session): """Test running benchmarks with API-configured LLM.""" # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: keep skip (justified) or move to integration). from local_deep_research.benchmarks.web_api.benchmark_service import ( BenchmarkService, ) with patch.object( SettingsManager, "set_setting", side_effect=mock_session._set_setting, ): with patch.object( SettingsManager, "get_all_settings", side_effect=mock_session._get_all_settings, ): settings_manager = SettingsManager(None) settings_manager.set_setting("llm.provider", "openai_endpoint") settings_manager.set_setting( "llm.openai_endpoint.api_key", "sk-benchmark-key" ) benchmark_service = BenchmarkService() # Create benchmark configuration search_config = { "iterations": 2, "questions_per_iteration": 3, "search_tool": "searxng", "provider": "openai_endpoint", "model_name": "claude-3-sonnet", } evaluation_config = { "provider": "openai_endpoint", "model_name": "claude-3-sonnet", } datasets_config = {"simpleqa": {"count": 5}} # Test benchmark creation with patch.object( settings_manager, "get_all_settings" ) as mock_get_settings: mock_get_settings.return_value = ( settings_manager.get_all_settings() ) benchmark_id = benchmark_service.create_benchmark_run( run_name="Test API Key Benchmark", search_config=search_config, evaluation_config=evaluation_config, datasets_config=datasets_config, username="testuser", ) assert isinstance(benchmark_id, int) if __name__ == "__main__": pytest.main([__file__, "-v"])