""" Enhanced search engine tests using scottvr's patterns. This demonstrates how to use the new mock utilities and fixtures. """ import os from unittest.mock import MagicMock import pytest from tests.test_utils import ( add_src_to_path, assert_search_result_format, mock_api_response, ) # Add src to path add_src_to_path() import local_deep_research.metrics.search_tracker as search_tracker_module # noqa: E402 # Mock the search tracker for all tests in this module mock_tracker = MagicMock() mock_tracker.record_search = MagicMock() search_tracker_module.get_search_tracker = MagicMock(return_value=mock_tracker) class TestWikipediaSearchEnhanced: """Enhanced Wikipedia search tests using new patterns.""" def test_wikipedia_search_success( self, monkeypatch, mock_wikipedia_response ): """Test successful Wikipedia search.""" # Mock the wikipedia library functions directly instead of requests.get mock_search_results = ["Artificial intelligence", "Machine learning"] monkeypatch.setattr( "wikipedia.search", lambda query, results=10: mock_search_results ) def mock_summary(title, sentences=3, auto_suggest=True): if title == "Artificial intelligence": return "Artificial intelligence (AI) is intelligence demonstrated by machines, in contrast to natural intelligence." if title == "Machine learning": return "Machine learning (ML) is a subset of artificial intelligence that focuses on algorithms." return "Generic summary" monkeypatch.setattr("wikipedia.summary", mock_summary) # Import and test from local_deep_research.web_search_engines.engines.search_engine_wikipedia import ( WikipediaSearchEngine, ) search = WikipediaSearchEngine(max_results=5) results = search.run("artificial intelligence") # Verify results assert len(results) == 2 for result in results: assert_search_result_format(result) # Check specific content assert results[0]["title"] == "Artificial intelligence" assert "intelligence demonstrated by machines" in results[0]["snippet"] assert results[0]["source"] == "Wikipedia" def test_wikipedia_search_error_handling(self, monkeypatch): """Test Wikipedia search error handling.""" # Mock wikipedia.search to raise an exception def mock_search_error(*args, **kwargs): raise Exception("Search failed") monkeypatch.setattr("wikipedia.search", mock_search_error) from local_deep_research.web_search_engines.engines.search_engine_wikipedia import ( WikipediaSearchEngine, ) search = WikipediaSearchEngine() results = search.run("test query") # Should return empty list on error assert isinstance(results, list) assert len(results) == 0 def test_wikipedia_search_network_error(self, monkeypatch): """Test Wikipedia search with network errors.""" # Mock a network exception on wikipedia.search def mock_network_error(*args, **kwargs): raise ConnectionError("Network error") monkeypatch.setattr("wikipedia.search", mock_network_error) monkeypatch.setattr("wikipedia.set_lang", lambda x: None) from local_deep_research.web_search_engines.engines.search_engine_wikipedia import ( WikipediaSearchEngine, ) search = WikipediaSearchEngine() results = search.run("test query") # Should handle exception gracefully assert isinstance(results, list) assert len(results) == 0 class TestArxivSearchEnhanced: """Enhanced arXiv search tests.""" def test_arxiv_search_success(self, monkeypatch, mock_arxiv_response): """Test successful arXiv search.""" # Mock search_config to ensure searxng is available - patch multiple locations def mock_search_config(): return { "searxng": { "module_path": ".engines.search_engine_searxng", "class_name": "SearXNGSearchEngine", "default_params": {"host_url": "http://localhost:8080"}, } } # Patch multiple potential import locations monkeypatch.setattr( "local_deep_research.web_search_engines.search_engines_config.search_config", mock_search_config, ) monkeypatch.setattr( "local_deep_research.web_search_engines.search_engines_config.search_config", mock_search_config, ) # Also mock the search engine factory create function to avoid the KeyError entirely def mock_create_search_engine(engine_name, **kwargs): from unittest.mock import Mock mock_engine = Mock() mock_engine.is_available = True mock_engine.run.return_value = [] return mock_engine monkeypatch.setattr( "local_deep_research.web_search_engines.search_engine_factory.create_search_engine", mock_create_search_engine, ) from local_deep_research.web_search_engines.engines.search_engine_arxiv import ( ArXivSearchEngine, ) # Mock the _get_search_results method to return empty list to avoid actual arXiv API calls def mock_get_search_results(self, query): from unittest.mock import Mock # Create a mock paper object mock_paper = Mock() mock_paper.entry_id = "2301.12345" mock_paper.title = "Test Machine Learning Paper" mock_paper.summary = ( "This is a test abstract about machine learning." ) mock_paper.authors = [ Mock(name="John Doe"), Mock(name="Jane Smith"), ] mock_paper.published = None mock_paper.journal_ref = None return [mock_paper] monkeypatch.setattr( ArXivSearchEngine, "_get_search_results", mock_get_search_results ) # Mock JournalReputationFilter.create_default to return None # This avoids the need for LLM initialization monkeypatch.setattr( "local_deep_research.advanced_search_system.filters.journal_reputation_filter.JournalReputationFilter.create_default", lambda *args, **kwargs: None, ) search = ArXivSearchEngine(max_results=5) results = search.run("machine learning") # Verify results assert len(results) >= 1 if results: # ArxivSearch might return empty if XML parsing fails for result in results: assert_search_result_format(result) assert results[0]["source"] == "arXiv" class TestSearchEngineFactory: """Test search engine factory with mocked configs.""" @pytest.mark.skipif( os.getenv("CI") == "true", reason="Skipping in CI due to timeout issues" ) def test_factory_with_mocked_llm(self, monkeypatch): """Test search engine factory with mocked LLM config.""" # Import the mock utilities from tests.mock_modules import ( create_mock_db_utils, create_mock_llm_config, ) # Create mock LLM config create_mock_llm_config(monkeypatch) # Mock database utilities with search engine configurations search_engine_config = { "search.engine.web": { "wikipedia": { "module_path": "local_deep_research.web_search_engines.engines.search_engine_wikipedia", "class_name": "WikipediaSearchEngine", "requires_api_key": False, "requires_llm": False, "default_params": {"max_results": 10}, } }, "search.engine.DEFAULT_SEARCH_ENGINE": "wikipedia", "search.max_results": 10, "search.engine.local": {}, } create_mock_db_utils(monkeypatch, search_engine_config) # Mock db_utils to avoid database access from tests.mock_modules import create_mock_db_utils create_mock_db_utils(monkeypatch) # Mock search_engines_config to avoid circular imports def mock_search_config(): return { "wikipedia": { "module_path": ".engines.search_engine_wikipedia", "class_name": "WikipediaSearchEngine", "default_params": {"max_results": 10}, "requires_api_key": False, "requires_llm": False, } } def mock_default_search_engine(): return "wikipedia" monkeypatch.setattr( "local_deep_research.web_search_engines.search_engines_config.search_config", mock_search_config, ) monkeypatch.setattr( "local_deep_research.web_search_engines.search_engines_config.default_search_engine", mock_default_search_engine, ) # Mock wikipedia library monkeypatch.setattr("wikipedia.set_lang", lambda *args, **kwargs: None) monkeypatch.setattr( "wikipedia.search", lambda query, results: ["Test Result"] ) monkeypatch.setattr( "wikipedia.summary", lambda title, sentences=5, auto_suggest=False: "Test summary", ) # Test factory from local_deep_research.web_search_engines.search_engine_factory import ( create_search_engine, ) # Create a minimal settings snapshot for the factory settings_snapshot = { "search.engine.web.wikipedia.module_path": { "value": "local_deep_research.web_search_engines.engines" ".search_engine_wikipedia", "ui_element": "text", }, "search.engine.web.wikipedia.class_name": { "value": "WikipediaSearchEngine", "ui_element": "text", }, "search.engine.web.wikipedia.requires_api_key": { "value": False, "ui_element": "checkbox", }, "search.engine.web.wikipedia.requires_llm": { "value": False, "ui_element": "checkbox", }, "search.engine.web.wikipedia.default_params.max_results": { "value": 10, "ui_element": "number", }, } # Should work even with mocked modules engine = create_search_engine( "wikipedia", settings_snapshot=settings_snapshot ) assert engine is not None # Test search results = engine.run("test") assert isinstance(results, list) class TestMultipleSearchEngines: """Test multiple search engines with shared fixtures.""" @pytest.mark.parametrize( "engine_name,response_fixture", [ ("wikipedia", "mock_wikipedia_response"), ("google_pse", "mock_google_pse_response"), ("semantic_scholar", "mock_semantic_scholar_response"), ], ) def test_search_engines( self, engine_name, response_fixture, request, monkeypatch ): """Test multiple search engines with parametrized fixtures.""" # Get the fixture value dynamically mock_response_data = request.getfixturevalue(response_fixture) # Mock the API response based on engine type if engine_name == "wikipedia": # Mock wikipedia library functions directly mock_search_results = [ "Artificial intelligence", "Machine learning", ] monkeypatch.setattr( "wikipedia.search", lambda query, results=10: mock_search_results, ) def mock_summary(title, sentences=3, auto_suggest=True): if title == "Artificial intelligence": return "Artificial intelligence (AI) is intelligence demonstrated by machines." if title == "Machine learning": return "Machine learning (ML) is a subset of artificial intelligence." return "Generic summary" monkeypatch.setattr("wikipedia.summary", mock_summary) else: # For other engines, use requests.get mocking if engine_name == "google_pse" or engine_name == "semantic_scholar": mock_response = mock_api_response( 200, json_data=mock_response_data ) monkeypatch.setattr( "requests.get", lambda *args, **kwargs: mock_response ) # Import the appropriate search engine if engine_name == "wikipedia": from local_deep_research.web_search_engines.engines.search_engine_wikipedia import ( WikipediaSearchEngine as SearchEngine, ) elif engine_name == "google_pse": # Mock settings snapshot for Google PSE def mock_get_setting_from_snapshot( key, default=None, settings_snapshot=None ): if "api_key" in key: return "test_api_key" if "engine_id" in key: return "test_engine_id" return default monkeypatch.setattr( "local_deep_research.config.thread_settings.get_setting_from_snapshot", mock_get_setting_from_snapshot, ) # Also set environment variables as fallback monkeypatch.setenv("GOOGLE_PSE_API_KEY", "test_api_key") monkeypatch.setenv("GOOGLE_PSE_ENGINE_ID", "test_engine_id") from local_deep_research.web_search_engines.engines.search_engine_google_pse import ( GooglePSESearchEngine as SearchEngine, ) elif engine_name == "semantic_scholar": from local_deep_research.web_search_engines.engines.search_engine_semantic_scholar import ( SemanticScholarSearchEngine as SearchEngine, ) # Test the search search = SearchEngine() results = search.run("test query") # Basic validation assert isinstance(results, list) if results: # Some engines might return empty on mock data for result in results: assert_search_result_format(result)