Karim shoair
303 lines
10 KiB
Python
303 lines
10 KiB
Python
import pytest
|
|
from pathlib import Path
|
|
|
|
from scrapling.engines.toolbelt.custom import StatusText, Response
|
|
from scrapling.engines.toolbelt.navigation import (
|
|
construct_proxy_dict,
|
|
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_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 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="<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
|