fix: Mock html2text in content truncation test

CLAUDE.md

@@ -241,6 +241,31 @@ def test_live_search():  # Missing @pytest.mark.integration
 - **Type hints**: Required for all function signatures (enforced by mypy)
 - **Docstrings**: Google-style docstrings for public functions
 
+### Architecture Standards
+
+**Maximum Nesting Depth**: 3 levels or less
+- Functions with deeper nesting must be refactored
+- Extract helper methods to reduce complexity
+- Each helper method should represent a single concept
+- Example violation: 6-level nesting in content_scraper.py (fixed by extracting 10 helper methods)
+
+**One Class Per File**:
+- Each file should contain exactly one class definition
+- Applies to both production code and test infrastructure
+- Example: Split `mock_ddgs.py` (3 classes) into 3 separate files
+
+**Methods for Concepts**:
+- Extract helper methods for each logical concept or operation
+- Helper methods should have clear, single responsibilities
+- Prefer multiple small methods over one large method
+- Example: `_fetch_content()`, `_convert_to_markdown()`, `_truncate_if_needed()`
+
+**No Raw Mocks**:
+- Never use `MagicMock()` with property assignment (e.g., `mock.status_code = 200`)
+- Create typed mock classes instead (e.g., `MockHttpResponse`, `MockSerperResponse`)
+- Use factory pattern for creating pre-configured test doubles
+- Use builder pattern with fluent API for test data (e.g., `a_search_result().with_title("X").build()`)
+
 ### Single Source of Truth
 
 **Tool Versions** are defined in `pyproject.toml` under `[project.optional-dependencies.dev]`:

docker/api_tools.py

@@ -251,79 +251,146 @@ async def get_server_info_tool() -> dict:
             return response.model_dump()
 
 
+async def _fetch_with_serper(query: str, api_key: str):
+    """Fetch search results using Serper API.
+
+    Args:
+        query: Search query
+        api_key: Serper API key
+
+    Returns:
+        Tuple of (results, search_source)
+    """
+    import asyncio
+
+    from mcp_server import fetch_search_results
+
+    logger.info("Using Serper API for search")
+    results = await asyncio.get_event_loop().run_in_executor(
+        None, fetch_search_results, query, api_key
+    )
+    return results, "Serper API"
+
+
+async def _fetch_with_duckduckgo(query: str, has_api_key: bool):
+    """Fetch search results using DuckDuckGo.
+
+    Args:
+        query: Search query
+        has_api_key: Whether Serper API key was configured
+
+    Returns:
+        Tuple of (results, search_source)
+    """
+    import asyncio
+
+    from mcp_server import fetch_duckduckgo_search_results
+
+    if not has_api_key:
+        logger.info("No Serper API key configured, using DuckDuckGo fallback")
+    else:
+        logger.warning("No results from Serper API, trying DuckDuckGo fallback")
+
+    results = await asyncio.get_event_loop().run_in_executor(
+        None, fetch_duckduckgo_search_results, query
+    )
+    return results, "DuckDuckGo (free fallback)"
+
+
+def _format_no_results_error(query: str, search_source: str) -> dict:
+    """Format error response for no results.
+
+    Args:
+        query: Search query
+        search_source: Source that was used
+
+    Returns:
+        Error dictionary
+    """
+    return {
+        "error": "No search results found from any source.",
+        "query": query,
+        "search_source": search_source,
+    }
+
+
+def _format_search_error(error: str, query: str, search_source: str) -> dict:
+    """Format error response for search failure.
+
+    Args:
+        error: Error message
+        query: Search query
+        search_source: Source that was attempted
+
+    Returns:
+        Error dictionary
+    """
+    return {
+        "error": f"Search failed: {error}",
+        "query": query,
+        "search_source": search_source,
+    }
+
+
+async def _process_and_format_results(results, query: str, search_source: str):
+    """Process search results and format response.
+
+    Args:
+        results: Raw search results
+        query: Search query
+        search_source: Source of results
+
+    Returns:
+        Formatted results dictionary
+    """
+    import asyncio
+
+    from mcp_server import process_search_results
+
+    processed_results = await asyncio.get_event_loop().run_in_executor(
+        None, process_search_results, results
+    )
+
+    return {
+        "query": query,
+        "search_source": search_source,
+        "results": [result.model_dump() for result in processed_results],
+    }
+
+
 def create_webcat_functions() -> Dict[str, Any]:
     """Create a dictionary of WebCat functions for the tools to use."""
 
     # Import the existing functions from the main server
-    from mcp_server import (
-        SERPER_API_KEY,
-        fetch_duckduckgo_search_results,
-        fetch_search_results,
-        process_search_results,
-    )
+    from mcp_server import SERPER_API_KEY
 
     async def search_function(query: str) -> Dict[str, Any]:
         """Wrapper for the search functionality."""
-        import asyncio
-
         results = []
         search_source = "Unknown"
 
         try:
             # Try Serper API first if key is available
             if SERPER_API_KEY:
-                logger.info("Using Serper API for search")
-                search_source = "Serper API"
-                # Run the synchronous function in a thread pool
-                results = await asyncio.get_event_loop().run_in_executor(
-                    None, fetch_search_results, query, SERPER_API_KEY
-                )
+                results, search_source = await _fetch_with_serper(query, SERPER_API_KEY)
 
             # Fall back to DuckDuckGo if no API key or no results from Serper
             if not results:
-                if not SERPER_API_KEY:
-                    logger.info(
-                        "No Serper API key configured, using DuckDuckGo fallback"
-                    )
-                else:
-                    logger.warning(
-                        "No results from Serper API, trying DuckDuckGo fallback"
-                    )
-
-                search_source = "DuckDuckGo (free fallback)"
-                # Run the synchronous function in a thread pool
-                results = await asyncio.get_event_loop().run_in_executor(
-                    None, fetch_duckduckgo_search_results, query
+                results, search_source = await _fetch_with_duckduckgo(
+                    query, bool(SERPER_API_KEY)
                 )
 
             # Check if we got any results
             if not results:
                 logger.warning(f"No search results found for query: {query}")
-                return {
-                    "error": "No search results found from any source.",
-                    "query": query,
-                    "search_source": search_source,
-                }
-
-            # Process the results in thread pool (since it involves web scraping)
-            processed_results = await asyncio.get_event_loop().run_in_executor(
-                None, process_search_results, results
-            )
+                return _format_no_results_error(query, search_source)
 
-            # Return formatted results
-            return {
-                "query": query,
-                "search_source": search_source,
-                "results": [result.model_dump() for result in processed_results],
-            }
+            # Process and format results
+            return await _process_and_format_results(results, query, search_source)
 
         except Exception as e:
             logger.error(f"Error in search function: {str(e)}")
-            return {
-                "error": f"Search failed: {str(e)}",
-                "query": query,
-                "search_source": search_source,
-            }
+            return _format_search_error(str(e), query, search_source)
 
     async def health_check_function() -> Dict[str, Any]:
         """Wrapper for the health check functionality."""

docker/clients/duckduckgo_client.py

@@ -22,6 +22,22 @@
     )
 
 
+def _convert_ddg_result(result: dict) -> APISearchResult:
+    """Convert DuckDuckGo result format to APISearchResult.
+
+    Args:
+        result: Raw result dictionary from DuckDuckGo
+
+    Returns:
+        APISearchResult object
+    """
+    return APISearchResult(
+        title=result.get("title", "Untitled"),
+        link=result.get("href", ""),
+        snippet=result.get("body", ""),
+    )
+
+
 def fetch_duckduckgo_search_results(
     query: str, max_results: int = 3
 ) -> List[APISearchResult]:
@@ -43,20 +59,8 @@ def fetch_duckduckgo_search_results(
         logger.info(f"Using DuckDuckGo fallback search for: {query}")
 
         with DDGS() as ddgs:
-            # Get search results from DuckDuckGo
-            results = []
             search_results = ddgs.text(query, max_results=max_results)
-
-            for result in search_results:
-                # Convert DuckDuckGo result format to APISearchResult
-                results.append(
-                    APISearchResult(
-                        title=result.get("title", "Untitled"),
-                        link=result.get("href", ""),
-                        snippet=result.get("body", ""),
-                    )
-                )
-
+            results = [_convert_ddg_result(result) for result in search_results]
             logger.info(f"DuckDuckGo returned {len(results)} results")
             return results
 

docker/clients/serper_client.py

@@ -16,6 +16,25 @@
 logger = logging.getLogger(__name__)
 
 
+def _convert_organic_results(organic_results: list) -> List[APISearchResult]:
+    """Convert organic search results to APISearchResult objects.
+
+    Args:
+        organic_results: List of organic result dictionaries from Serper API
+
+    Returns:
+        List of APISearchResult objects
+    """
+    return [
+        APISearchResult(
+            title=result.get("title", "Untitled"),
+            link=result.get("link", ""),
+            snippet=result.get("snippet", ""),
+        )
+        for result in organic_results
+    ]
+
+
 def fetch_search_results(query: str, api_key: str) -> List[APISearchResult]:
     """
     Fetches search results from the Serper API.
@@ -38,15 +57,7 @@ def fetch_search_results(query: str, api_key: str) -> List[APISearchResult]:
 
         # Process and return the search results
         if "organic" in data:
-            # Convert to APISearchResult objects
-            return [
-                APISearchResult(
-                    title=result.get("title", "Untitled"),
-                    link=result.get("link", ""),
-                    snippet=result.get("snippet", ""),
-                )
-                for result in data["organic"]
-            ]
+            return _convert_organic_results(data["organic"])
         return []
     except Exception as e:
         logger.error(f"Error fetching search results: {str(e)}")