""" Extended tests for utilities/search_utilities.py Tests cover edge cases and scenarios not covered in the base test file: - Phase parsing for Follow-up and Sub-query formats - Invalid phase format handling - Source aggregation behavior - Edge cases in format_findings """ class TestFormatFindingsPhaseParsingFollowUp: """Tests for Follow-up Iteration phase parsing in format_findings.""" def test_followup_iteration_format_basic(self): """Test Follow-up Iteration X.Y format is parsed correctly.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Follow-up Iteration 1.1", "content": "First follow-up content", "search_results": [], } ] questions = {1: ["First question", "Second question"]} result = format_findings(findings, "Summary", questions) # The question should be displayed assert "First question" in result assert "First follow-up content" in result def test_followup_iteration_second_question(self): """Test Follow-up Iteration X.2 shows second question.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Follow-up Iteration 1.2", "content": "Second follow-up", "search_results": [], } ] questions = {1: ["First question", "Second question", "Third question"]} result = format_findings(findings, "Summary", questions) assert "Second question" in result def test_followup_iteration_multiple_iterations(self): """Test Follow-up Iteration across multiple iterations.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Follow-up Iteration 1.1", "content": "Iter 1 Q1", "search_results": [], }, { "phase": "Follow-up Iteration 2.1", "content": "Iter 2 Q1", "search_results": [], }, ] questions = { 1: ["Iteration 1 Question 1"], 2: ["Iteration 2 Question 1"], } result = format_findings(findings, "Summary", questions) assert "Iteration 1 Question 1" in result assert "Iteration 2 Question 1" in result def test_followup_iteration_missing_question_index(self): """Test Follow-up Iteration with question index out of bounds.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Follow-up Iteration 1.5", # Index 5 doesn't exist "content": "Out of bounds", "search_results": [], } ] questions = {1: ["Only one question"]} # Should not raise, just skip showing question result = format_findings(findings, "Summary", questions) assert "Out of bounds" in result # Should not crash def test_followup_iteration_missing_iteration(self): """Test Follow-up Iteration with iteration not in questions dict.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Follow-up Iteration 3.1", # Iteration 3 doesn't exist "content": "No matching iteration", "search_results": [], } ] questions = {1: ["Question 1"]} result = format_findings(findings, "Summary", questions) assert "No matching iteration" in result class TestFormatFindingsPhaseParsingSubQuery: """Tests for Sub-query phase parsing in format_findings.""" def test_subquery_format_basic(self): """Test Sub-query X format is parsed correctly.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Sub-query 1", "content": "Sub-query content", "search_results": [], } ] # IterDRAG stores sub-queries in iteration 0 questions = {0: ["First sub-query", "Second sub-query"]} result = format_findings(findings, "Summary", questions) assert "First sub-query" in result assert "Sub-query content" in result def test_subquery_second_question(self): """Test Sub-query 2 shows second question.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Sub-query 2", "content": "Second sub-query content", "search_results": [], } ] questions = {0: ["First", "Second", "Third"]} result = format_findings(findings, "Summary", questions) assert "Second" in result def test_subquery_out_of_bounds(self): """Test Sub-query with index out of bounds.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Sub-query 10", # Index 10 doesn't exist "content": "Out of bounds sub-query", "search_results": [], } ] questions = {0: ["Only one"]} result = format_findings(findings, "Summary", questions) assert "Out of bounds sub-query" in result def test_subquery_no_iteration_zero(self): """Test Sub-query when iteration 0 doesn't exist.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Sub-query 1", "content": "Sub-query content", "search_results": [], } ] questions = {1: ["Not iteration zero"]} result = format_findings(findings, "Summary", questions) assert "Sub-query content" in result class TestFormatFindingsInvalidPhaseFormat: """Tests for invalid phase format handling in format_findings.""" def test_invalid_followup_format_non_numeric(self): """Test Follow-up Iteration with non-numeric parts.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Follow-up Iteration abc.def", "content": "Invalid format content", "search_results": [], } ] questions = {1: ["Question 1"]} result = format_findings(findings, "Summary", questions) assert "Invalid format content" in result def test_invalid_followup_format_missing_dot(self): """Test Follow-up Iteration without dot separator.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Follow-up Iteration 1", # Missing .X "content": "Missing dot content", "search_results": [], } ] questions = {1: ["Question 1"]} result = format_findings(findings, "Summary", questions) assert "Missing dot content" in result def test_invalid_subquery_format_non_numeric(self): """Test Sub-query with non-numeric index.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Sub-query abc", "content": "Invalid sub-query content", "search_results": [], } ] questions = {0: ["Question 1"]} result = format_findings(findings, "Summary", questions) assert "Invalid sub-query content" in result def test_phase_with_special_characters(self): """Test phase with special characters.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Phase: Special ", "content": "Special content", "search_results": [], } ] result = format_findings(findings, "Summary", {}) assert "Special content" in result assert "Phase: Special " in result def test_phase_none_value(self): """Test finding with None phase.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": None, "content": "Content with None phase", "search_results": [], } ] result = format_findings(findings, "Summary", {}) # Should use "Unknown Phase" default assert "Content with None phase" in result class TestFormatFindingsSourceAggregation: """Tests for source aggregation in format_findings.""" def test_aggregates_sources_from_multiple_findings(self): """Test sources are aggregated from multiple findings.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Search 1", "content": "Content 1", "search_results": [ {"title": "Source A", "link": "https://a.com", "index": "1"} ], }, { "phase": "Search 2", "content": "Content 2", "search_results": [ {"title": "Source B", "link": "https://b.com", "index": "2"} ], }, ] result = format_findings(findings, "Summary", {}) assert "ALL SOURCES" in result assert "https://a.com" in result assert "https://b.com" in result def test_deduplicates_sources(self): """Test duplicate sources are deduplicated.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Search 1", "content": "Content 1", "search_results": [ {"title": "Same", "link": "https://same.com", "index": "1"} ], }, { "phase": "Search 2", "content": "Content 2", "search_results": [ {"title": "Same", "link": "https://same.com", "index": "2"} ], }, ] result = format_findings(findings, "Summary", {}) # URL should only appear once in the ALL SOURCES section all_sources_section = ( result.split("ALL SOURCES")[1] if "ALL SOURCES" in result else "" ) assert all_sources_section.count("https://same.com") == 1 def test_handles_finding_without_search_results(self): """Test findings without search_results key.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Search", "content": "Content without search results", # No search_results key } ] result = format_findings(findings, "Summary", {}) assert "Content without search results" in result def test_handles_empty_search_results(self): """Test findings with empty search_results list.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Search", "content": "Content with empty results", "search_results": [], } ] result = format_findings(findings, "Summary", {}) assert "Content with empty results" in result class TestFormatFindingsQuestionInFinding: """Tests for question field in finding itself.""" def test_displays_question_from_finding(self): """Test question from finding is displayed if not from phase.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Custom Phase", "content": "Content here", "question": "What is the meaning of life?", "search_results": [], } ] result = format_findings(findings, "Summary", {}) assert "What is the meaning of life?" in result assert "SEARCH QUESTION" in result def test_phase_question_overrides_finding_question(self): """Test question from phase takes precedence over finding question.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Follow-up Iteration 1.1", "content": "Content", "question": "Question from finding", "search_results": [], } ] questions = {1: ["Question from iteration"]} result = format_findings(findings, "Summary", questions) # Should show iteration question, not finding question assert "Question from iteration" in result def test_empty_question_field_ignored(self): """Test empty question field is ignored.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "Search", "content": "Content", "question": "", "search_results": [], } ] result = format_findings(findings, "Summary", {}) assert "SEARCH QUESTION" not in result class TestFormatFindingsEdgeCases: """Tests for edge cases in format_findings.""" def test_empty_synthesized_content(self): """Test with empty synthesized content.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) result = format_findings([], "", {}) # Should not crash with empty content assert result is not None def test_synthesized_content_with_newlines(self): """Test synthesized content with newlines is preserved.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) content = "Line 1\n\nLine 2\n\nLine 3" result = format_findings([], content, {}) assert "Line 1" in result assert "Line 2" in result assert "Line 3" in result def test_large_number_of_findings(self): """Test with large number of findings.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": f"Phase {i}", "content": f"Content {i}", "search_results": [], } for i in range(100) ] result = format_findings(findings, "Summary", {}) assert "Phase 0" in result assert "Phase 99" in result assert "Content 0" in result assert "Content 99" in result def test_findings_with_all_none_values(self): """Test findings with all None values use defaults.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": None, "content": None, "search_results": None, } ] result = format_findings(findings, "Summary", {}) assert "Unknown Phase" in result or result is not None def test_unicode_content_handling(self): """Test unicode content is handled correctly.""" from local_deep_research.utilities.search_utilities import ( format_findings, ) findings = [ { "phase": "搜索结果", "content": "日本語テキスト с русским 🎉", "search_results": [ { "title": "日本語", "link": "https://example.com", "index": "1", } ], } ] result = format_findings(findings, "概要 Summary", {}) assert "日本語テキスト" in result assert "🎉" in result class TestExtractLinksEdgeCases: """Additional edge case tests for extract_links_from_search_results.""" def test_handles_integer_index(self): """Test handles integer index instead of string.""" from local_deep_research.utilities.search_utilities import ( extract_links_from_search_results, ) results = [ { "title": "Test", "link": "https://example.com", "index": 1, # Integer instead of string } ] # This might fail if strip() is called on integer # The function should handle this gracefully try: links = extract_links_from_search_results(results) # If it succeeds, check the result assert len(links) >= 0 except Exception: # If it fails, that's also acceptable behavior pass def test_handles_mixed_key_formats(self): """Test handles results with different key formats.""" from local_deep_research.utilities.search_utilities import ( extract_links_from_search_results, ) results = [ {"title": "Normal", "link": "https://normal.com", "index": "1"}, { "title": " Spaces ", "link": " https://spaces.com ", "index": "2", }, ] links = extract_links_from_search_results(results) assert len(links) == 2 assert links[1]["title"] == "Spaces" assert links[1]["url"] == "https://spaces.com" class TestFormatLinksEdgeCases: """Additional edge case tests for format_links_to_markdown.""" def test_handles_untitled_default(self): """Test handles links without title using default.""" from local_deep_research.utilities.search_utilities import ( format_links_to_markdown, ) links = [ {"url": "https://example.com", "index": "1"} # No title ] result = format_links_to_markdown(links) assert "Untitled" in result or "https://example.com" in result def test_multiple_indices_same_url(self): """Test multiple indices for same URL are aggregated.""" from local_deep_research.utilities.search_utilities import ( format_links_to_markdown, ) links = [ {"title": "Same", "url": "https://same.com", "index": "1"}, {"title": "Same", "url": "https://same.com", "index": "3"}, {"title": "Same", "url": "https://same.com", "index": "5"}, ] result = format_links_to_markdown(links) # URL should appear once with aggregated indices assert result.count("https://same.com") == 1 # Should show multiple indices assert "1" in result assert "3" in result assert "5" in result