wehub-resource-sync
247153575d
Tests / tests (map[TOXENV:py310], macos-latest, 3.10) (push) Has been cancelled
Tests / tests (map[TOXENV:py311], macos-latest, 3.11) (push) Has been cancelled
Tests / tests (map[TOXENV:py312], macos-latest, 3.12) (push) Has been cancelled
Tests / tests (map[TOXENV:py313], macos-latest, 3.13) (push) Has been cancelled
304 lines
11 KiB
Python
304 lines
11 KiB
Python
import pytest
|
|
import random
|
|
from threading import Thread
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from scrapling.engines.toolbelt import ProxyRotator, is_proxy_error, cyclic_rotation
|
|
|
|
|
|
class TestCyclicRotationStrategy:
|
|
"""Test the default cyclic_rotation strategy function"""
|
|
|
|
def test_cyclic_rotation_cycles_through_proxies(self):
|
|
"""Test that cyclic_rotation returns proxies in order"""
|
|
proxies = ["http://p1:8080", "http://p2:8080", "http://p3:8080"]
|
|
|
|
proxy, next_idx = cyclic_rotation(proxies, 0)
|
|
assert proxy == "http://p1:8080"
|
|
assert next_idx == 1
|
|
|
|
proxy, next_idx = cyclic_rotation(proxies, 1)
|
|
assert proxy == "http://p2:8080"
|
|
assert next_idx == 2
|
|
|
|
proxy, next_idx = cyclic_rotation(proxies, 2)
|
|
assert proxy == "http://p3:8080"
|
|
assert next_idx == 0 # Wraps around
|
|
|
|
def test_cyclic_rotation_wraps_index(self):
|
|
"""Test that cyclic_rotation handles index overflow"""
|
|
proxies = ["http://p1:8080", "http://p2:8080"]
|
|
|
|
# Index larger than list length should wrap
|
|
proxy, next_idx = cyclic_rotation(proxies, 5)
|
|
assert proxy == "http://p2:8080" # 5 % 2 = 1
|
|
assert next_idx == 0
|
|
|
|
|
|
class TestProxyRotatorCreation:
|
|
"""Test ProxyRotator initialization and validation"""
|
|
|
|
def test_create_with_string_proxies(self):
|
|
"""Test creating rotator with string proxy URLs"""
|
|
proxies = ["http://p1:8080", "http://p2:8080"]
|
|
rotator = ProxyRotator(proxies)
|
|
|
|
assert len(rotator) == 2
|
|
assert rotator.proxies == proxies
|
|
|
|
def test_create_with_dict_proxies(self):
|
|
"""Test creating rotator with dict proxies"""
|
|
proxies = [
|
|
{"server": "http://p1:8080", "username": "user1", "password": "pass1"},
|
|
{"server": "http://p2:8080"},
|
|
]
|
|
rotator = ProxyRotator(proxies)
|
|
|
|
assert len(rotator) == 2
|
|
assert rotator.proxies == proxies
|
|
|
|
def test_create_with_mixed_proxies(self):
|
|
"""Test creating rotator with mixed string and dict proxies"""
|
|
proxies = [
|
|
"http://p1:8080",
|
|
{"server": "http://p2:8080", "username": "user"},
|
|
]
|
|
rotator = ProxyRotator(proxies)
|
|
|
|
assert len(rotator) == 2
|
|
|
|
def test_empty_proxies_raises_error(self):
|
|
"""Test that empty proxy list raises ValueError"""
|
|
with pytest.raises(ValueError, match="At least one proxy must be provided"):
|
|
ProxyRotator([])
|
|
|
|
def test_dict_without_server_raises_error(self):
|
|
"""Test that dict proxy without 'server' key raises ValueError"""
|
|
with pytest.raises(ValueError, match="Proxy dict must have a 'server' key"):
|
|
ProxyRotator([{"username": "user", "password": "pass"}])
|
|
|
|
def test_invalid_proxy_type_raises_error(self):
|
|
"""Test that invalid proxy type raises TypeError"""
|
|
with pytest.raises(TypeError, match="Invalid proxy type"):
|
|
ProxyRotator([123])
|
|
|
|
with pytest.raises(TypeError, match="Invalid proxy type"):
|
|
ProxyRotator([None])
|
|
|
|
def test_non_callable_strategy_raises_error(self):
|
|
"""Test that non-callable strategy raises TypeError"""
|
|
with pytest.raises(TypeError, match="strategy must be callable"):
|
|
ProxyRotator(["http://p1:8080"], strategy="cyclic_rotation")
|
|
|
|
with pytest.raises(TypeError, match="strategy must be callable"):
|
|
ProxyRotator(["http://p1:8080"], strategy=123)
|
|
|
|
|
|
class TestProxyRotatorRotation:
|
|
"""Test ProxyRotator rotation behavior"""
|
|
|
|
def test_get_proxy_cyclic_rotation(self):
|
|
"""Test that get_proxy cycles through proxies in order"""
|
|
proxies = ["http://p1:8080", "http://p2:8080", "http://p3:8080"]
|
|
rotator = ProxyRotator(proxies)
|
|
|
|
# First cycle
|
|
assert rotator.get_proxy() == "http://p1:8080"
|
|
assert rotator.get_proxy() == "http://p2:8080"
|
|
assert rotator.get_proxy() == "http://p3:8080"
|
|
|
|
# Second cycle - wraps around
|
|
assert rotator.get_proxy() == "http://p1:8080"
|
|
assert rotator.get_proxy() == "http://p2:8080"
|
|
assert rotator.get_proxy() == "http://p3:8080"
|
|
|
|
def test_get_proxy_single_proxy(self):
|
|
"""Test rotation with single proxy always returns the same proxy"""
|
|
rotator = ProxyRotator(["http://only:8080"])
|
|
|
|
for _ in range(5):
|
|
assert rotator.get_proxy() == "http://only:8080"
|
|
|
|
def test_get_proxy_with_dict_proxies(self):
|
|
"""Test rotation with dict proxies"""
|
|
proxies = [
|
|
{"server": "http://p1:8080"},
|
|
{"server": "http://p2:8080"},
|
|
]
|
|
rotator = ProxyRotator(proxies)
|
|
|
|
assert rotator.get_proxy() == {"server": "http://p1:8080"}
|
|
assert rotator.get_proxy() == {"server": "http://p2:8080"}
|
|
assert rotator.get_proxy() == {"server": "http://p1:8080"}
|
|
|
|
|
|
class TestCustomStrategies:
|
|
"""Test ProxyRotator with custom rotation strategies"""
|
|
|
|
def test_random_strategy(self):
|
|
"""Test custom random selection strategy"""
|
|
def random_strategy(proxies, idx):
|
|
return random.choice(proxies), idx
|
|
|
|
proxies = ["http://p1:8080", "http://p2:8080", "http://p3:8080"]
|
|
rotator = ProxyRotator(proxies, strategy=random_strategy)
|
|
|
|
# Get multiple proxies - they should all be valid
|
|
results = [rotator.get_proxy() for _ in range(10)]
|
|
assert all(p in proxies for p in results)
|
|
|
|
def test_sticky_strategy(self):
|
|
"""Test custom sticky strategy that always returns first proxy"""
|
|
def sticky_strategy(proxies, idx):
|
|
return proxies[0], idx
|
|
|
|
rotator = ProxyRotator(
|
|
["http://p1:8080", "http://p2:8080"],
|
|
strategy=sticky_strategy
|
|
)
|
|
|
|
for _ in range(5):
|
|
assert rotator.get_proxy() == "http://p1:8080"
|
|
|
|
def test_weighted_strategy(self):
|
|
"""Test custom weighted strategy"""
|
|
call_count = {"count": 0}
|
|
|
|
def alternating_strategy(proxies, idx):
|
|
# Returns first proxy twice, then second proxy once
|
|
call_count["count"] += 1
|
|
if call_count["count"] % 3 == 0:
|
|
return proxies[1], idx
|
|
return proxies[0], idx
|
|
|
|
rotator = ProxyRotator(
|
|
["http://primary:8080", "http://backup:8080"],
|
|
strategy=alternating_strategy
|
|
)
|
|
|
|
assert rotator.get_proxy() == "http://primary:8080"
|
|
assert rotator.get_proxy() == "http://primary:8080"
|
|
assert rotator.get_proxy() == "http://backup:8080"
|
|
|
|
def test_lambda_strategy(self):
|
|
"""Test using lambda as strategy"""
|
|
rotator = ProxyRotator(
|
|
["http://p1:8080", "http://p2:8080"],
|
|
strategy=lambda proxies, idx: (proxies[-1], idx) # Always last
|
|
)
|
|
|
|
assert rotator.get_proxy() == "http://p2:8080"
|
|
assert rotator.get_proxy() == "http://p2:8080"
|
|
|
|
|
|
class TestProxyRotatorProperties:
|
|
"""Test ProxyRotator properties and methods"""
|
|
|
|
def test_proxies_property_returns_copy(self):
|
|
"""Test that proxies property returns a copy, not the original list"""
|
|
original = ["http://p1:8080", "http://p2:8080"]
|
|
rotator = ProxyRotator(original)
|
|
|
|
proxies_copy = rotator.proxies
|
|
proxies_copy.append("http://p3:8080")
|
|
|
|
# Original should be unchanged
|
|
assert len(rotator) == 2
|
|
assert len(rotator.proxies) == 2
|
|
|
|
def test_len_returns_proxy_count(self):
|
|
"""Test __len__ returns correct count"""
|
|
assert len(ProxyRotator(["http://p1:8080"])) == 1
|
|
assert len(ProxyRotator(["http://p1:8080", "http://p2:8080"])) == 2
|
|
assert len(ProxyRotator(["a", "b", "c", "d", "e"])) == 5
|
|
|
|
def test_repr(self):
|
|
"""Test __repr__ format"""
|
|
rotator = ProxyRotator(["http://p1:8080", "http://p2:8080", "http://p3:8080"])
|
|
assert repr(rotator) == "ProxyRotator(proxies=3)"
|
|
|
|
|
|
class TestProxyRotatorThreadSafety:
|
|
"""Test ProxyRotator thread safety"""
|
|
|
|
def test_concurrent_get_proxy(self):
|
|
"""Test that concurrent get_proxy calls don't cause errors"""
|
|
proxies = [f"http://p{i}:8080" for i in range(10)]
|
|
rotator = ProxyRotator(proxies)
|
|
results = []
|
|
|
|
def get_proxies(n):
|
|
for _ in range(n):
|
|
results.append(rotator.get_proxy())
|
|
|
|
threads = [Thread(target=get_proxies, args=(100,)) for _ in range(10)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# All results should be valid proxies
|
|
assert len(results) == 1000
|
|
assert all(p in proxies for p in results)
|
|
|
|
def test_thread_pool_concurrent_access(self):
|
|
"""Test concurrent access using ThreadPoolExecutor"""
|
|
proxies = ["http://p1:8080", "http://p2:8080", "http://p3:8080"]
|
|
rotator = ProxyRotator(proxies)
|
|
|
|
with ThreadPoolExecutor(max_workers=5) as executor:
|
|
futures = [executor.submit(rotator.get_proxy) for _ in range(100)]
|
|
results = [f.result() for f in futures]
|
|
|
|
assert len(results) == 100
|
|
assert all(p in proxies for p in results)
|
|
|
|
|
|
class TestIsProxyError:
|
|
"""Test is_proxy_error utility function"""
|
|
|
|
@pytest.mark.parametrize("error_msg", [
|
|
"net::err_proxy_connection_failed",
|
|
"NET::ERR_PROXY_AUTH_FAILED",
|
|
"net::err_tunnel_connection_failed",
|
|
"Connection refused by proxy",
|
|
"Connection reset by peer",
|
|
"Connection timed out while connecting to proxy",
|
|
"Failed to connect to proxy server",
|
|
"Could not resolve proxy host",
|
|
])
|
|
def test_proxy_errors_detected(self, error_msg):
|
|
"""Test that proxy-related errors are detected"""
|
|
assert is_proxy_error(Exception(error_msg)) is True
|
|
|
|
@pytest.mark.parametrize("error_msg", [
|
|
"Page not found",
|
|
"404 Not Found",
|
|
"Internal server error",
|
|
"DNS resolution failed",
|
|
"SSL certificate error",
|
|
"Timeout waiting for response",
|
|
"Invalid JSON response",
|
|
])
|
|
def test_non_proxy_errors_not_detected(self, error_msg):
|
|
"""Test that non-proxy errors are not detected as proxy errors"""
|
|
assert is_proxy_error(Exception(error_msg)) is False
|
|
|
|
def test_case_insensitive_detection(self):
|
|
"""Test that error detection is case-insensitive"""
|
|
assert is_proxy_error(Exception("NET::ERR_PROXY")) is True
|
|
assert is_proxy_error(Exception("Net::Err_Proxy")) is True
|
|
assert is_proxy_error(Exception("CONNECTION REFUSED")) is True
|
|
|
|
def test_empty_error_message(self):
|
|
"""Test handling of empty error message"""
|
|
assert is_proxy_error(Exception("")) is False
|
|
|
|
def test_custom_exception_types(self):
|
|
"""Test with custom exception types"""
|
|
class CustomError(Exception):
|
|
pass
|
|
|
|
assert is_proxy_error(CustomError("net::err_proxy_failed")) is True
|
|
assert is_proxy_error(CustomError("normal error")) is False
|