learningcircuit--local-deep-research/tests/retriever_integration/test_e2e_basic.py

"""End-to-end tests for retriever integration."""

import pytest
from langchain_core.retrievers import BaseRetriever, Document

from local_deep_research.api import detailed_research, quick_summary
from local_deep_research.web_search_engines.retriever_registry import (
    retriever_registry,
)
from local_deep_research.web_search_engines.search_engine_factory import (
    create_search_engine,
)

# Minimal non-empty snapshot so the factory's egress PEP can run; under the
# default ``both`` scope a local retriever is permitted. The factory now
# fails closed (no_snapshot) without one.
_BOTH_SNAPSHOT = {"policy.egress_scope": {"value": "both"}}


class SimpleRetriever(BaseRetriever):
    """Simple retriever that returns fixed content."""

    def _get_relevant_documents(self, query: str):
        return [
            Document(
                page_content=f"This is a comprehensive answer about {query}. It contains detailed information that should help answer the question.",
                metadata={
                    "title": f"Document about {query}",
                    "source": "test_knowledge_base",
                    "author": "Test System",
                    "score": 0.95,
                },
            ),
            Document(
                page_content=f"Additional context about {query}. This provides supporting information and examples.",
                metadata={
                    "title": f"Supporting info for {query}",
                    "source": "test_knowledge_base",
                    "score": 0.85,
                },
            ),
        ]

    async def _aget_relevant_documents(self, query: str):
        return self._get_relevant_documents(query)


@pytest.mark.skipif(
    condition=True,  # Skip in CI by default as it requires full LLM
    reason="E2E test requires working LLM - set E2E_TESTS=true to run",
)
class TestE2EBasic:
    """Basic end-to-end tests that use real components."""

    def test_simple_retriever_e2e(self):
        """Test using a retriever with quick_summary end-to-end."""
        import os

        if not os.environ.get("E2E_TESTS"):
            pytest.skip("Set E2E_TESTS=true to run end-to-end tests")

        # Create retriever
        retriever = SimpleRetriever()

        # Use with quick_summary
        result = quick_summary(
            query="What is machine learning?",
            retrievers={"test_kb": retriever},
            search_tool="test_kb",
            iterations=1,
            questions_per_iteration=1,
        )

        # Verify basic structure
        assert "summary" in result
        assert len(result["summary"]) > 0
        assert "findings" in result
        assert "sources" in result

    def test_retriever_through_factory_e2e(self):
        """Test creating retriever search engine through factory."""
        import os

        if not os.environ.get("E2E_TESTS"):
            pytest.skip("Set E2E_TESTS=true to run end-to-end tests")

        # Register retriever
        retriever = SimpleRetriever()
        retriever_registry.register("e2e_test", retriever)

        # Create search engine through factory
        engine = create_search_engine(
            "e2e_test", settings_snapshot=_BOTH_SNAPSHOT
        )

        assert engine is not None
        assert engine.name == "e2e_test"

        # Test search
        results = engine.run("test query")
        assert len(results) == 2
        assert "Document about test query" in results[0]["title"]

    def test_multiple_retrievers_e2e(self):
        """Test using multiple retrievers."""
        import os

        if not os.environ.get("E2E_TESTS"):
            pytest.skip("Set E2E_TESTS=true to run end-to-end tests")

        # Create multiple retrievers with different content
        tech_retriever = SimpleRetriever()
        business_retriever = SimpleRetriever()

        result = detailed_research(
            query="What are the implications of AI?",
            retrievers={"tech": tech_retriever, "business": business_retriever},
            search_tool="tech",
            iterations=1,
        )

        assert result["query"] == "What are the implications of AI?"
        assert "research_id" in result
        assert "metadata" in result


class TestBasicIntegration:
    """Basic integration tests that don't require LLM."""

    def test_retriever_registration_and_search(self):
        """Test that retrievers can be registered and used for search."""
        # Create and register retriever
        retriever = SimpleRetriever()
        retriever_registry.register("integration_test", retriever)

        # Create search engine
        engine = create_search_engine(
            "integration_test", settings_snapshot=_BOTH_SNAPSHOT
        )

        # Perform search
        results = engine.run("integration test query")

        # Verify results
        assert len(results) == 2
        assert results[0]["source"] == "integration_test"
        assert results[0]["retriever_type"] == "SimpleRetriever"
        assert "integration test query" in results[0]["title"]
        assert results[0]["score"] == 0.95

    def test_retriever_appears_in_config(self):
        """Test that registered retrievers appear in search config."""
        from local_deep_research.web_search_engines.search_engines_config import (
            search_config,
        )

        # Register retriever
        retriever = SimpleRetriever()
        retriever_registry.register("config_test", retriever)

        # Get config
        config = search_config()

        # Verify retriever is in config
        assert "config_test" in config
        assert config["config_test"]["is_retriever"]
        assert config["config_test"]["class_name"] == "RetrieverSearchEngine"

    def test_retriever_cleanup_between_tests(self):
        """Test that registry is cleaned between tests."""
        # Should start empty due to cleanup fixture
        assert len(retriever_registry.list_registered()) == 0

        # Register some retrievers
        for i in range(3):
            retriever_registry.register(f"cleanup_test_{i}", SimpleRetriever())

        assert len(retriever_registry.list_registered()) == 3
        # Cleanup fixture will clear these after test