feat(proxy): refactor proxy configuration and add DataImpulse provider

- Updated proxy configuration in `.env.example` files to use `PROXY_URL` and `PROXY_URLS` instead of `CUSTOM_PROXY_URL` and `CUSTOM_PROXY_URLS`.
- Introduced `DataImpulseProvider` for proxy management, replacing the deprecated `AnonymousProxiesProvider`.
- Enhanced documentation to reflect changes in proxy setup and usage.
- Adjusted related code in the proxy registry and configuration files to support the new provider structure.
This commit is contained in:
DESKTOP-RTLN3BA\$punk 2026-07-04 17:18:40 -07:00
parent 203215adc0
commit 701f888b9e
19 changed files with 303 additions and 193 deletions

View file

@ -15,12 +15,12 @@ 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)
monkeypatch.setattr(Config, "PROXY_URL", None)
monkeypatch.setattr(Config, "PROXY_URLS", None)
def test_single_url_is_static(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "CUSTOM_PROXY_URL", "http://u:p@host:8080")
monkeypatch.setattr(Config, "PROXY_URL", "http://u:p@host:8080")
provider = CustomProxyProvider()
assert provider.is_pool_backed is False
@ -32,7 +32,7 @@ def test_single_url_is_static(monkeypatch: pytest.MonkeyPatch) -> None:
def test_pool_rotates_cyclically(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
Config,
"CUSTOM_PROXY_URLS",
"PROXY_URLS",
"http://a:1@h1:8080, http://b:2@h2:8080 ,http://c:3@h3:8080",
)
provider = CustomProxyProvider()
@ -48,9 +48,9 @@ def test_pool_rotates_cyclically(monkeypatch: pytest.MonkeyPatch) -> None:
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")
"""A URL present in both PROXY_URL and the pool is not duplicated."""
monkeypatch.setattr(Config, "PROXY_URLS", "http://a@h1:80,http://b@h2:80")
monkeypatch.setattr(Config, "PROXY_URL", "http://a@h1:80")
provider = CustomProxyProvider()
assert provider.is_pool_backed is True
@ -69,7 +69,7 @@ def test_unconfigured_returns_none() -> None:
def test_playwright_proxy_parses_credentials(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(Config, "CUSTOM_PROXY_URL", "http://user:pass@host:8080")
monkeypatch.setattr(Config, "PROXY_URL", "http://user:pass@host:8080")
provider = CustomProxyProvider()
assert provider.get_playwright_proxy() == {

View file

@ -0,0 +1,72 @@
"""Unit tests for the DataImpulseProvider.
Takes a single full URL (like ``custom``); the vendor-specific bit is parsing the
``__cr.<country>`` username suffix for :meth:`get_location`. Playwright/requests
shapes come from the shared base parse, so a couple of checks cover the wiring.
"""
import pytest
from app.config import Config
from app.utils.proxy.providers.dataimpulse import DataImpulseProvider
pytestmark = pytest.mark.unit
_URL = "http://tok123__cr.us:secret@gw.dataimpulse.com:823"
@pytest.fixture(autouse=True)
def _clear_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "PROXY_URL", None)
def test_returns_configured_url(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "PROXY_URL", _URL)
provider = DataImpulseProvider()
assert provider.is_pool_backed is False
assert provider.get_proxy_url() == _URL
assert provider.get_requests_proxies() == {"http": _URL, "https": _URL}
def test_location_parsed_from_country_suffix(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "PROXY_URL", _URL)
assert DataImpulseProvider().get_location() == "us"
def test_location_stops_at_next_param(monkeypatch: pytest.MonkeyPatch) -> None:
# A sticky/session suffix after the country must not bleed into the location.
monkeypatch.setattr(
Config,
"PROXY_URL",
"http://tok__cr.de__sid.abc123:secret@gw.dataimpulse.com:823",
)
assert DataImpulseProvider().get_location() == "de"
def test_no_country_suffix_yields_empty_location(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
Config, "PROXY_URL", "http://tok:secret@gw.dataimpulse.com:823"
)
assert DataImpulseProvider().get_location() == ""
def test_unconfigured_returns_none() -> None:
provider = DataImpulseProvider()
assert provider.get_proxy_url() is None
assert provider.get_requests_proxies() is None
assert provider.get_playwright_proxy() is None
assert provider.get_location() == ""
def test_playwright_proxy_from_base_parse(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "PROXY_URL", _URL)
assert DataImpulseProvider().get_playwright_proxy() == {
"server": "http://gw.dataimpulse.com:823",
"username": "tok123__cr.us",
"password": "secret",
}

View file

@ -1,16 +1,16 @@
"""Unit tests for proxy provider selection (Phase 3b).
"""Unit tests for proxy provider selection.
``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.
``PROXY_PROVIDER`` selects the single app-wide provider; ``custom`` (the default)
and ``dataimpulse`` are registered, 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
from app.utils.proxy.providers.dataimpulse import DataImpulseProvider
pytestmark = pytest.mark.unit
@ -26,11 +26,11 @@ def test_resolves_custom(monkeypatch: pytest.MonkeyPatch) -> None:
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_resolves_dataimpulse(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(Config, "PROXY_PROVIDER", "dataimpulse")
assert isinstance(registry.get_active_provider(), DataImpulseProvider)
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)
assert isinstance(registry.get_active_provider(), CustomProxyProvider)