e78fc8dc3c
Replace _http_check (HTTP GET to httpbin.org) with _tls_check that performs a TLS handshake through the proxy chain. Multiple targets (google, cloudflare, amazon) rotated round-robin eliminate the single point of failure. Lighter, faster, harder to block than HTTP. - Add test_targets config field (replaces test_url) - Backward compat: legacy test_url extracts hostname automatically - Add ssl.create_default_context() and round-robin index to ProxyPool - Update docs (example.yaml, USAGE.md, CHEATSHEET.md)
589 lines
21 KiB
Python
589 lines
21 KiB
Python
"""Tests for the managed proxy pool."""
|
|
|
|
import asyncio
|
|
import ssl
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from s5p.config import ChainHop, PoolSourceConfig, ProxyPoolConfig
|
|
from s5p.pool import ProxyEntry, ProxyPool
|
|
|
|
|
|
class TestProxyEntry:
|
|
"""Test ProxyEntry defaults."""
|
|
|
|
def test_defaults(self):
|
|
hop = ChainHop(proto="socks5", host="1.2.3.4", port=1080)
|
|
entry = ProxyEntry(hop=hop)
|
|
assert entry.alive is False
|
|
assert entry.fails == 0
|
|
assert entry.tests == 0
|
|
|
|
|
|
class TestProxyPoolMerge:
|
|
"""Test proxy deduplication and merge."""
|
|
|
|
def test_merge_dedup(self):
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
proxies = [
|
|
ChainHop(proto="socks5", host="1.2.3.4", port=1080),
|
|
ChainHop(proto="socks5", host="1.2.3.4", port=1080),
|
|
ChainHop(proto="socks5", host="5.6.7.8", port=1080),
|
|
]
|
|
pool._merge(proxies)
|
|
assert pool.count == 2
|
|
|
|
def test_merge_updates_existing(self):
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
hop = ChainHop(proto="socks5", host="1.2.3.4", port=1080)
|
|
pool._merge([hop])
|
|
first_seen = pool._proxies["socks5://1.2.3.4:1080"].last_seen
|
|
|
|
# merge again -- last_seen should update
|
|
time.sleep(0.01)
|
|
pool._merge([hop])
|
|
assert pool._proxies["socks5://1.2.3.4:1080"].last_seen >= first_seen
|
|
assert pool.count == 1
|
|
|
|
|
|
class TestProxyPoolGet:
|
|
"""Test proxy selection."""
|
|
|
|
def test_get_empty(self):
|
|
import asyncio
|
|
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
result = asyncio.run(pool.get())
|
|
assert result is None
|
|
|
|
def test_get_returns_alive(self):
|
|
import asyncio
|
|
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
hop = ChainHop(proto="socks5", host="1.2.3.4", port=1080)
|
|
pool._proxies["socks5://1.2.3.4:1080"] = ProxyEntry(hop=hop, alive=True)
|
|
pool._rebuild_alive()
|
|
result = asyncio.run(pool.get())
|
|
assert result is not None
|
|
assert result.host == "1.2.3.4"
|
|
|
|
|
|
class TestProxyPoolWeight:
|
|
"""Test weighted proxy selection."""
|
|
|
|
def test_recent_proxy_preferred(self):
|
|
import asyncio
|
|
from collections import Counter
|
|
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
now = time.time()
|
|
fresh = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
stale = ChainHop(proto="socks5", host="10.0.0.2", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=fresh, alive=True, last_ok=now,
|
|
)
|
|
pool._proxies["socks5://10.0.0.2:1080"] = ProxyEntry(
|
|
hop=stale, alive=True, last_ok=now - 3600,
|
|
)
|
|
pool._rebuild_alive()
|
|
|
|
counts: Counter[str] = Counter()
|
|
for _ in range(1000):
|
|
hop = asyncio.run(pool.get())
|
|
counts[hop.host] += 1
|
|
|
|
assert counts["10.0.0.1"] > counts["10.0.0.2"] * 3
|
|
|
|
def test_weight_values(self):
|
|
hop = ChainHop(proto="socks5", host="1.2.3.4", port=1080)
|
|
now = 1000.0
|
|
|
|
# just tested
|
|
entry = ProxyEntry(hop=hop, last_ok=now)
|
|
assert ProxyPool._weight(entry, now) == pytest.approx(1.0)
|
|
|
|
# 5 minutes ago
|
|
entry.last_ok = now - 300
|
|
assert ProxyPool._weight(entry, now) == pytest.approx(0.5)
|
|
|
|
# never tested
|
|
entry.last_ok = 0
|
|
assert ProxyPool._weight(entry, now) == pytest.approx(0.01)
|
|
|
|
def test_weight_failure_penalty(self):
|
|
hop = ChainHop(proto="socks5", host="1.2.3.4", port=1080)
|
|
now = 1000.0
|
|
|
|
# healthy proxy: weight ~1.0
|
|
entry = ProxyEntry(hop=hop, last_ok=now)
|
|
assert ProxyPool._weight(entry, now) == pytest.approx(1.0)
|
|
|
|
# just failed: weight drops to floor (fail_age=0 -> multiplier=0)
|
|
entry.last_fail = now
|
|
assert ProxyPool._weight(entry, now) == pytest.approx(0.01)
|
|
|
|
# 30s after failure: base=1/(1+30/300)=0.909, penalty=30/60=0.5 -> ~0.45
|
|
assert ProxyPool._weight(entry, now + 30) == pytest.approx(0.4545, abs=0.05)
|
|
|
|
# 60s after failure: penalty=1.0, only base decay -> 1/(1+60/300)=0.833
|
|
assert ProxyPool._weight(entry, now + 60) == pytest.approx(0.833, abs=0.05)
|
|
|
|
def test_report_failure(self):
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
hop = ChainHop(proto="socks5", host="1.2.3.4", port=1080)
|
|
pool._proxies["socks5://1.2.3.4:1080"] = ProxyEntry(
|
|
hop=hop, alive=True, last_ok=time.time(),
|
|
)
|
|
|
|
assert pool._proxies["socks5://1.2.3.4:1080"].last_fail == 0.0
|
|
pool.report_failure(hop)
|
|
assert pool._proxies["socks5://1.2.3.4:1080"].last_fail > 0.0
|
|
|
|
def test_report_failure_unknown_proxy(self):
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
hop = ChainHop(proto="socks5", host="9.9.9.9", port=1080)
|
|
pool.report_failure(hop) # should not raise
|
|
|
|
|
|
class TestProxyPoolHealthTests:
|
|
"""Test selective health testing."""
|
|
|
|
def test_selective_keys(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
now = time.time()
|
|
hop_a = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
hop_b = ChainHop(proto="socks5", host="10.0.0.2", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=hop_a, alive=False, last_seen=now,
|
|
)
|
|
pool._proxies["socks5://10.0.0.2:1080"] = ProxyEntry(
|
|
hop=hop_b, alive=False, last_seen=now,
|
|
)
|
|
|
|
# only test proxy A
|
|
with patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=True) as mock:
|
|
asyncio.run(pool._run_health_tests(keys=["socks5://10.0.0.1:1080"]))
|
|
# should only have been called for proxy A
|
|
assert mock.call_count == 1
|
|
assert mock.call_args[0][0] == "socks5://10.0.0.1:1080"
|
|
|
|
assert pool._proxies["socks5://10.0.0.1:1080"].alive is True
|
|
# proxy B untouched
|
|
assert pool._proxies["socks5://10.0.0.2:1080"].alive is False
|
|
|
|
def test_chain_check_skips_on_failure(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
chain_hop = ChainHop(proto="socks5", host="127.0.0.1", port=9050)
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [chain_hop], timeout=10.0)
|
|
|
|
now = time.time()
|
|
hop = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=hop, alive=True, last_seen=now, last_ok=now,
|
|
)
|
|
pool._rebuild_alive()
|
|
|
|
# chain test fails -> proxy tests should be skipped
|
|
with (
|
|
patch.object(pool, "_test_chain", new_callable=AsyncMock, return_value=False),
|
|
patch.object(pool, "_test_proxy", new_callable=AsyncMock) as mock_proxy,
|
|
):
|
|
asyncio.run(pool._run_health_tests())
|
|
mock_proxy.assert_not_called()
|
|
|
|
# proxy should remain in its previous state (untouched)
|
|
assert pool._proxies["socks5://10.0.0.1:1080"].alive is True
|
|
|
|
def test_chain_check_passes(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
chain_hop = ChainHop(proto="socks5", host="127.0.0.1", port=9050)
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [chain_hop], timeout=10.0)
|
|
|
|
now = time.time()
|
|
hop = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=hop, alive=False, last_seen=now,
|
|
)
|
|
|
|
# chain test passes -> proxy tests should run
|
|
with (
|
|
patch.object(pool, "_test_chain", new_callable=AsyncMock, return_value=True),
|
|
patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=True),
|
|
):
|
|
asyncio.run(pool._run_health_tests())
|
|
|
|
assert pool._proxies["socks5://10.0.0.1:1080"].alive is True
|
|
|
|
def test_no_chain_skips_check(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0) # no static chain
|
|
|
|
now = time.time()
|
|
hop = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=hop, alive=False, last_seen=now,
|
|
)
|
|
|
|
# no chain -> _test_chain should not be called, proxy tests run
|
|
with (
|
|
patch.object(pool, "_test_chain", new_callable=AsyncMock) as mock_chain,
|
|
patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=True),
|
|
):
|
|
asyncio.run(pool._run_health_tests())
|
|
mock_chain.assert_not_called()
|
|
|
|
assert pool._proxies["socks5://10.0.0.1:1080"].alive is True
|
|
|
|
|
|
class TestProxyPoolReport:
|
|
"""Test dead proxy reporting."""
|
|
|
|
def test_report_called_on_eviction(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
cfg = ProxyPoolConfig(sources=[], report_url="http://api:8081/report", max_fails=1)
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
now = time.time()
|
|
hop = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=hop, alive=False, last_seen=now, fails=0,
|
|
)
|
|
|
|
with (
|
|
patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=False),
|
|
patch.object(pool, "_report_dead", new_callable=AsyncMock) as mock_report,
|
|
):
|
|
asyncio.run(pool._run_health_tests())
|
|
# proxy should be evicted (fails=1 >= max_fails=1)
|
|
assert "socks5://10.0.0.1:1080" not in pool._proxies
|
|
mock_report.assert_called_once()
|
|
keys = mock_report.call_args[0][0]
|
|
assert "socks5://10.0.0.1:1080" in keys
|
|
|
|
def test_report_not_called_without_url(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
cfg = ProxyPoolConfig(sources=[], max_fails=1) # no report_url
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
now = time.time()
|
|
hop = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=hop, alive=False, last_seen=now, fails=0,
|
|
)
|
|
|
|
with (
|
|
patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=False),
|
|
patch.object(pool, "_report_dead", new_callable=AsyncMock) as mock_report,
|
|
):
|
|
asyncio.run(pool._run_health_tests())
|
|
mock_report.assert_not_called()
|
|
|
|
def test_report_async_payload(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
cfg = ProxyPoolConfig(sources=[], report_url="http://api:8081/report")
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
with patch("s5p.pool.http_post_json", new_callable=AsyncMock) as mock_post:
|
|
asyncio.run(pool._report_dead(["socks5://10.0.0.1:1080"]))
|
|
mock_post.assert_called_once()
|
|
url = mock_post.call_args[0][0]
|
|
payload = mock_post.call_args[0][1]
|
|
assert url == "http://api:8081/report"
|
|
assert payload == {"dead": [{"proto": "socks5", "proxy": "10.0.0.1:1080"}]}
|
|
|
|
|
|
class TestProxyPoolStaleExpiry:
|
|
"""Test stale proxy eviction."""
|
|
|
|
def test_stale_dead_proxy_evicted(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
cfg = ProxyPoolConfig(sources=[], refresh=300)
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
now = time.time()
|
|
# stale + dead: should be evicted
|
|
stale = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=stale, alive=False, last_seen=now - 1200,
|
|
)
|
|
# fresh + dead: should survive (recently seen, might recover)
|
|
fresh_dead = ChainHop(proto="socks5", host="10.0.0.2", port=1080)
|
|
pool._proxies["socks5://10.0.0.2:1080"] = ProxyEntry(
|
|
hop=fresh_dead, alive=False, last_seen=now,
|
|
)
|
|
|
|
with patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=False):
|
|
asyncio.run(pool._run_health_tests())
|
|
|
|
assert "socks5://10.0.0.1:1080" not in pool._proxies
|
|
assert "socks5://10.0.0.2:1080" in pool._proxies
|
|
|
|
def test_stale_alive_proxy_kept(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
cfg = ProxyPoolConfig(sources=[], refresh=300)
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
now = time.time()
|
|
# stale but alive: should survive (still passing health tests)
|
|
hop = ChainHop(proto="socks5", host="10.0.0.1", port=1080)
|
|
pool._proxies["socks5://10.0.0.1:1080"] = ProxyEntry(
|
|
hop=hop, alive=True, last_seen=now - 1200, last_ok=now,
|
|
)
|
|
|
|
with patch.object(pool, "_test_proxy", new_callable=AsyncMock, return_value=True):
|
|
asyncio.run(pool._run_health_tests())
|
|
|
|
assert "socks5://10.0.0.1:1080" in pool._proxies
|
|
|
|
|
|
class TestProxyPoolFetchFile:
|
|
"""Test file source parsing."""
|
|
|
|
def test_parse_file(self, tmp_path):
|
|
proxy_file = tmp_path / "proxies.txt"
|
|
proxy_file.write_text(
|
|
"# comment\n"
|
|
"socks5://1.2.3.4:1080\n"
|
|
"socks5://user:pass@5.6.7.8:1080\n"
|
|
"http://proxy.example.com:8080\n"
|
|
"\n"
|
|
" # another comment\n"
|
|
)
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
src = PoolSourceConfig(file=str(proxy_file))
|
|
result = pool._fetch_file_sync(src)
|
|
assert len(result) == 3
|
|
assert result[0].proto == "socks5"
|
|
assert result[0].host == "1.2.3.4"
|
|
assert result[1].username == "user"
|
|
assert result[1].password == "pass"
|
|
assert result[2].proto == "http"
|
|
|
|
def test_parse_file_with_proto_filter(self, tmp_path):
|
|
proxy_file = tmp_path / "proxies.txt"
|
|
proxy_file.write_text(
|
|
"socks5://1.2.3.4:1080\n"
|
|
"http://proxy.example.com:8080\n"
|
|
)
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
src = PoolSourceConfig(file=str(proxy_file), proto="socks5")
|
|
result = pool._fetch_file_sync(src)
|
|
assert len(result) == 1
|
|
assert result[0].proto == "socks5"
|
|
|
|
def test_missing_file(self, tmp_path):
|
|
cfg = ProxyPoolConfig(sources=[])
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
src = PoolSourceConfig(file=str(tmp_path / "nonexistent.txt"))
|
|
result = pool._fetch_file_sync(src)
|
|
assert result == []
|
|
|
|
|
|
class TestProxyPoolPersistence:
|
|
"""Test state save/load."""
|
|
|
|
def test_save_and_load(self, tmp_path):
|
|
state_file = str(tmp_path / "pool.json")
|
|
cfg = ProxyPoolConfig(sources=[], state_file=state_file)
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
hop = ChainHop(proto="socks5", host="1.2.3.4", port=1080)
|
|
pool._proxies["socks5://1.2.3.4:1080"] = ProxyEntry(
|
|
hop=hop, alive=True, fails=0, tests=5, last_ok=1000.0,
|
|
)
|
|
pool._rebuild_alive()
|
|
pool._save_state()
|
|
|
|
# load into a fresh pool
|
|
pool2 = ProxyPool(cfg, [], timeout=10.0)
|
|
pool2._load_state()
|
|
assert pool2.count == 1
|
|
assert pool2.alive_count == 1
|
|
entry = pool2._proxies["socks5://1.2.3.4:1080"]
|
|
assert entry.hop.host == "1.2.3.4"
|
|
assert entry.tests == 5
|
|
assert entry.alive is True
|
|
|
|
def test_corrupt_state(self, tmp_path):
|
|
state_file = tmp_path / "pool.json"
|
|
state_file.write_text("{invalid json")
|
|
cfg = ProxyPoolConfig(sources=[], state_file=str(state_file))
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
pool._load_state()
|
|
assert pool.count == 0
|
|
|
|
def test_missing_state(self, tmp_path):
|
|
cfg = ProxyPoolConfig(sources=[], state_file=str(tmp_path / "missing.json"))
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
pool._load_state() # should not raise
|
|
assert pool.count == 0
|
|
|
|
def test_state_with_auth(self, tmp_path):
|
|
state_file = str(tmp_path / "pool.json")
|
|
cfg = ProxyPoolConfig(sources=[], state_file=state_file)
|
|
pool = ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
hop = ChainHop(
|
|
proto="socks5", host="1.2.3.4", port=1080,
|
|
username="user", password="pass",
|
|
)
|
|
pool._proxies["socks5://1.2.3.4:1080"] = ProxyEntry(hop=hop, alive=True)
|
|
pool._save_state()
|
|
|
|
pool2 = ProxyPool(cfg, [], timeout=10.0)
|
|
pool2._load_state()
|
|
entry = pool2._proxies["socks5://1.2.3.4:1080"]
|
|
assert entry.hop.username == "user"
|
|
assert entry.hop.password == "pass"
|
|
|
|
|
|
class TestTlsCheck:
|
|
"""Test TLS handshake health check."""
|
|
|
|
def _make_pool(self, **kwargs):
|
|
cfg = ProxyPoolConfig(sources=[], **kwargs)
|
|
return ProxyPool(cfg, [], timeout=10.0)
|
|
|
|
def test_success(self):
|
|
pool = self._make_pool(test_targets=["www.example.com"])
|
|
|
|
mock_writer = MagicMock()
|
|
mock_writer.is_closing.return_value = False
|
|
mock_transport = MagicMock()
|
|
mock_protocol = MagicMock()
|
|
mock_transport.get_protocol.return_value = mock_protocol
|
|
mock_writer.transport = mock_transport
|
|
|
|
new_transport = MagicMock()
|
|
|
|
chain_ret = (MagicMock(), mock_writer)
|
|
with (
|
|
patch("s5p.pool.build_chain", new_callable=AsyncMock, return_value=chain_ret),
|
|
patch("asyncio.get_running_loop") as mock_loop_fn,
|
|
):
|
|
mock_loop = MagicMock()
|
|
mock_loop.start_tls = AsyncMock(return_value=new_transport)
|
|
mock_loop_fn.return_value = mock_loop
|
|
|
|
result = asyncio.run(pool._tls_check([]))
|
|
|
|
assert result is True
|
|
mock_loop.start_tls.assert_called_once_with(
|
|
mock_transport, mock_protocol, pool._ssl_ctx,
|
|
server_hostname="www.example.com",
|
|
)
|
|
new_transport.close.assert_called_once()
|
|
|
|
def test_build_chain_failure(self):
|
|
pool = self._make_pool(test_targets=["www.example.com"])
|
|
|
|
with patch(
|
|
"s5p.pool.build_chain", new_callable=AsyncMock,
|
|
side_effect=ConnectionError("refused"),
|
|
):
|
|
result = asyncio.run(pool._tls_check([]))
|
|
|
|
assert result is False
|
|
|
|
def test_handshake_failure(self):
|
|
pool = self._make_pool(test_targets=["www.example.com"])
|
|
|
|
mock_writer = MagicMock()
|
|
mock_writer.is_closing.return_value = False
|
|
mock_transport = MagicMock()
|
|
mock_transport.get_protocol.return_value = MagicMock()
|
|
mock_writer.transport = mock_transport
|
|
|
|
chain_ret = (MagicMock(), mock_writer)
|
|
with (
|
|
patch("s5p.pool.build_chain", new_callable=AsyncMock, return_value=chain_ret),
|
|
patch("asyncio.get_running_loop") as mock_loop_fn,
|
|
):
|
|
mock_loop = MagicMock()
|
|
mock_loop.start_tls = AsyncMock(
|
|
side_effect=ssl.SSLError("handshake failed"),
|
|
)
|
|
mock_loop_fn.return_value = mock_loop
|
|
|
|
result = asyncio.run(pool._tls_check([]))
|
|
|
|
assert result is False
|
|
|
|
def test_round_robin_rotation(self):
|
|
targets = ["host-a.example.com", "host-b.example.com", "host-c.example.com"]
|
|
pool = self._make_pool(test_targets=targets)
|
|
|
|
selected: list[str] = []
|
|
|
|
async def fake_build_chain(chain, host, port, timeout=None):
|
|
selected.append(host)
|
|
raise ConnectionError("test")
|
|
|
|
with patch("s5p.pool.build_chain", side_effect=fake_build_chain):
|
|
for _ in range(6):
|
|
asyncio.run(pool._tls_check([]))
|
|
|
|
assert selected == ["host-a.example.com", "host-b.example.com", "host-c.example.com",
|
|
"host-a.example.com", "host-b.example.com", "host-c.example.com"]
|
|
|
|
def test_empty_targets(self):
|
|
pool = self._make_pool(test_targets=[])
|
|
result = asyncio.run(pool._tls_check([]))
|
|
assert result is False
|
|
|
|
|
|
class TestProxyPoolConfigCompat:
|
|
"""Test backward compatibility for test_url -> test_targets."""
|
|
|
|
def test_legacy_test_url_converts(self):
|
|
cfg = ProxyPoolConfig(test_url="http://httpbin.org/ip")
|
|
assert cfg.test_targets == ["httpbin.org"]
|
|
|
|
def test_explicit_test_targets_wins(self):
|
|
cfg = ProxyPoolConfig(
|
|
test_url="http://httpbin.org/ip",
|
|
test_targets=["custom.example.com"],
|
|
)
|
|
assert cfg.test_targets == ["custom.example.com"]
|
|
|
|
def test_defaults_when_neither_set(self):
|
|
cfg = ProxyPoolConfig()
|
|
assert cfg.test_targets == ["www.google.com", "www.cloudflare.com", "www.amazon.com"]
|
|
assert cfg.test_url == ""
|