import pytest 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" 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="Test", 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 = "Test".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