learningcircuit--local-deep-research/tests/content_fetcher/test_fetcher.py

"""
Tests for ContentFetcher unified interface.
"""

import pytest
from unittest.mock import MagicMock, patch

from local_deep_research.content_fetcher import ContentFetcher, URLType
from local_deep_research.research_library.downloaders.base import DownloadResult


class TestContentFetcherInit:
    """Test ContentFetcher initialization."""

    def test_init_default_timeout(self):
        """Test default timeout."""
        fetcher = ContentFetcher()
        assert fetcher.timeout == 30

    def test_init_custom_timeout(self):
        """Test custom timeout."""
        fetcher = ContentFetcher(timeout=60)
        assert fetcher.timeout == 60

    def test_init_default_js_rendering_disabled(self):
        """JS rendering must be off by default — issue #3826."""
        fetcher = ContentFetcher()
        assert fetcher.enable_js_rendering is False

    def test_init_explicit_js_rendering_enabled(self):
        """Caller can opt in to JS rendering explicitly."""
        fetcher = ContentFetcher(enable_js_rendering=True)
        assert fetcher.enable_js_rendering is True

    def test_html_downloader_inherits_js_rendering_flag(self):
        """The HTML downloader receives the JS rendering toggle."""
        fetcher = ContentFetcher(enable_js_rendering=False)
        downloader = fetcher._get_downloader(URLType.HTML)
        assert downloader is not None
        assert downloader.enable_js_rendering is False

        fetcher_on = ContentFetcher(enable_js_rendering=True)
        downloader_on = fetcher_on._get_downloader(URLType.HTML)
        assert downloader_on is not None
        assert downloader_on.enable_js_rendering is True

    def test_doi_downloader_inherits_js_rendering_flag(self):
        """The DOI downloader (also HTML-based) receives the toggle."""
        fetcher = ContentFetcher(enable_js_rendering=False)
        downloader = fetcher._get_downloader(URLType.DOI)
        assert downloader is not None
        assert downloader.enable_js_rendering is False


class TestContentFetcherGetDownloader:
    """Test downloader selection."""

    def test_get_html_downloader(self):
        """Test HTML downloader is returned for HTML type."""
        fetcher = ContentFetcher()
        downloader = fetcher._get_downloader(URLType.HTML)

        assert downloader is not None
        # Should be cached
        assert fetcher._get_downloader(URLType.HTML) is downloader

    def test_get_arxiv_downloader(self):
        """Test arXiv downloader is returned."""
        from local_deep_research.research_library.downloaders.arxiv import (
            ArxivDownloader,
        )

        fetcher = ContentFetcher()
        downloader = fetcher._get_downloader(URLType.ARXIV)

        # ArxivDownloader is bundled with the project dependencies, so
        # it should always be available in CI; if a future refactor
        # makes it optional, swap this for `is None or isinstance(...)`
        # — but never `is None or is not None`, which asserts nothing.
        assert isinstance(downloader, ArxivDownloader)

    def test_downloader_caching(self):
        """Test downloaders are cached."""
        fetcher = ContentFetcher()

        # Get HTML downloader twice
        d1 = fetcher._get_downloader(URLType.HTML)
        d2 = fetcher._get_downloader(URLType.HTML)

        assert d1 is d2


class TestContentFetcherFetch:
    """Test fetch functionality."""

    @patch(
        "local_deep_research.content_fetcher.fetcher.ContentFetcher._get_downloader"
    )
    def test_fetch_success(self, mock_get_downloader):
        """Test successful fetch."""
        # Setup mock downloader
        mock_downloader = MagicMock()
        mock_downloader.download_with_result.return_value = DownloadResult(
            content=b"Test content from the article.",
            is_success=True,
        )
        mock_downloader.get_metadata.return_value = {"title": "Test Article"}
        mock_get_downloader.return_value = mock_downloader

        fetcher = ContentFetcher()
        result = fetcher.fetch("https://example.com/article")

        assert result["status"] == "success"
        assert "Test content" in result["content"]
        assert result["url"] == "https://example.com/article"
        assert result["source_type"] == "Web Page"

    @patch(
        "local_deep_research.content_fetcher.fetcher.ContentFetcher._get_downloader"
    )
    def test_fetch_failure(self, mock_get_downloader):
        """Test fetch failure."""
        # Setup mock downloader
        mock_downloader = MagicMock()
        mock_downloader.download_with_result.return_value = DownloadResult(
            content=None,
            is_success=False,
            skip_reason="Page not found",
        )
        mock_get_downloader.return_value = mock_downloader

        fetcher = ContentFetcher()
        result = fetcher.fetch("https://example.com/notfound")

        assert result["status"] == "error"
        assert "not found" in result["error"].lower()

    @patch(
        "local_deep_research.content_fetcher.fetcher.ContentFetcher._get_downloader"
    )
    def test_fetch_with_max_length(self, mock_get_downloader):
        """Test fetch with max_length truncation."""
        long_content = "A" * 20000  # 20k characters

        mock_downloader = MagicMock()
        mock_downloader.download_with_result.return_value = DownloadResult(
            content=long_content.encode("utf-8"),
            is_success=True,
        )
        mock_downloader.get_metadata.return_value = {}
        mock_get_downloader.return_value = mock_downloader

        fetcher = ContentFetcher()
        result = fetcher.fetch("https://example.com/long", max_length=1000)

        assert result["status"] == "success"
        assert len(result["content"]) < 20000
        assert "truncated" in result["content"].lower()

    @patch(
        "local_deep_research.content_fetcher.fetcher.ContentFetcher._get_downloader"
    )
    def test_fetch_exception_handling(self, mock_get_downloader):
        """Test fetch handles exceptions gracefully."""
        # Setup mock downloader that raises an exception
        mock_downloader = MagicMock()
        mock_downloader.download_with_result.side_effect = Exception(
            "Network error"
        )
        mock_get_downloader.return_value = mock_downloader

        fetcher = ContentFetcher()
        result = fetcher.fetch("https://example.com/article")

        assert result["status"] == "error"
        assert (
            "error" in result["error"].lower()
            or "network" in result["error"].lower()
        )


class TestContentFetcherFetchText:
    """Test fetch_text convenience method."""

    @patch(
        "local_deep_research.content_fetcher.fetcher.ContentFetcher._get_downloader"
    )
    def test_fetch_text_success(self, mock_get_downloader):
        """Test fetch_text returns text on success."""
        mock_downloader = MagicMock()
        mock_downloader.download_with_result.return_value = DownloadResult(
            content=b"Article text content.",
            is_success=True,
        )
        mock_downloader.get_metadata.return_value = {}
        mock_get_downloader.return_value = mock_downloader

        fetcher = ContentFetcher()
        result = fetcher.fetch_text("https://example.com/article")

        assert result is not None
        assert "Article text content" in result

    @patch(
        "local_deep_research.content_fetcher.fetcher.ContentFetcher._get_downloader"
    )
    def test_fetch_text_failure(self, mock_get_downloader):
        """Test fetch_text returns None on failure."""
        mock_downloader = MagicMock()
        mock_downloader.download_with_result.return_value = DownloadResult(
            content=None,
            is_success=False,
            skip_reason="Error",
        )
        mock_get_downloader.return_value = mock_downloader

        fetcher = ContentFetcher()
        result = fetcher.fetch_text("https://example.com/error")

        assert result is None


class TestContentFetcherGetURLInfo:
    """Test URL info without downloading."""

    def test_get_url_info_arxiv(self):
        """Test URL info for arXiv."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info("https://arxiv.org/abs/2301.12345")

        assert info["url"] == "https://arxiv.org/abs/2301.12345"
        assert info["url_type"] == "arxiv"
        assert info["source_name"] == "arXiv"
        assert info["extracted_id"] == "2301.12345"

    def test_get_url_info_pubmed(self):
        """Test URL info for PubMed."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info("https://pubmed.ncbi.nlm.nih.gov/12345678")

        assert info["url_type"] == "pubmed"
        assert info["source_name"] == "PubMed"
        assert info["extracted_id"] == "12345678"

    def test_get_url_info_generic(self):
        """Test URL info for generic web page."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info("https://example.com/article/123")

        assert info["url_type"] == "html"
        assert info["source_name"] == "Web Page"
        assert info["extracted_id"] is None


class TestContentFetcherURLTypeRouting:
    """Test that URLs are routed to correct downloaders."""

    def test_arxiv_routing(self):
        """Test arXiv URLs are detected correctly."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info("https://arxiv.org/abs/2301.12345")
        assert info["url_type"] == "arxiv"

    def test_pubmed_routing(self):
        """Test PubMed URLs are detected correctly."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info("https://pubmed.ncbi.nlm.nih.gov/12345")
        assert info["url_type"] == "pubmed"

    def test_semantic_scholar_routing(self):
        """Test Semantic Scholar URLs are detected correctly."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info(
            "https://www.semanticscholar.org/paper/Title/abc123"
        )
        assert info["url_type"] == "semantic_scholar"

    def test_biorxiv_routing(self):
        """Test bioRxiv URLs are detected correctly."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info(
            "https://www.biorxiv.org/content/10.1101/123"
        )
        assert info["url_type"] == "biorxiv"

    def test_doi_routing(self):
        """Test DOI URLs are detected correctly."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info("https://doi.org/10.1234/example")
        assert info["url_type"] == "doi"

    def test_pdf_routing(self):
        """Test direct PDF URLs are detected correctly."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info("https://example.com/paper.pdf")
        assert info["url_type"] == "pdf"

    def test_html_routing(self):
        """Test generic HTML URLs are detected correctly."""
        fetcher = ContentFetcher()
        info = fetcher.get_url_info("https://news.ycombinator.com/item?id=123")
        assert info["url_type"] == "html"


class TestContentFetcherIntegration:
    """Integration tests (may require network, marked for CI skip if needed)."""

    @pytest.mark.skip(reason="Requires network access - run manually")
    def test_fetch_real_webpage(self):
        """Test fetching a real webpage."""
        fetcher = ContentFetcher()
        result = fetcher.fetch("https://httpbin.org/html")

        assert result["status"] == "success"
        assert len(result["content"]) > 0

    @pytest.mark.skip(reason="Requires network access - run manually")
    def test_fetch_real_arxiv(self):
        """Test fetching a real arXiv paper."""
        fetcher = ContentFetcher()
        # Use a known stable paper
        result = fetcher.fetch("https://arxiv.org/abs/1706.03762")

        # May succeed or fail depending on arXiv availability
        assert result["status"] in ("success", "error")


class TestContentFetcherSecurity:
    """Security tests for ContentFetcher."""

    def test_fetch_javascript_url_rejected(self):
        """Test that javascript: URLs are rejected."""
        fetcher = ContentFetcher()
        result = fetcher.fetch("javascript:alert('XSS')")

        assert result["status"] == "error"
        assert (
            "invalid" in result["error"].lower()
            or "unsupported" in result["error"].lower()
        )
        assert result["source_type"] == "Invalid URL"

    def test_fetch_data_url_rejected(self):
        """Test that data: URLs are rejected."""
        fetcher = ContentFetcher()
        result = fetcher.fetch("data:text/html,<script>evil()</script>")

        assert result["status"] == "error"
        assert result["source_type"] == "Invalid URL"

    def test_fetch_file_url_rejected(self):
        """Test that file: URLs are rejected."""
        fetcher = ContentFetcher()
        result = fetcher.fetch("file:///etc/passwd")

        assert result["status"] == "error"
        assert result["source_type"] == "Invalid URL"


class TestContentFetcherDefaults:
    """Test default values and configuration."""

    def test_default_max_length_applied(self):
        """Test that default max_length is applied when not specified."""
        from local_deep_research.content_fetcher.fetcher import (
            DEFAULT_MAX_CONTENT_LENGTH,
        )

        # Verify the constant exists and is reasonable
        assert DEFAULT_MAX_CONTENT_LENGTH == 500_000  # 500KB

    @patch(
        "local_deep_research.content_fetcher.fetcher.ContentFetcher._get_downloader"
    )
    def test_fetch_uses_default_max_length(self, mock_get_downloader):
        """Test that fetch applies default max_length to large content."""
        # Create content larger than default max length
        large_content = "X" * 600_000  # 600KB

        mock_downloader = MagicMock()
        mock_downloader.download_with_result.return_value = DownloadResult(
            content=large_content.encode("utf-8"),
            is_success=True,
        )
        mock_downloader.get_metadata.return_value = {}
        mock_get_downloader.return_value = mock_downloader

        fetcher = ContentFetcher()
        result = fetcher.fetch("https://example.com/large")

        assert result["status"] == "success"
        # Content should be truncated to default max length
        assert len(result["content"]) < 600_000
        assert "truncated" in result["content"].lower()