feat(03b): add BYO custom proxy provider + bounded crawler rotation-retry

Add CustomProxyProvider (single endpoint or rotating pool via Scrapling ProxyRotator), registered as 'custom' alongside anonymous_proxies and selectable via PROXY_PROVIDER. Adds is_pool_backed to the ProxyProvider ABC + a zero-arg package helper. The web crawler does a bounded one-per-tier is_proxy_error rotation-retry gated on is_pool_backed() (single-endpoint providers no-op). Config/.env.example gain CUSTOM_PROXY_URL(S). Zero-arg getter contract unchanged for all consumers. Documents the proprietary boundary test (generic proxy infra stays Apache-2). Tests: provider, registry, crawler rotation (16).

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
DESKTOP-RTLN3BA\$punk 2026-06-29 21:03:34 -07:00
parent 5c36cd3071
commit 62260125f7
12 changed files with 358 additions and 4 deletions

View file

@ -0,0 +1,79 @@
"""Unit tests for the BYO ``CustomProxyProvider`` (Phase 3b).
Covers single-endpoint vs pool behavior, cyclic rotation, env de-duplication,
the empty/unconfigured case, and the playwright-dict parse.
"""
import pytest
from app.config import Config
from app.utils.proxy.providers.custom import CustomProxyProvider
pytestmark = pytest.mark.unit
@pytest.fixture(autouse=True)
def _clear_proxy_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""Default both knobs to unset; individual tests set what they need."""
monkeypatch.setattr(Config, "CUSTOM_PROXY_URL", None)
monkeypatch.setattr(Config, "CUSTOM_PROXY_URLS", None)
def test_single_url_is_static(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "CUSTOM_PROXY_URL", "http://u:p@host:8080")
provider = CustomProxyProvider()
assert provider.is_pool_backed is False
assert provider.get_proxy_url() == "http://u:p@host:8080"
# Static endpoint returns the same value every call.
assert provider.get_proxy_url() == "http://u:p@host:8080"
def test_pool_rotates_cyclically(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
Config,
"CUSTOM_PROXY_URLS",
"http://a:1@h1:8080, http://b:2@h2:8080 ,http://c:3@h3:8080",
)
provider = CustomProxyProvider()
assert provider.is_pool_backed is True
seen = [provider.get_proxy_url() for _ in range(4)]
assert seen == [
"http://a:1@h1:8080",
"http://b:2@h2:8080",
"http://c:3@h3:8080",
"http://a:1@h1:8080", # wraps around
]
def test_single_plus_pool_dedupes(monkeypatch: pytest.MonkeyPatch) -> None:
"""A URL present in both CUSTOM_PROXY_URL and the pool is not duplicated."""
monkeypatch.setattr(Config, "CUSTOM_PROXY_URLS", "http://a@h1:80,http://b@h2:80")
monkeypatch.setattr(Config, "CUSTOM_PROXY_URL", "http://a@h1:80")
provider = CustomProxyProvider()
assert provider.is_pool_backed is True
seen = [provider.get_proxy_url() for _ in range(3)]
assert seen == ["http://a@h1:80", "http://b@h2:80", "http://a@h1:80"]
def test_unconfigured_returns_none() -> None:
provider = CustomProxyProvider()
assert provider.is_pool_backed is False
assert provider.get_proxy_url() is None
assert provider.get_requests_proxies() is None
def test_playwright_proxy_parses_credentials(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(Config, "CUSTOM_PROXY_URL", "http://user:pass@host:8080")
provider = CustomProxyProvider()
assert provider.get_playwright_proxy() == {
"server": "http://host:8080",
"username": "user",
"password": "pass",
}

View file

@ -0,0 +1,36 @@
"""Unit tests for proxy provider selection (Phase 3b).
``PROXY_PROVIDER`` selects the single app-wide provider; ``custom`` is now
registered alongside ``anonymous_proxies``, and unknown values still warn and
fall back to the default.
"""
import pytest
from app.config import Config
from app.utils.proxy import registry
from app.utils.proxy.providers.anonymous_proxies import AnonymousProxiesProvider
from app.utils.proxy.providers.custom import CustomProxyProvider
pytestmark = pytest.mark.unit
@pytest.fixture(autouse=True)
def _reset_active_provider(monkeypatch: pytest.MonkeyPatch) -> None:
"""Clear the process-wide provider cache so each test resolves fresh."""
monkeypatch.setattr(registry, "_active_provider", None)
def test_resolves_custom(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "PROXY_PROVIDER", "custom")
assert isinstance(registry.get_active_provider(), CustomProxyProvider)
def test_resolves_anonymous(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "PROXY_PROVIDER", "anonymous_proxies")
assert isinstance(registry.get_active_provider(), AnonymousProxiesProvider)
def test_unknown_falls_back_to_default(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "PROXY_PROVIDER", "does_not_exist")
assert isinstance(registry.get_active_provider(), AnonymousProxiesProvider)