learningcircuit--local-deep-research/tests/test_utils.py

# allow: no-sut-import — shared test-helper module (not a test); provides utilities imported by other tests
"""
Test utilities and helper functions - incorporating scottvr's patterns.

This module provides common utilities for tests.
"""

import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional
from unittest.mock import Mock


def add_src_to_path():
    """Add the src directory to Python path for imports."""
    src_path = str(Path(__file__).parent.parent / "src")
    if src_path not in sys.path:
        sys.path.insert(0, src_path)


def mock_api_response(
    status_code: int = 200,
    json_data: Optional[Dict] = None,
    text: Optional[str] = None,
    raise_exception: Optional[Exception] = None,
):
    """
    Create a mock API response object.

    Args:
        status_code: HTTP status code
        json_data: JSON response data
        text: Text response data
        raise_exception: Exception to raise when accessed

    Returns:
        Mock response object
    """
    mock_response = Mock()
    mock_response.status_code = status_code

    if raise_exception:
        mock_response.json.side_effect = raise_exception
        mock_response.text = str(raise_exception)
    else:
        if json_data is not None:
            mock_response.json.return_value = json_data
            mock_response.text = json.dumps(json_data)
        elif text is not None:
            mock_response.text = text
            mock_response.json.side_effect = json.JSONDecodeError(
                "Not JSON", text, 0
            )
        else:
            mock_response.json.return_value = {}
            mock_response.text = "{}"

    return mock_response


def assert_search_result_format(result: Dict[str, Any]):
    """
    Assert that a search result has the correct format.

    Args:
        result: Search result dictionary to validate
    """
    assert isinstance(result, dict), "Result should be a dictionary"
    assert "title" in result, "Result should have a title"
    # Some engines use 'link', others use 'url'
    assert "link" in result or "url" in result, (
        "Result should have a link or url"
    )
    assert "snippet" in result, "Result should have a snippet"
    assert isinstance(result["title"], str), "Title should be a string"
    # Check whichever field is present
    if "link" in result:
        assert isinstance(result["link"], str), "Link should be a string"
    if "url" in result:
        assert isinstance(result["url"], str), "URL should be a string"
    assert isinstance(result["snippet"], str), "Snippet should be a string"


def assert_findings_format(findings: Dict[str, Any]):
    """
    Assert that findings have the correct format.

    Args:
        findings: Findings dictionary to validate
    """
    assert isinstance(findings, dict), "Findings should be a dictionary"
    assert "findings" in findings, "Should have findings list"
    assert "current_knowledge" in findings, "Should have current_knowledge"
    assert "iterations" in findings, "Should have iterations count"
    assert "questions_by_iteration" in findings, (
        "Should have questions_by_iteration"
    )

    assert isinstance(findings["findings"], list), "Findings should be a list"
    assert isinstance(findings["current_knowledge"], str), (
        "Knowledge should be a string"
    )
    assert isinstance(findings["iterations"], int), (
        "Iterations should be an integer"
    )
    assert isinstance(findings["questions_by_iteration"], dict), (
        "Questions should be a dict"
    )


def create_test_research_context(query: str = "test query") -> Dict[str, Any]:
    """
    Create a test research context with all required fields.

    Args:
        query: Research query

    Returns:
        Complete research context dictionary
    """
    return {
        "query": query,
        "search_tool": "searxng",
        "llm_provider": "ollama",
        "llm_model": "gemma3:12b",
        "search_iterations": 3,
        "questions_per_iteration": 2,
        "temperature": 0.7,
        "max_tokens": 4096,
        "enable_fact_checking": True,
        "search_strategy": "standard",
    }


def mock_progress_callback():
    """
    Create a mock progress callback that tracks calls.

    Returns:
        Tuple of (callback function, calls list)
    """
    calls = []

    def callback(message: str, progress: int, metadata: Optional[Dict] = None):
        calls.append(
            {
                "message": message,
                "progress": progress,
                "metadata": metadata or {},
            }
        )

    return callback, calls


def assert_progress_callback_called(
    calls: List[Dict],
    expected_message: str = None,
    expected_progress: int = None,
):
    """
    Assert that a progress callback was called with expected parameters.

    Args:
        calls: List of callback calls
        expected_message: Expected message (substring match)
        expected_progress: Expected progress value
    """
    assert len(calls) > 0, "Progress callback should have been called"

    if expected_message:
        messages = [call["message"] for call in calls]
        assert any(expected_message in msg for msg in messages), (
            f"Expected message '{expected_message}' not found in calls: {messages}"
        )

    if expected_progress is not None:
        progresses = [call["progress"] for call in calls]
        assert expected_progress in progresses, (
            f"Expected progress {expected_progress} not found in calls: {progresses}"
        )


class MockCache:
    """Mock cache implementation for testing."""

    def __init__(self):
        self.cache = {}
        self.hits = 0
        self.misses = 0

    def get(self, key: str) -> Optional[Any]:
        if key in self.cache:
            self.hits += 1
            return self.cache[key]
        self.misses += 1
        return None

    def set(self, key: str, value: Any):
        self.cache[key] = value

    def clear(self):
        self.cache.clear()
        self.hits = 0
        self.misses = 0

    def get_stats(self) -> Dict[str, int]:
        return {
            "hits": self.hits,
            "misses": self.misses,
            "size": len(self.cache),
        }