125 lines
5 KiB
Python
125 lines
5 KiB
Python
"""Unit-тесты для KN-loader anti-ban рычагов в scrapers/stealth (#1945).
|
||
|
||
DOM.РФ WAF банит VPS-IP по volume/rate на full sweep (~17k запросов при
|
||
concurrency=8). Лечим: (1) per-instance throttle (concurrency + шире jitter)
|
||
для KN-sweep, НЕ трогая модульный дефолт остальных скраперов; (2) опциональный
|
||
прокси (default off → поведение без изменений).
|
||
|
||
Тесты проверяют:
|
||
• parse_proxy_url корректно разбирает http://user:pass@host:port;
|
||
• BrowserSession прокидывает concurrency/jitter/proxy в self-поля;
|
||
• дефолтное (None) поведение НЕ меняется (concurrency=8, jitter 600–1500, без прокси).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from app.services.scrapers.stealth import (
|
||
_BROWSER_CONCURRENCY,
|
||
_DEFAULT_JITTER_MAX_MS,
|
||
_DEFAULT_JITTER_MIN_MS,
|
||
BrowserSession,
|
||
parse_proxy_url,
|
||
)
|
||
|
||
|
||
class TestParseProxyUrl:
|
||
"""parse_proxy_url → Playwright proxy-dict {server, username?, password?}."""
|
||
|
||
def test_none_returns_none(self) -> None:
|
||
assert parse_proxy_url(None) is None
|
||
|
||
def test_empty_string_returns_none(self) -> None:
|
||
assert parse_proxy_url("") is None
|
||
|
||
def test_full_creds_url(self) -> None:
|
||
got = parse_proxy_url("http://user:pass@host.example:8080")
|
||
assert got == {
|
||
"server": "http://host.example:8080",
|
||
"username": "user",
|
||
"password": "pass",
|
||
}
|
||
|
||
def test_no_creds(self) -> None:
|
||
got = parse_proxy_url("http://host.example:3128")
|
||
assert got == {"server": "http://host.example:3128"}
|
||
|
||
def test_creds_not_leaked_into_server(self) -> None:
|
||
# Playwright требует creds ОТДЕЛЬНЫМИ ключами, не в server-URL.
|
||
# Используем уникальные токены, которых нет в host/scheme.
|
||
got = parse_proxy_url("http://secretuser:secretpass@1.2.3.4:9999")
|
||
assert got is not None
|
||
assert got["server"] == "http://1.2.3.4:9999"
|
||
assert "secretuser" not in got["server"]
|
||
assert "secretpass" not in got["server"]
|
||
assert "@" not in got["server"]
|
||
|
||
def test_scheme_preserved(self) -> None:
|
||
got = parse_proxy_url("https://u:p@host:443")
|
||
assert got is not None
|
||
assert got["server"] == "https://host:443"
|
||
|
||
def test_default_scheme_when_missing(self) -> None:
|
||
# urlsplit без схемы кладёт всё в path → host не резолвится → ValueError.
|
||
with pytest.raises(ValueError):
|
||
parse_proxy_url("host.example:8080")
|
||
|
||
def test_host_required(self) -> None:
|
||
with pytest.raises(ValueError):
|
||
parse_proxy_url("http://:8080")
|
||
|
||
def test_error_does_not_leak_credentials(self) -> None:
|
||
# ValueError всплывает в kn_scrape_runs.error / log_progress / Sentry —
|
||
# НЕ должен содержать пароль (#1945 sec-review).
|
||
bad = "http://secretuser:secretpass@:8080" # есть creds, но нет host
|
||
with pytest.raises(ValueError) as exc:
|
||
parse_proxy_url(bad)
|
||
msg = str(exc.value)
|
||
assert "secretpass" not in msg
|
||
assert "secretuser" not in msg
|
||
assert bad not in msg
|
||
|
||
|
||
class TestBrowserSessionThrottleThreading:
|
||
"""concurrency/jitter/proxy прокидываются в инстанс-поля BrowserSession."""
|
||
|
||
def test_concurrency_threaded(self) -> None:
|
||
sess = BrowserSession(concurrency=2)
|
||
assert sess._concurrency == 2
|
||
assert sess._sem._value == 2
|
||
|
||
def test_jitter_threaded(self) -> None:
|
||
sess = BrowserSession(jitter_min_ms=1200, jitter_max_ms=3000)
|
||
assert sess._jitter_min_ms == 1200
|
||
assert sess._jitter_max_ms == 3000
|
||
|
||
def test_proxy_threaded_and_parsed(self) -> None:
|
||
sess = BrowserSession(proxy_url="http://u:p@proxy.local:8000")
|
||
assert sess._proxy == {
|
||
"server": "http://proxy.local:8000",
|
||
"username": "u",
|
||
"password": "p",
|
||
}
|
||
|
||
|
||
class TestBrowserSessionDefaultsUnchanged:
|
||
"""None у каждого рычага = старое поведение (другие скраперы НЕ затронуты)."""
|
||
|
||
def test_default_concurrency_is_module_default(self) -> None:
|
||
sess = BrowserSession()
|
||
assert sess._concurrency == _BROWSER_CONCURRENCY == 8
|
||
assert sess._sem._value == 8
|
||
|
||
def test_default_jitter_is_module_default(self) -> None:
|
||
sess = BrowserSession()
|
||
assert sess._jitter_min_ms == _DEFAULT_JITTER_MIN_MS == 600
|
||
assert sess._jitter_max_ms == _DEFAULT_JITTER_MAX_MS == 1500
|
||
|
||
def test_default_proxy_is_none(self) -> None:
|
||
sess = BrowserSession()
|
||
assert sess._proxy is None
|
||
|
||
def test_explicit_none_proxy_is_none(self) -> None:
|
||
sess = BrowserSession(proxy_url=None)
|
||
assert sess._proxy is None
|