d4vinci--scrapling/tests/fetchers/test_utils.py

import pytest

from scrapling.engines.toolbelt.convertor import ResponseFactory
from scrapling.engines.toolbelt.custom import StatusText, Response
from scrapling.engines.toolbelt.navigation import (
    construct_proxy_dict,
    create_intercept_handler,
    create_async_intercept_handler,
    _is_domain_blocked,
)
from scrapling.engines.toolbelt.fingerprints import get_os_name, generate_headers


@pytest.fixture
def content_type_map():
    return {
        # A map generated by ChatGPT for most possible `content_type` values and the expected outcome
        "text/html; charset=UTF-8": "UTF-8",
        "text/html; charset=ISO-8859-1": "ISO-8859-1",
        "text/html": "ISO-8859-1",
        "application/json; charset=UTF-8": "UTF-8",
        "application/json": "utf-8",
        "text/json": "utf-8",
        "application/javascript; charset=UTF-8": "UTF-8",
        "application/javascript": "utf-8",
        "text/plain; charset=UTF-8": "UTF-8",
        "text/plain; charset=ISO-8859-1": "ISO-8859-1",
        "text/plain": "ISO-8859-1",
        "application/xhtml+xml; charset=UTF-8": "UTF-8",
        "application/xhtml+xml": "utf-8",
        "text/html; charset=windows-1252": "windows-1252",
        "application/json; charset=windows-1252": "windows-1252",
        "text/plain; charset=windows-1252": "windows-1252",
        'text/html; charset="UTF-8"': "UTF-8",
        'text/html; charset="ISO-8859-1"': "ISO-8859-1",
        'text/html; charset="windows-1252"': "windows-1252",
        'application/json; charset="UTF-8"': "UTF-8",
        'application/json; charset="ISO-8859-1"': "ISO-8859-1",
        'application/json; charset="windows-1252"': "windows-1252",
        'text/json; charset="UTF-8"': "UTF-8",
        'application/javascript; charset="UTF-8"': "UTF-8",
        'application/javascript; charset="ISO-8859-1"': "ISO-8859-1",
        'text/plain; charset="UTF-8"': "UTF-8",
        'text/plain; charset="ISO-8859-1"': "ISO-8859-1",
        'text/plain; charset="windows-1252"': "windows-1252",
        'application/xhtml+xml; charset="UTF-8"': "UTF-8",
        'application/xhtml+xml; charset="ISO-8859-1"': "ISO-8859-1",
        'application/xhtml+xml; charset="windows-1252"': "windows-1252",
        'text/html; charset="US-ASCII"': "US-ASCII",
        'application/json; charset="US-ASCII"': "US-ASCII",
        'text/plain; charset="US-ASCII"': "US-ASCII",
        'text/html; charset="Shift_JIS"': "Shift_JIS",
        'application/json; charset="Shift_JIS"': "Shift_JIS",
        'text/plain; charset="Shift_JIS"': "Shift_JIS",
        'application/xml; charset="UTF-8"': "UTF-8",
        'application/xml; charset="ISO-8859-1"': "ISO-8859-1",
        "application/xml": "utf-8",
        'text/xml; charset="UTF-8"': "UTF-8",
        'text/xml; charset="ISO-8859-1"': "ISO-8859-1",
        "text/xml": "utf-8",
    }


@pytest.fixture
def status_map():
    return {
        100: "Continue",
        101: "Switching Protocols",
        102: "Processing",
        103: "Early Hints",
        200: "OK",
        201: "Created",
        202: "Accepted",
        203: "Non-Authoritative Information",
        204: "No Content",
        205: "Reset Content",
        206: "Partial Content",
        207: "Multi-Status",
        208: "Already Reported",
        226: "IM Used",
        300: "Multiple Choices",
        301: "Moved Permanently",
        302: "Found",
        303: "See Other",
        304: "Not Modified",
        305: "Use Proxy",
        307: "Temporary Redirect",
        308: "Permanent Redirect",
        400: "Bad Request",
        401: "Unauthorized",
        402: "Payment Required",
        403: "Forbidden",
        404: "Not Found",
        405: "Method Not Allowed",
        406: "Not Acceptable",
        407: "Proxy Authentication Required",
        408: "Request Timeout",
        409: "Conflict",
        410: "Gone",
        411: "Length Required",
        412: "Precondition Failed",
        413: "Payload Too Large",
        414: "URI Too Long",
        415: "Unsupported Media Type",
        416: "Range Not Satisfiable",
        417: "Expectation Failed",
        418: "I'm a teapot",
        421: "Misdirected Request",
        422: "Unprocessable Entity",
        423: "Locked",
        424: "Failed Dependency",
        425: "Too Early",
        426: "Upgrade Required",
        428: "Precondition Required",
        429: "Too Many Requests",
        431: "Request Header Fields Too Large",
        451: "Unavailable For Legal Reasons",
        500: "Internal Server Error",
        501: "Not Implemented",
        502: "Bad Gateway",
        503: "Service Unavailable",
        504: "Gateway Timeout",
        505: "HTTP Version Not Supported",
        506: "Variant Also Negotiates",
        507: "Insufficient Storage",
        508: "Loop Detected",
        510: "Not Extended",
        511: "Network Authentication Required",
    }


def test_parsing_response_status(status_map):
    """Test if using different http responses' status codes returns the expected result"""
    for status_code, expected_status_text in status_map.items():
        assert StatusText.get(status_code) == expected_status_text


def test_unknown_status_code():
    """Test handling of an unknown status code"""
    assert StatusText.get(1000) == "Unknown Status Code"


# The private classmethod is name-mangled; resolve it once for the tests below.
_extract_encoding = getattr(ResponseFactory, "_ResponseFactory__extract_browser_encoding")


def test_browser_encoding_unquoted_charset():
    """A charset declared without quotes is returned verbatim."""
    assert _extract_encoding("text/html; charset=utf-8") == "utf-8"
    assert _extract_encoding("text/html; charset=ISO-8859-1") == "ISO-8859-1"
    assert _extract_encoding("text/html;charset=windows-1252") == "windows-1252"


def test_browser_encoding_quoted_charset():
    """A quoted charset value (RFC 7231 allows quoting) is unwrapped, not dropped."""
    assert _extract_encoding('text/html; charset="utf-8"') == "utf-8"
    assert _extract_encoding('text/html; charset="ISO-8859-1"') == "ISO-8859-1"
    assert _extract_encoding("text/html; charset='Shift_JIS'") == "Shift_JIS"
    assert _extract_encoding('text/plain; charset="windows-1252"; boundary=x') == "windows-1252"


def test_browser_encoding_defaults_when_missing():
    """Fall back to the default when no charset is present or the header is empty."""
    assert _extract_encoding("text/html") == "utf-8"
    assert _extract_encoding("") == "utf-8"
    assert _extract_encoding(None) == "utf-8"


class TestConstructProxyDict:
    """Test proxy dictionary construction"""

    def test_proxy_string_basic(self):
        """Test a basic proxy string"""
        result = construct_proxy_dict("http://proxy.example.com:8080")

        expected = {"server": "http://proxy.example.com:8080", "username": "", "password": ""}
        assert result == expected

    def test_proxy_string_with_auth(self):
        """Test proxy string with authentication"""
        result = construct_proxy_dict("http://user:pass@proxy.example.com:8080")

        expected = {"server": "http://proxy.example.com:8080", "username": "user", "password": "pass"}
        assert result == expected

    def test_proxy_dict_input(self):
        """Test proxy dictionary input"""
        input_dict = {"server": "http://proxy.example.com:8080", "username": "user", "password": "pass"}
        result = construct_proxy_dict(input_dict)

        assert result == input_dict

    def test_proxy_dict_minimal(self):
        """Test minimal proxy dictionary"""
        input_dict = {"server": "http://proxy.example.com:8080"}
        result = construct_proxy_dict(input_dict)

        expected = {"server": "http://proxy.example.com:8080", "username": "", "password": ""}
        assert result == expected

    def test_invalid_proxy_string(self):
        """Test invalid proxy string"""
        with pytest.raises(ValueError):
            construct_proxy_dict("invalid-proxy-format")

    def test_invalid_proxy_dict(self):
        """Test invalid proxy dictionary"""
        with pytest.raises(TypeError):
            construct_proxy_dict({"invalid": "structure"})


class TestFingerprintFunctions:
    """Test fingerprint generation functions"""

    def test_get_os_name(self):
        """Test OS name detection"""
        result = get_os_name()

        # Should return one of the known OS names or None
        valid_names = ["linux", "macos", "windows", "ios"]
        assert result is None or result in valid_names

    def test_generate_headers_basic(self):
        """Test basic header generation"""
        headers = generate_headers()

        assert isinstance(headers, dict)
        assert "User-Agent" in headers
        assert len(headers["User-Agent"]) > 0

    def test_generate_headers_browser_mode(self):
        """Test header generation in browser mode"""
        headers = generate_headers(browser_mode=True)

        assert isinstance(headers, dict)
        assert "User-Agent" in headers


class TestResponse:
    """Test Response class functionality"""

    def test_response_creation(self):
        """Test Response object creation"""
        response = Response(
            url="https://example.com",
            content="<html><body>Test</body></html>",
            status=200,
            reason="OK",
            cookies={"session": "abc123"},
            headers={"Content-Type": "text/html"},
            request_headers={"User-Agent": "Test"},
            encoding="utf-8",
        )

        assert response.url == "https://example.com"
        assert response.status == 200
        assert response.reason == "OK"
        assert response.cookies == {"session": "abc123"}

    def test_response_with_bytes_content(self):
        """Test Response with 'bytes' content"""
        content_bytes = "<html><body>Test</body></html>".encode("utf-8")

        response = Response(
            url="https://example.com",
            content=content_bytes,
            status=200,
            reason="OK",
            cookies={},
            headers={},
            request_headers={},
        )

        # Should handle 'bytes' content properly
        assert response.status == 200


class _MockRequest:
    """Minimal mock for Playwright's Request object."""

    def __init__(self, url: str, resource_type: str = "document"):
        self.url = url
        self.resource_type = resource_type


class _MockRoute:
    """Minimal mock for Playwright's sync Route object."""

    def __init__(self, url: str, resource_type: str = "document"):
        self.request = _MockRequest(url, resource_type)
        self.aborted = False
        self.continued = False

    def abort(self):
        self.aborted = True

    def continue_(self):
        self.continued = True


class _AsyncMockRoute:
    """Minimal mock for Playwright's async Route object."""

    def __init__(self, url: str, resource_type: str = "document"):
        self.request = _MockRequest(url, resource_type)
        self.aborted = False
        self.continued = False

    async def abort(self):
        self.aborted = True

    async def continue_(self):
        self.continued = True


class TestCreateInterceptHandler:
    """Test the unified sync route handler factory."""

    def test_blocks_disabled_resource_types(self):
        handler = create_intercept_handler(disable_resources=True)
        route = _MockRoute("https://example.com/image.png", resource_type="image")
        handler(route)
        assert route.aborted

    def test_continues_allowed_resource_types(self):
        handler = create_intercept_handler(disable_resources=True)
        route = _MockRoute("https://example.com/page", resource_type="document")
        handler(route)
        assert route.continued

    def test_blocks_exact_domain(self):
        handler = create_intercept_handler(disable_resources=False, blocked_domains={"ads.example.com"})
        route = _MockRoute("https://ads.example.com/tracker.js")
        handler(route)
        assert route.aborted

    def test_blocks_subdomain(self):
        handler = create_intercept_handler(disable_resources=False, blocked_domains={"example.com"})
        route = _MockRoute("https://sub.example.com/page")
        handler(route)
        assert route.aborted

    def test_continues_non_blocked_domain(self):
        handler = create_intercept_handler(disable_resources=False, blocked_domains={"ads.example.com"})
        route = _MockRoute("https://safe.example.com/page")
        handler(route)
        assert route.continued

    def test_resource_blocking_takes_priority_over_domain(self):
        """When both are active, resource type check comes first."""
        handler = create_intercept_handler(disable_resources=True, blocked_domains={"example.com"})
        route = _MockRoute("https://example.com/style.css", resource_type="stylesheet")
        handler(route)
        assert route.aborted

    def test_domain_blocking_with_resources_disabled(self):
        """Non-blocked resource type from a blocked domain should still be aborted."""
        handler = create_intercept_handler(disable_resources=True, blocked_domains={"tracker.io"})
        route = _MockRoute("https://tracker.io/api", resource_type="document")
        handler(route)
        assert route.aborted

    def test_no_blocking_continues(self):
        handler = create_intercept_handler(disable_resources=False)
        route = _MockRoute("https://example.com/page")
        handler(route)
        assert route.continued

    def test_does_not_block_partial_domain_match(self):
        """'example.com' should not block 'notexample.com'."""
        handler = create_intercept_handler(disable_resources=False, blocked_domains={"example.com"})
        route = _MockRoute("https://notexample.com/page")
        handler(route)
        assert route.continued

    def test_multiple_blocked_domains(self):
        handler = create_intercept_handler(disable_resources=False, blocked_domains={"ads.com", "tracker.io"})
        route_ads = _MockRoute("https://ads.com/banner")
        route_tracker = _MockRoute("https://cdn.tracker.io/script.js")
        route_safe = _MockRoute("https://example.com/page")
        handler(route_ads)
        handler(route_tracker)
        handler(route_safe)
        assert route_ads.aborted
        assert route_tracker.aborted
        assert route_safe.continued


class TestCreateAsyncInterceptHandler:
    """Test the unified async route handler factory."""

    @pytest.mark.asyncio
    async def test_blocks_disabled_resource_types(self):
        handler = create_async_intercept_handler(disable_resources=True)
        route = _AsyncMockRoute("https://example.com/font.woff", resource_type="font")
        await handler(route)
        assert route.aborted

    @pytest.mark.asyncio
    async def test_blocks_domain(self):
        handler = create_async_intercept_handler(disable_resources=False, blocked_domains={"ads.example.com"})
        route = _AsyncMockRoute("https://ads.example.com/track")
        await handler(route)
        assert route.aborted

    @pytest.mark.asyncio
    async def test_continues_non_blocked(self):
        handler = create_async_intercept_handler(disable_resources=False, blocked_domains={"ads.example.com"})
        route = _AsyncMockRoute("https://safe.example.com/page")
        await handler(route)
        assert route.continued

    @pytest.mark.asyncio
    async def test_blocks_subdomain(self):
        handler = create_async_intercept_handler(disable_resources=False, blocked_domains={"tracker.io"})
        route = _AsyncMockRoute("https://cdn.tracker.io/script.js")
        await handler(route)
        assert route.aborted

    @pytest.mark.asyncio
    async def test_does_not_block_partial_domain_match(self):
        handler = create_async_intercept_handler(disable_resources=False, blocked_domains={"example.com"})
        route = _AsyncMockRoute("https://notexample.com/page")
        await handler(route)
        assert route.continued


class TestIsDomainBlocked:
    """Test the frozenset-based domain matching helper."""

    def test_exact_match(self):
        domains = frozenset({"doubleclick.net"})
        assert _is_domain_blocked("doubleclick.net", domains) is True

    def test_subdomain_match(self):
        domains = frozenset({"doubleclick.net"})
        assert _is_domain_blocked("ads.doubleclick.net", domains) is True

    def test_deep_subdomain_match(self):
        domains = frozenset({"doubleclick.net"})
        assert _is_domain_blocked("tracker.ads.doubleclick.net", domains) is True

    def test_no_partial_match(self):
        domains = frozenset({"doubleclick.net"})
        assert _is_domain_blocked("notdoubleclick.net", domains) is False

    def test_no_match(self):
        domains = frozenset({"doubleclick.net"})
        assert _is_domain_blocked("example.com", domains) is False

    def test_empty_domains(self):
        assert _is_domain_blocked("example.com", frozenset()) is False

    def test_multiple_domains(self):
        domains = frozenset({"ads.com", "tracker.io", "doubleclick.net"})
        assert _is_domain_blocked("cdn.ads.com", domains) is True
        assert _is_domain_blocked("tracker.io", domains) is True
        assert _is_domain_blocked("safe.example.com", domains) is False


class TestAdDomains:
    """Test the built-in ad domain list."""

    def test_ad_domains_is_frozenset(self):
        from scrapling.engines.toolbelt.ad_domains import AD_DOMAINS

        assert isinstance(AD_DOMAINS, frozenset)

    def test_ad_domains_has_entries(self):
        from scrapling.engines.toolbelt.ad_domains import AD_DOMAINS

        assert len(AD_DOMAINS) > 1000

    def test_ad_domains_contains_known_entries(self):
        from scrapling.engines.toolbelt.ad_domains import AD_DOMAINS

        assert "doubleclick.net" in AD_DOMAINS
        assert "googlesyndication.com" in AD_DOMAINS


class TestBlockAdsConfig:
    """Test that block_ads merges ad domains into blocked_domains at config level."""

    def test_block_ads_populates_blocked_domains(self):
        from scrapling.engines._browsers._validators import PlaywrightConfig

        config = PlaywrightConfig(block_ads=True)
        assert config.blocked_domains is not None
        assert len(config.blocked_domains) > 1000
        assert "doubleclick.net" in config.blocked_domains

    def test_block_ads_false_leaves_blocked_domains_none(self):
        from scrapling.engines._browsers._validators import PlaywrightConfig

        config = PlaywrightConfig(block_ads=False)
        assert config.blocked_domains is None

    def test_block_ads_merges_with_user_domains(self):
        from scrapling.engines._browsers._validators import PlaywrightConfig

        user_domains = {"my-custom-block.com"}
        config = PlaywrightConfig(block_ads=True, blocked_domains=user_domains)
        assert config.blocked_domains is not None
        assert "my-custom-block.com" in config.blocked_domains
        assert "doubleclick.net" in config.blocked_domains

    def test_block_ads_does_not_modify_original_set(self):
        from scrapling.engines._browsers._validators import PlaywrightConfig

        user_domains = {"my-custom-block.com"}
        _ = PlaywrightConfig(block_ads=True, blocked_domains=user_domains)
        assert len(user_domains) == 1