test(10-02): add health check tests for proxy pool

- test_fail_threshold: verify 3 consecutive failures mark proxy unavailable
- test_success_resets_count: verify success resets failure count
- test_cooldown_recovery: verify proxy recovers after cooldown
- test_cooldown_not_elapsed: verify proxy stays unavailable during cooldown
- test_all_unavailable_fallback: verify fallback to first proxy when all fail
- test_get_available_count: verify available proxy count tracking

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-18 16:46:37 +08:00
parent 96dfdce59c
commit 36a45095c1
+128
View File
@@ -176,3 +176,131 @@ class TestProxyEntry:
assert entry.fail_count == 0
assert entry.last_fail_time is None
assert entry.is_available is True
def test_fail_threshold(self):
"""Test that proxy becomes unavailable after 3 consecutive failures."""
entry = ProxyEntry(url="http://proxy:8080")
# First two failures should not mark as unavailable
entry.record_failure(max_fail_count=3)
assert entry.fail_count == 1
assert entry.is_available is True
entry.record_failure(max_fail_count=3)
assert entry.fail_count == 2
assert entry.is_available is True
# Third failure should mark as unavailable
entry.record_failure(max_fail_count=3)
assert entry.fail_count == 3
assert entry.is_available is False
def test_success_resets_count(self):
"""Test that recording success resets failure count and restores availability."""
entry = ProxyEntry(url="http://proxy:8080")
# Record some failures
entry.record_failure(max_fail_count=3)
entry.record_failure(max_fail_count=3)
assert entry.fail_count == 2
# Record success
entry.record_success()
assert entry.fail_count == 0
assert entry.is_available is True
def test_cooldown_recovery(self):
"""Test that failed proxy recovers after cooldown period."""
import time
entry = ProxyEntry(url="http://proxy:8080")
# Mark as unavailable
entry.record_failure(max_fail_count=3)
entry.record_failure(max_fail_count=3)
entry.record_failure(max_fail_count=3)
assert entry.is_available is False
# Wait for cooldown (using short cooldown for testing)
time.sleep(1)
# Check recovery with 1 second cooldown
result = entry.check_recovery(cooldown_seconds=1)
assert result is True
assert entry.is_available is True
assert entry.fail_count == 0
def test_cooldown_not_elapsed(self):
"""Test that failed proxy does not recover before cooldown period."""
entry = ProxyEntry(url="http://proxy:8080")
# Mark as unavailable
entry.record_failure(max_fail_count=3)
entry.record_failure(max_fail_count=3)
entry.record_failure(max_fail_count=3)
assert entry.is_available is False
# Check recovery immediately with 60 second cooldown
result = entry.check_recovery(cooldown_seconds=60)
assert result is False
assert entry.is_available is False
class TestProxyPoolHealthCheck:
"""Tests for ProxyPool health check and availability methods."""
def test_all_unavailable_fallback(self):
"""Test that get_next_proxy returns first proxy when all are unavailable."""
config = ProxyPoolConfig(
proxies=[
"http://proxy-a:8080",
"http://proxy-b:8080",
"http://proxy-c:8080",
],
max_fail_count=3
)
pool = ProxyPool(config)
# Mark all proxies as unavailable
for _ in range(3):
pool._entries[0].record_failure(max_fail_count=3)
pool._entries[1].record_failure(max_fail_count=3)
pool._entries[2].record_failure(max_fail_count=3)
# All should be unavailable
assert pool._entries[0].is_available is False
assert pool._entries[1].is_available is False
assert pool._entries[2].is_available is False
# Should fallback to first proxy
proxy = pool.get_next_proxy()
assert proxy is not None
assert proxy.url == "http://proxy-a:8080"
def test_get_available_count(self):
"""Test that get_available_count returns correct number of available proxies."""
config = ProxyPoolConfig(
proxies=[
"http://proxy-a:8080",
"http://proxy-b:8080",
"http://proxy-c:8080",
],
max_fail_count=3
)
pool = ProxyPool(config)
# Initially all available
assert pool.get_available_count() == 3
# Mark one as unavailable
for _ in range(3):
pool._entries[0].record_failure(max_fail_count=3)
assert pool.get_available_count() == 2
# Mark another as unavailable
for _ in range(3):
pool._entries[1].record_failure(max_fail_count=3)
assert pool.get_available_count() == 1