learningcircuit--local-deep-research/tests/search_engines/test_search_engine_paperless.py

"""
Comprehensive tests for the Paperless-ngx search engine.
Tests initialization, API requests, search functionality, and document conversion.

Note: These tests mock HTTP requests to avoid requiring an actual Paperless instance.
"""

import pytest
from unittest.mock import Mock, patch


@pytest.fixture
def mock_requests_get():
    """Mock requests.get to avoid actual HTTP calls."""
    with patch(
        "local_deep_research.web_search_engines.engines.search_engine_paperless.requests.get"
    ) as mock_get:
        yield mock_get


class TestPaperlessSearchEngineInit:
    """Tests for Paperless search engine initialization."""

    def test_init_with_api_url_and_token(self, mock_requests_get):
        """Test initialization with API URL and token."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(
            api_url="http://localhost:8000", api_key="test_token"
        )

        assert engine.api_url == "http://localhost:8000"
        assert engine.api_token == "test_token"
        assert engine.headers["Authorization"] == "Token test_token"

    def test_init_with_api_token_compatibility(self, mock_requests_get):
        """Test initialization with api_token parameter (backwards compatibility)."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(
            api_url="http://localhost:8000", api_token="legacy_token"
        )

        assert engine.api_token == "legacy_token"

    def test_init_strips_trailing_slash(self, mock_requests_get):
        """Test that trailing slash is stripped from API URL."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(api_url="http://localhost:8000/")

        assert engine.api_url == "http://localhost:8000"

    def test_init_default_values(self, mock_requests_get):
        """Test initialization with default values."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine()

        assert engine.max_results == 10
        assert engine.timeout == 30
        assert engine.verify_ssl is True
        assert engine.include_content is True

    def test_init_with_custom_max_results(self, mock_requests_get):
        """Test initialization with custom max results."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(max_results=25)

        assert engine.max_results == 25

    def test_init_with_custom_timeout(self, mock_requests_get):
        """Test initialization with custom timeout."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(timeout=60)

        assert engine.timeout == 60

    def test_init_with_ssl_disabled(self, mock_requests_get):
        """Test initialization with SSL verification disabled."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(verify_ssl=False)

        assert engine.verify_ssl is False

    def test_init_with_content_disabled(self, mock_requests_get):
        """Test initialization with content inclusion disabled."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(include_content=False)

        assert engine.include_content is False

    def test_init_from_settings_snapshot(self, mock_requests_get):
        """Test initialization from settings snapshot."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        settings = {
            "search.engine.web.paperless.default_params.api_url": "http://paperless.local:9000",
            "search.engine.web.paperless.api_key": "snapshot_token",
        }

        engine = PaperlessSearchEngine(settings_snapshot=settings)

        assert engine.api_url == "http://paperless.local:9000"
        assert engine.api_token == "snapshot_token"

    def test_init_with_no_token_empty_headers(self, mock_requests_get):
        """Test that headers are empty when no token provided."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(api_url="http://localhost:8000")

        assert "Authorization" not in engine.headers


class TestPaperlessAPIRequest:
    """Tests for Paperless API request handling."""

    @pytest.fixture
    def engine(self, mock_requests_get):
        """Create a Paperless engine."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        return PaperlessSearchEngine(
            api_url="http://localhost:8000", api_key="test_token"
        )

    def test_make_request_success(self, engine, mock_requests_get):
        """Test successful API request."""
        mock_response = Mock()
        mock_response.json.return_value = {"results": [], "count": 0}
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        result = engine._make_request("/api/documents/", {"query": "test"})

        assert result == {"results": [], "count": 0}
        mock_requests_get.assert_called_once()

    def test_make_request_includes_auth_header(self, engine, mock_requests_get):
        """Test that request includes authorization header."""
        mock_response = Mock()
        mock_response.json.return_value = {}
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        engine._make_request("/api/documents/")

        call_kwargs = mock_requests_get.call_args[1]
        assert call_kwargs["headers"]["Authorization"] == "Token test_token"

    def test_make_request_handles_exception(self, engine, mock_requests_get):
        """Test that request exceptions are handled gracefully."""
        import requests

        mock_requests_get.side_effect = requests.exceptions.RequestException(
            "Connection error"
        )

        result = engine._make_request("/api/documents/")

        assert result == {}

    def test_make_request_uses_correct_timeout(self, engine, mock_requests_get):
        """Test that request uses configured timeout."""
        mock_response = Mock()
        mock_response.json.return_value = {}
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        engine._make_request("/api/documents/")

        call_kwargs = mock_requests_get.call_args[1]
        assert call_kwargs["timeout"] == 30


class TestPaperlessSearchExecution:
    """Tests for Paperless search execution."""

    @pytest.fixture
    def engine(self, mock_requests_get):
        """Create a Paperless engine."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        return PaperlessSearchEngine(
            api_url="http://localhost:8000",
            api_key="test_token",
            max_results=10,
        )

    def test_get_previews_success(self, engine, mock_requests_get):
        """Test successful preview retrieval."""
        mock_response = Mock()
        mock_response.json.return_value = {
            "results": [
                {
                    "id": 1,
                    "title": "Test Document",
                    "content": "This is test content about the query.",
                    "__search_hit__": {
                        "score": 0.95,
                        "rank": 1,
                        "highlights": "Test <span>query</span> content",
                    },
                }
            ],
            "count": 1,
        }
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        previews = engine._get_previews("test query")

        assert len(previews) == 1
        assert previews[0]["title"] == "Test Document"
        assert "query" in previews[0]["snippet"]

    def test_get_previews_empty_results(self, engine, mock_requests_get):
        """Test preview retrieval with no results."""
        mock_response = Mock()
        mock_response.json.return_value = {"results": [], "count": 0}
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        previews = engine._get_previews("nonexistent query")

        assert previews == []

    def test_get_previews_handles_exception(self, engine, mock_requests_get):
        """Test that preview exceptions are handled."""
        import requests

        mock_requests_get.side_effect = requests.exceptions.RequestException(
            "Error"
        )

        previews = engine._get_previews("test")

        assert previews == []

    def test_multi_pass_search_deduplicates(self, engine, mock_requests_get):
        """Test that multi-pass search deduplicates results."""
        # Return same document in multiple passes
        mock_response = Mock()
        mock_response.json.return_value = {
            "results": [
                {"id": 1, "title": "Doc 1", "__search_hit__": {"score": 0.9}},
                {"id": 2, "title": "Doc 2", "__search_hit__": {"score": 0.8}},
            ],
            "count": 2,
        }
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        results = engine._multi_pass_search("test")

        # Should deduplicate by doc_id
        assert len(results) == 2


class TestPaperlessDocumentConversion:
    """Tests for document conversion to LDR format."""

    @pytest.fixture
    def engine(self, mock_requests_get):
        """Create a Paperless engine."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        return PaperlessSearchEngine(api_url="http://localhost:8000")

    def test_convert_document_with_highlights(self, engine, mock_requests_get):
        """Test document conversion with search highlights."""
        doc_data = {
            "id": 123,
            "title": "Invoice 2024",
            "content": "Full content here",
            "correspondent_name": "Acme Corp",
            "created": "2024-01-15",
            "__search_hit__": {
                "score": 0.95,
                "rank": 1,
                "highlights": "Found <span>matching</span> text here",
            },
        }

        preview = engine._convert_document_to_preview(doc_data, "matching")

        assert "Invoice 2024" in preview["title"]
        assert preview["url"] == "http://localhost:8000/documents/123/details"
        assert "matching" in preview["snippet"]
        assert preview["author"] == "Acme Corp"
        assert preview["source"] == "Paperless"

    def test_convert_document_cleans_html_tags(self, engine, mock_requests_get):
        """Test that HTML tags are cleaned from highlights."""
        doc_data = {
            "id": 1,
            "title": "Test",
            "__search_hit__": {
                "highlights": "<span class='match'>keyword</span> in text"
            },
        }

        preview = engine._convert_document_to_preview(doc_data)

        assert "<span" not in preview["snippet"]
        assert "</span>" not in preview["snippet"]
        # Highlights are converted to markdown bold
        assert "**keyword**" in preview["snippet"]

    def test_convert_document_without_highlights(
        self, engine, mock_requests_get
    ):
        """Test document conversion when no highlights available."""
        doc_data = {
            "id": 1,
            "title": "Test Document",
            "content": "This is the document content with the query term inside.",
        }

        preview = engine._convert_document_to_preview(doc_data, "query")

        # Should use content as fallback
        assert "content" in preview["snippet"]

    def test_convert_document_builds_enhanced_title(
        self, engine, mock_requests_get
    ):
        """Test that enhanced title includes metadata."""
        doc_data = {
            "id": 1,
            "title": "Report Q4",
            "correspondent_name": "Finance Dept",
            "document_type_name": "Financial Report",
            "created": "2024-03-15",
            "__search_hit__": {"highlights": "text"},
        }

        preview = engine._convert_document_to_preview(doc_data)

        # Enhanced title should include correspondent, title, type, year
        assert "Finance Dept" in preview["title"]
        assert "Report Q4" in preview["title"]
        assert "Financial Report" in preview["title"]
        assert "2024" in preview["title"]

    def test_convert_document_includes_metadata(
        self, engine, mock_requests_get
    ):
        """Test that metadata is included in preview."""
        doc_data = {
            "id": 42,
            "title": "Test",
            "correspondent_name": "Test Sender",
            "document_type_name": "Invoice",
            "created": "2024-01-01",
            "modified": "2024-01-02",
            "archive_serial_number": "ASN-001",
            "__search_hit__": {"score": 0.85, "rank": 3, "highlights": "text"},
        }

        preview = engine._convert_document_to_preview(doc_data)

        assert preview["metadata"]["doc_id"] == "42"
        assert preview["metadata"]["correspondent"] == "Test Sender"
        assert preview["metadata"]["document_type"] == "Invoice"
        assert preview["metadata"]["search_score"] == 0.85
        assert preview["metadata"]["search_rank"] == 3

    def test_convert_document_multiple_highlights(
        self, engine, mock_requests_get
    ):
        """Test document conversion with multiple highlights returns multiple previews."""
        doc_data = {
            "id": 1,
            "title": "Multi-match Document",
            "__search_hit__": {
                "highlights": [
                    "First <span>match</span> here",
                    "Second <span>match</span> there",
                    "Third <span>match</span> elsewhere",
                ]
            },
        }

        previews = engine._convert_document_to_preview(doc_data)

        # Should return list of previews for multiple highlights
        assert isinstance(previews, list)
        assert len(previews) == 3
        assert "(excerpt 1)" in previews[0]["title"]
        assert "(excerpt 2)" in previews[1]["title"]


class TestPaperlessFullContent:
    """Tests for Paperless full content retrieval."""

    @pytest.fixture
    def engine(self, mock_requests_get):
        """Create a Paperless engine."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        return PaperlessSearchEngine(api_url="http://localhost:8000")

    def test_get_full_content_from_raw_data(self, engine, mock_requests_get):
        """Test full content extraction from raw data."""
        items = [
            {
                "title": "Test Doc",
                "snippet": "Short snippet",
                "metadata": {"doc_id": "1"},
                "_raw_data": {"content": "This is the full document content."},
            }
        ]

        results = engine._get_full_content(items)

        assert (
            results[0]["full_content"] == "This is the full document content."
        )
        assert "_raw_data" not in results[0]

    def test_get_full_content_fetches_if_missing(
        self, engine, mock_requests_get
    ):
        """Test that content is fetched if not in raw data."""
        mock_response = Mock()
        mock_response.json.return_value = {"content": "Fetched content"}
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        items = [
            {
                "title": "Test",
                "snippet": "Snippet",
                "metadata": {"doc_id": "123"},
                "_raw_data": {},  # No content in raw data
            }
        ]

        results = engine._get_full_content(items)

        assert results[0]["full_content"] == "Fetched content"

    def test_get_full_content_disabled(self, mock_requests_get):
        """Test that content is not fetched when disabled."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        engine = PaperlessSearchEngine(
            api_url="http://localhost:8000", include_content=False
        )

        items = [{"title": "Test", "snippet": "Snippet"}]

        results = engine._get_full_content(items)

        # Should return items unchanged
        assert results == items

    def test_get_full_content_handles_exception(
        self, engine, mock_requests_get
    ):
        """Test that exceptions during full content retrieval are handled."""
        items = [
            {
                "title": "Test",
                "snippet": "Snippet fallback",
                "metadata": {"doc_id": "1"},
                # Missing _raw_data will cause issues
            }
        ]

        results = engine._get_full_content(items)

        # Should fallback to snippet
        assert results[0]["full_content"] == "Snippet fallback"


class TestPaperlessQueryExpansion:
    """Tests for LLM-based query expansion."""

    @pytest.fixture
    def engine(self, mock_requests_get):
        """Create a Paperless engine without LLM."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        return PaperlessSearchEngine(api_url="http://localhost:8000")

    def test_expand_query_without_llm(self, engine, mock_requests_get):
        """Test that query is returned unchanged without LLM."""
        result = engine._expand_query_with_llm("original query")

        assert result == "original query"

    def test_expand_query_with_llm(self, mock_requests_get):
        """Test query expansion with LLM."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        mock_llm = Mock()
        mock_llm.invoke.return_value = Mock(
            content='invoice OR "billing statement" OR receipt'
        )

        engine = PaperlessSearchEngine(
            api_url="http://localhost:8000", llm=mock_llm
        )

        result = engine._expand_query_with_llm("find my invoices")

        assert "invoice" in result.lower() or "billing" in result.lower()

    def test_expand_query_handles_llm_exception(self, mock_requests_get):
        """Test that LLM exceptions fall back to original query."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        mock_llm = Mock()
        mock_llm.invoke.side_effect = Exception("LLM error")

        engine = PaperlessSearchEngine(
            api_url="http://localhost:8000", llm=mock_llm
        )

        result = engine._expand_query_with_llm("test query")

        assert result == "test query"


class TestPaperlessRun:
    """Tests for Paperless run method."""

    @pytest.fixture
    def engine(self, mock_requests_get):
        """Create a Paperless engine."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        return PaperlessSearchEngine(api_url="http://localhost:8000")

    def test_run_success(self, engine, mock_requests_get):
        """Test successful search run."""
        mock_response = Mock()
        mock_response.json.return_value = {
            "results": [
                {
                    "id": 1,
                    "title": "Test Doc",
                    "content": "Full content",
                    "__search_hit__": {"score": 0.9, "highlights": "text"},
                }
            ],
            "count": 1,
        }
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        results = engine.run("test query")

        assert len(results) == 1
        assert "full_content" in results[0]

    def test_run_empty_results(self, engine, mock_requests_get):
        """Test run with no results."""
        mock_response = Mock()
        mock_response.json.return_value = {"results": [], "count": 0}
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        results = engine.run("nonexistent")

        assert results == []

    def test_run_handles_exception(self, engine, mock_requests_get):
        """Test that run handles exceptions."""
        import requests

        mock_requests_get.side_effect = requests.exceptions.RequestException(
            "Error"
        )

        results = engine.run("test")

        assert results == []


class TestPaperlessUtilities:
    """Tests for Paperless utility methods."""

    @pytest.fixture
    def engine(self, mock_requests_get):
        """Create a Paperless engine."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        return PaperlessSearchEngine(
            api_url="http://localhost:8000", api_key="token"
        )

    def test_test_connection_success(self, engine, mock_requests_get):
        """Test connection test success."""
        mock_response = Mock()
        mock_response.json.return_value = {"status": "ok"}
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        result = engine.test_connection()

        assert result is True

    def test_test_connection_failure(self, engine, mock_requests_get):
        """Test connection test failure."""
        import requests

        mock_requests_get.side_effect = requests.exceptions.RequestException(
            "Connection failed"
        )

        result = engine.test_connection()

        assert result is False

    def test_get_document_count_success(self, engine, mock_requests_get):
        """Test document count retrieval."""
        mock_response = Mock()
        mock_response.json.return_value = {"count": 150, "results": []}
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        count = engine.get_document_count()

        assert count == 150

    def test_get_document_count_error(self, engine, mock_requests_get):
        """Test document count on error."""
        import requests

        mock_requests_get.side_effect = requests.exceptions.RequestException(
            "Error"
        )

        count = engine.get_document_count()

        assert count == -1


class TestPaperlessAsync:
    """Tests for Paperless async methods."""

    @pytest.fixture
    def engine(self, mock_requests_get):
        """Create a Paperless engine."""
        from local_deep_research.web_search_engines.engines.search_engine_paperless import (
            PaperlessSearchEngine,
        )

        return PaperlessSearchEngine(api_url="http://localhost:8000")

    @pytest.mark.asyncio
    async def test_arun_falls_back_to_sync(self, engine, mock_requests_get):
        """Test that arun falls back to sync run."""
        mock_response = Mock()
        mock_response.json.return_value = {
            "results": [
                {
                    "id": 1,
                    "title": "Test",
                    "__search_hit__": {"highlights": "text"},
                }
            ],
            "count": 1,
        }
        mock_response.raise_for_status = Mock()
        mock_requests_get.return_value = mock_response

        results = await engine.arun("test")

        assert len(results) == 1