import pytest from pathlib import Path from scrapling.engines.toolbelt.custom import ResponseEncoding, StatusText, Response from scrapling.engines.toolbelt.navigation import ( construct_proxy_dict, construct_cdp_url, js_bypass_path ) from scrapling.engines.toolbelt.fingerprints import ( generate_convincing_referer, 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_content_type(content_type_map): """Test if parsing different types of 'content-type' returns the expected result""" for header_value, expected_encoding in content_type_map.items(): assert ResponseEncoding.get_value(header_value) == expected_encoding 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_proxy_as_tuple(self): """Test returning proxy as a tuple""" result = construct_proxy_dict("http://proxy.example.com:8080", as_tuple=True) assert isinstance(result, tuple) result_dict = dict(result) assert result_dict["server"] == "http://proxy.example.com:8080" 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 TestConstructCdpUrl: """Test CDP URL construction""" def test_basic_cdp_url(self): """Test basic CDP URL""" result = construct_cdp_url("ws://localhost:9222/devtools/browser") assert result == "ws://localhost:9222/devtools/browser" def test_cdp_url_with_params(self): """Test CDP URL with query parameters""" params = {"timeout": "30000", "headless": "true"} result = construct_cdp_url("ws://localhost:9222/devtools/browser", params) assert "timeout=30000" in result assert "headless=true" in result def test_cdp_url_without_leading_slash(self): """Test CDP URL without a leading slash in the path""" with pytest.raises(ValueError): construct_cdp_url("ws://localhost:9222devtools/browser") def test_invalid_cdp_scheme(self): """Test invalid CDP URL scheme""" with pytest.raises(ValueError): construct_cdp_url("http://localhost:9222/devtools/browser") def test_invalid_cdp_netloc(self): """Test invalid CDP URL network location""" with pytest.raises(ValueError): construct_cdp_url("ws:///devtools/browser") def test_malformed_cdp_url(self): """Test malformed CDP URL""" with pytest.raises(ValueError): construct_cdp_url("not-a-url") class TestJsBypassPath: """Test JavaScript bypass path utility""" def test_js_bypass_path(self): """Test getting JavaScript bypass file path""" result = js_bypass_path("webdriver_fully.js") assert isinstance(result, str) assert result.endswith("webdriver_fully.js") assert Path(result).exists() def test_js_bypass_path_caching(self): """Test that js_bypass_path is cached""" result1 = js_bypass_path("webdriver_fully.js") result2 = js_bypass_path("webdriver_fully.js") assert result1 == result2 class TestFingerprintFunctions: """Test fingerprint generation functions""" def test_generate_convincing_referer(self): """Test referer generation""" url = "https://sub.example.com/page.html" result = generate_convincing_referer(url) assert result.startswith("https://www.google.com/search?q=") assert "example" in result def test_generate_convincing_referer_caching(self): """Test referer generation caching""" url = "https://example.com" result1 = generate_convincing_referer(url) result2 = generate_convincing_referer(url) assert result1 == result2 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