179 lines
6.5 KiB
Python
179 lines
6.5 KiB
Python
from __future__ import annotations
|
|
|
|
from crawlee import Request
|
|
from crawlee.proxy_configuration import ProxyConfiguration
|
|
|
|
|
|
async def test_rotates_proxies_uniformly_with_no_request() -> None:
|
|
tiered_proxy_urls: list[list[str | None]] = [
|
|
['http://proxy:1111', 'http://proxy:2222'],
|
|
['http://proxy:3333', 'http://proxy:4444'],
|
|
]
|
|
|
|
config = ProxyConfiguration(tiered_proxy_urls=tiered_proxy_urls)
|
|
|
|
info = await config.new_proxy_info(None, None, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[0][0]
|
|
|
|
info = await config.new_proxy_info(None, None, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[0][1]
|
|
|
|
info = await config.new_proxy_info(None, None, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[1][0]
|
|
|
|
info = await config.new_proxy_info(None, None, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[1][1]
|
|
|
|
info = await config.new_proxy_info(None, None, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[0][0]
|
|
|
|
|
|
async def test_retrying_request_makes_tier_go_up() -> None:
|
|
tiered_proxy_urls: list[list[str | None]] = [
|
|
['http://proxy:1111'],
|
|
['http://proxy:2222'],
|
|
['http://proxy:3333'],
|
|
['http://proxy:4444'],
|
|
]
|
|
|
|
config = ProxyConfiguration(tiered_proxy_urls=tiered_proxy_urls)
|
|
|
|
# Calling `new_proxy_info` with the same request most probably means it's being retried
|
|
request_1 = Request(url='http://some.domain/abc', unique_key='1')
|
|
|
|
info = await config.new_proxy_info(None, request_1, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[0][0]
|
|
|
|
info = await config.new_proxy_info(None, request_1, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[1][0]
|
|
|
|
info = await config.new_proxy_info(None, request_1, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[2][0]
|
|
|
|
# Subsequent requests with the same domain should use the same tier
|
|
request_2 = Request(url='http://some.domain/xyz', unique_key='2')
|
|
|
|
info = await config.new_proxy_info(None, request_2, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[2][0]
|
|
|
|
|
|
async def test_retrying_request_makes_tier_go_up_with_sessions() -> None:
|
|
tiered_proxy_urls: list[list[str | None]] = [
|
|
['http://proxy:1111'],
|
|
['http://proxy:2222'],
|
|
['http://proxy:3333'],
|
|
['http://proxy:4444'],
|
|
]
|
|
|
|
config = ProxyConfiguration(tiered_proxy_urls=tiered_proxy_urls)
|
|
|
|
request = Request(url='http://some.domain/abc', unique_key='1')
|
|
|
|
# Calling `new_proxy_info` with the same request likely means that it is being retried.
|
|
# However, a single session should always receive the same proxy
|
|
info = await config.new_proxy_info('session_id', request, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[0][0]
|
|
|
|
info = await config.new_proxy_info('session_id', request, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[0][0]
|
|
|
|
info = await config.new_proxy_info('session_id', request, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[0][0]
|
|
|
|
# For a new session, we will get a proxy from the corresponding tier
|
|
info = await config.new_proxy_info('session_id2', request, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[3][0]
|
|
|
|
info = await config.new_proxy_info('session_id2', request, None)
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[3][0]
|
|
|
|
|
|
async def test_successful_request_makes_tier_go_down() -> None:
|
|
"""Repeatedly requesting a proxy for a single request will cause the proxy tier to go up -
|
|
ProxyConfiguration assumes those are retries. Then, requesting a proxy for different requests to the same domain
|
|
will cause the tier to drop back down."""
|
|
|
|
tiered_proxy_urls: list[list[str | None]] = [
|
|
['http://proxy:1111'],
|
|
['http://proxy:2222'],
|
|
['http://proxy:3333'],
|
|
['http://proxy:4444'],
|
|
]
|
|
|
|
config = ProxyConfiguration(tiered_proxy_urls=tiered_proxy_urls)
|
|
|
|
request_1 = Request(url='http://some.domain/abc', unique_key='1')
|
|
|
|
info = None
|
|
for tier in tiered_proxy_urls:
|
|
info = await config.new_proxy_info(None, request_1, None)
|
|
assert info is not None
|
|
assert info.url == tier[0]
|
|
|
|
for i in range(100):
|
|
new_request = Request(url=f'http://some.domain/{i}', unique_key=str(i))
|
|
info = await config.new_proxy_info(None, new_request, None)
|
|
|
|
assert info is not None
|
|
assert info.url == tiered_proxy_urls[0][0]
|
|
|
|
|
|
async def test_none_proxy_retrying_request_makes_tier_go_up() -> None:
|
|
tiered_proxy_urls: list[list[str | None]] = [
|
|
[None],
|
|
['http://proxy:1111'],
|
|
]
|
|
|
|
config = ProxyConfiguration(tiered_proxy_urls=tiered_proxy_urls)
|
|
|
|
# Calling `new_proxy_info` with the same request most probably means it's being retried
|
|
request_1 = Request(url='http://some.domain/abc', unique_key='1')
|
|
|
|
# No proxy used.
|
|
info = await config.new_proxy_info(None, request_1, None)
|
|
assert info is None, 'First entry in tired_proxy_urls is None. config.new_proxy_info is expected to generate None.'
|
|
|
|
# Proxy should go up one tier for same request that was already sent before.
|
|
info = await config.new_proxy_info(None, request_1, None)
|
|
assert info is not None, (
|
|
'config.new_proxy_info is expected to generate non-none proxy info from non-none tiered_proxy_urls.'
|
|
)
|
|
assert info.url == tiered_proxy_urls[1][0]
|
|
|
|
|
|
async def test_none_proxy_rotates_proxies_uniformly_with_no_request() -> None:
|
|
tiered_proxy_urls = [
|
|
[None, 'http://proxy:1111'],
|
|
]
|
|
|
|
config = ProxyConfiguration(tiered_proxy_urls=tiered_proxy_urls)
|
|
|
|
# No proxy used.
|
|
info = await config.new_proxy_info(None, None, None)
|
|
assert info is None, 'First entry in tired_proxy_urls is None. config.new_proxy_info is expected to generate None.'
|
|
|
|
# Proxy should be rotated on the same proxy tier for a new request.
|
|
info = await config.new_proxy_info(None, None, None)
|
|
assert info is not None, (
|
|
'config.new_proxy_info is expected to generate non-none proxy info from non-none tiered_proxy_urls.'
|
|
)
|
|
assert info.url == tiered_proxy_urls[0][1]
|
|
|
|
# Proxy rotation starts from the beginning of the proxy list after last proxy in tier was used. No proxy used again.
|
|
info = await config.new_proxy_info(None, None, None)
|
|
assert info is None, 'First entry in tired_proxy_urls is None. config.new_proxy_info is expected to generate None.'
|