learningcircuit--local-deep-research/tests/web_services/test_pdf_extraction_service.py

"""
Comprehensive tests for PDFExtractionService.
Tests PDF text extraction, metadata extraction, and batch processing.
"""

from unittest.mock import Mock, patch, MagicMock


class TestPDFExtractionServiceExtractTextAndMetadata:
    """Tests for extract_text_and_metadata method."""

    def test_returns_dict_with_required_keys(self, valid_pdf_bytes):
        """Test that extraction returns dict with all required keys."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_text_and_metadata(
            valid_pdf_bytes, "test.pdf"
        )

        assert "text" in result
        assert "pages" in result
        assert "size" in result
        assert "filename" in result
        assert "success" in result
        assert "error" in result

    def test_extracts_text_from_valid_pdf(self, valid_pdf_bytes):
        """Test text extraction from valid PDF."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_text_and_metadata(
            valid_pdf_bytes, "test.pdf"
        )

        # The minimal PDF contains "Hello World"
        assert "Hello World" in result["text"]
        assert result["success"] is True

    def test_returns_page_count(self, valid_pdf_bytes):
        """Test that page count is returned."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_text_and_metadata(
            valid_pdf_bytes, "test.pdf"
        )

        assert result["pages"] == 1

    def test_returns_file_size(self, valid_pdf_bytes):
        """Test that file size is returned."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_text_and_metadata(
            valid_pdf_bytes, "test.pdf"
        )

        assert result["size"] == len(valid_pdf_bytes)

    def test_returns_filename(self, valid_pdf_bytes):
        """Test that filename is returned."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_text_and_metadata(
            valid_pdf_bytes, "my_document.pdf"
        )

        assert result["filename"] == "my_document.pdf"

    def test_handles_invalid_pdf(self, invalid_pdf_bytes):
        """Test handling of invalid/corrupted PDF."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_text_and_metadata(
            invalid_pdf_bytes, "corrupt.pdf"
        )

        assert result["success"] is False
        assert result["error"] is not None
        assert "Failed to extract" in result["error"]

    def test_handles_empty_file(self, empty_file_bytes):
        """Test handling of empty file."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_text_and_metadata(
            empty_file_bytes, "empty.pdf"
        )

        assert result["success"] is False
        assert result["error"] is not None

    def test_handles_pdf_with_no_text(self):
        """Test handling of PDF with no extractable text."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        # Mock pdfplumber to return empty text
        mock_page = Mock()
        mock_page.extract_text.return_value = ""

        mock_pdf = MagicMock()
        mock_pdf.__enter__.return_value = mock_pdf
        mock_pdf.__exit__.return_value = None
        mock_pdf.pages = [mock_page]

        with patch(
            "local_deep_research.web.services.pdf_extraction_service.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdfplumber.open.return_value = mock_pdf

            result = PDFExtractionService.extract_text_and_metadata(
                b"fake pdf content", "empty_text.pdf"
            )

            assert result["success"] is False
            assert "No extractable text found" in result["error"]

    def test_logs_warning_for_no_text(self):
        """Test that warning is logged when no text found."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        mock_page = Mock()
        mock_page.extract_text.return_value = ""

        mock_pdf = MagicMock()
        mock_pdf.__enter__.return_value = mock_pdf
        mock_pdf.__exit__.return_value = None
        mock_pdf.pages = [mock_page]

        with patch(
            "local_deep_research.web.services.pdf_extraction_service.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdfplumber.open.return_value = mock_pdf

            with patch(
                "local_deep_research.web.services.pdf_extraction_service.logger"
            ) as mock_logger:
                PDFExtractionService.extract_text_and_metadata(
                    b"fake pdf", "no_text.pdf"
                )

                mock_logger.warning.assert_called_once()
                assert (
                    "No extractable text" in mock_logger.warning.call_args[0][0]
                )

    def test_logs_success_for_extraction(self, valid_pdf_bytes):
        """Test that success is logged on extraction."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        with patch(
            "local_deep_research.web.services.pdf_extraction_service.logger"
        ) as mock_logger:
            result = PDFExtractionService.extract_text_and_metadata(
                valid_pdf_bytes, "test.pdf"
            )

            if result["success"]:
                mock_logger.info.assert_called()

    def test_logs_exception_on_error(self, invalid_pdf_bytes):
        """Test that exceptions are logged."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        with patch(
            "local_deep_research.web.services.pdf_extraction_service.logger"
        ) as mock_logger:
            PDFExtractionService.extract_text_and_metadata(
                invalid_pdf_bytes, "corrupt.pdf"
            )

            mock_logger.exception.assert_called()

    def test_returns_generic_error_message(self, invalid_pdf_bytes):
        """Test that generic error message is returned (no internal details)."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_text_and_metadata(
            invalid_pdf_bytes, "corrupt.pdf"
        )

        # Should not expose internal exception details
        assert "Failed to extract text from PDF" in result["error"]


class TestPDFExtractionServiceExtractBatch:
    """Tests for extract_batch method."""

    def test_returns_dict_with_required_keys(self, valid_pdf_bytes):
        """Test that batch returns dict with all required keys."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        files_data = [{"content": valid_pdf_bytes, "filename": "test.pdf"}]

        result = PDFExtractionService.extract_batch(files_data)

        assert "results" in result
        assert "total_files" in result
        assert "successful" in result
        assert "failed" in result
        assert "errors" in result

    def test_processes_multiple_files(self, valid_pdf_bytes):
        """Test processing multiple PDF files."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        files_data = [
            {"content": valid_pdf_bytes, "filename": "test1.pdf"},
            {"content": valid_pdf_bytes, "filename": "test2.pdf"},
            {"content": valid_pdf_bytes, "filename": "test3.pdf"},
        ]

        result = PDFExtractionService.extract_batch(files_data)

        assert result["total_files"] == 3
        assert len(result["results"]) == 3

    def test_counts_successful_extractions(self, valid_pdf_bytes):
        """Test counting of successful extractions."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        files_data = [
            {"content": valid_pdf_bytes, "filename": "test1.pdf"},
            {"content": valid_pdf_bytes, "filename": "test2.pdf"},
        ]

        result = PDFExtractionService.extract_batch(files_data)

        assert result["successful"] == 2
        assert result["failed"] == 0

    def test_counts_failed_extractions(
        self, valid_pdf_bytes, invalid_pdf_bytes
    ):
        """Test counting of failed extractions."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        files_data = [
            {"content": valid_pdf_bytes, "filename": "valid.pdf"},
            {"content": invalid_pdf_bytes, "filename": "invalid.pdf"},
        ]

        result = PDFExtractionService.extract_batch(files_data)

        assert result["successful"] == 1
        assert result["failed"] == 1

    def test_aggregates_errors(self, invalid_pdf_bytes):
        """Test error aggregation in batch mode."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        files_data = [
            {"content": invalid_pdf_bytes, "filename": "bad1.pdf"},
            {"content": invalid_pdf_bytes, "filename": "bad2.pdf"},
        ]

        result = PDFExtractionService.extract_batch(files_data)

        assert len(result["errors"]) == 2
        assert any("bad1.pdf" in err for err in result["errors"])
        assert any("bad2.pdf" in err for err in result["errors"])

    def test_handles_empty_batch(self):
        """Test handling of empty file list."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        result = PDFExtractionService.extract_batch([])

        assert result["total_files"] == 0
        assert result["successful"] == 0
        assert result["failed"] == 0
        assert result["results"] == []
        assert result["errors"] == []

    def test_preserves_order_of_results(self, valid_pdf_bytes):
        """Test that results preserve the order of input files."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        files_data = [
            {"content": valid_pdf_bytes, "filename": "first.pdf"},
            {"content": valid_pdf_bytes, "filename": "second.pdf"},
            {"content": valid_pdf_bytes, "filename": "third.pdf"},
        ]

        result = PDFExtractionService.extract_batch(files_data)

        assert result["results"][0]["filename"] == "first.pdf"
        assert result["results"][1]["filename"] == "second.pdf"
        assert result["results"][2]["filename"] == "third.pdf"

    def test_each_result_has_extraction_data(self, valid_pdf_bytes):
        """Test that each result has complete extraction data."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        files_data = [{"content": valid_pdf_bytes, "filename": "test.pdf"}]

        result = PDFExtractionService.extract_batch(files_data)

        file_result = result["results"][0]
        assert "text" in file_result
        assert "pages" in file_result
        assert "size" in file_result
        assert "filename" in file_result
        assert "success" in file_result


class TestPDFExtractionServiceSingleton:
    """Tests for get_pdf_extraction_service singleton."""

    def test_get_service_returns_instance(self):
        """Test that get_pdf_extraction_service returns an instance."""
        from local_deep_research.web.services.pdf_extraction_service import (
            get_pdf_extraction_service,
            PDFExtractionService,
        )

        service = get_pdf_extraction_service()

        assert isinstance(service, PDFExtractionService)

    def test_get_service_returns_same_instance(self):
        """Test that get_pdf_extraction_service returns same instance."""
        from local_deep_research.web.services.pdf_extraction_service import (
            get_pdf_extraction_service,
        )

        service1 = get_pdf_extraction_service()
        service2 = get_pdf_extraction_service()

        assert service1 is service2

    def test_singleton_can_be_reset(self):
        """Test that singleton can be reset for testing."""
        import local_deep_research.web.services.pdf_extraction_service as extraction_module

        # Reset singleton
        extraction_module._pdf_extraction_service = None

        service1 = extraction_module.get_pdf_extraction_service()

        # Reset again
        extraction_module._pdf_extraction_service = None

        service2 = extraction_module.get_pdf_extraction_service()

        # Different instances after reset
        assert service1 is not service2


class TestPDFExtractionServiceMultiPage:
    """Tests for multi-page PDF extraction."""

    def test_extracts_all_pages(self):
        """Test that text is extracted from all pages."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        # Mock multi-page PDF
        mock_page1 = Mock()
        mock_page1.extract_text.return_value = "Page 1 content"

        mock_page2 = Mock()
        mock_page2.extract_text.return_value = "Page 2 content"

        mock_page3 = Mock()
        mock_page3.extract_text.return_value = "Page 3 content"

        mock_pdf = MagicMock()
        mock_pdf.__enter__.return_value = mock_pdf
        mock_pdf.__exit__.return_value = None
        mock_pdf.pages = [mock_page1, mock_page2, mock_page3]

        with patch(
            "local_deep_research.web.services.pdf_extraction_service.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdfplumber.open.return_value = mock_pdf

            result = PDFExtractionService.extract_text_and_metadata(
                b"fake pdf", "multipage.pdf"
            )

            assert result["pages"] == 3
            assert "Page 1 content" in result["text"]
            assert "Page 2 content" in result["text"]
            assert "Page 3 content" in result["text"]

    def test_handles_page_with_no_text(self):
        """Test handling pages that return no text."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        mock_page1 = Mock()
        mock_page1.extract_text.return_value = "Page 1 content"

        mock_page2 = Mock()
        mock_page2.extract_text.return_value = None  # Image-only page

        mock_page3 = Mock()
        mock_page3.extract_text.return_value = "Page 3 content"

        mock_pdf = MagicMock()
        mock_pdf.__enter__.return_value = mock_pdf
        mock_pdf.__exit__.return_value = None
        mock_pdf.pages = [mock_page1, mock_page2, mock_page3]

        with patch(
            "local_deep_research.web.services.pdf_extraction_service.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdfplumber.open.return_value = mock_pdf

            result = PDFExtractionService.extract_text_and_metadata(
                b"fake pdf", "mixed.pdf"
            )

            assert result["pages"] == 3
            assert "Page 1 content" in result["text"]
            assert "Page 3 content" in result["text"]
            assert result["success"] is True

    def test_joins_pages_with_newline(self):
        """Test that pages are joined with newlines."""
        from local_deep_research.web.services.pdf_extraction_service import (
            PDFExtractionService,
        )

        mock_page1 = Mock()
        mock_page1.extract_text.return_value = "First"

        mock_page2 = Mock()
        mock_page2.extract_text.return_value = "Second"

        mock_pdf = MagicMock()
        mock_pdf.__enter__.return_value = mock_pdf
        mock_pdf.__exit__.return_value = None
        mock_pdf.pages = [mock_page1, mock_page2]

        with patch(
            "local_deep_research.web.services.pdf_extraction_service.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdfplumber.open.return_value = mock_pdf

            result = PDFExtractionService.extract_text_and_metadata(
                b"fake pdf", "test.pdf"
            )

            # Text should be joined with newline
            assert result["text"] == "First\nSecond"