104 lines
4.7 KiB
Python
104 lines
4.7 KiB
Python
"""test_server_pacing.py — юниты для per-provider pacing-интервалов (#1812 follow-up).
|
||
|
||
Проверяет:
|
||
1. per-provider env override переопределяет глобал для конкретного провайдера;
|
||
2. отсутствие per-provider env → фолбэк на глобальный BROWSER_MIN_PAGE_INTERVAL_S;
|
||
3. невалидный per-provider env (не float) → фолбэк на глобал, без исключения;
|
||
4. пустая строка как per-provider env → фолбэк на глобал (ValueError при float("")).
|
||
|
||
camoufox/Playwright НЕ поднимается: _resolve_min_interval вызывается с фейковым environ.
|
||
|
||
Запуск (из tradein-mvp/browser/)::
|
||
|
||
python -m pytest test_server_pacing.py -q
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import importlib.util
|
||
from pathlib import Path
|
||
|
||
# server.py — не пакет (отдельный сервис без __init__/pyproject). Грузим по пути.
|
||
_SERVER_PATH = Path(__file__).resolve().parent / "server.py"
|
||
_spec = importlib.util.spec_from_file_location("tradein_browser_server", _SERVER_PATH)
|
||
assert _spec is not None and _spec.loader is not None
|
||
server = importlib.util.module_from_spec(_spec)
|
||
_spec.loader.exec_module(server)
|
||
|
||
|
||
# ── _resolve_min_interval ────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_per_provider_override_replaces_global() -> None:
|
||
"""BROWSER_MIN_PAGE_INTERVAL_S_CIAN=18 + глобал=12 → cian=18, avito=12."""
|
||
fake_env = {
|
||
"BROWSER_MIN_PAGE_INTERVAL_S": "12",
|
||
"BROWSER_MIN_PAGE_INTERVAL_S_CIAN": "18",
|
||
}
|
||
# Патчим глобал в модуле, чтобы фолбэк брался из него (не из os.environ при load).
|
||
original_global = server.BROWSER_MIN_PAGE_INTERVAL_S
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = 12.0
|
||
try:
|
||
assert server._resolve_min_interval("cian", fake_env) == 18.0
|
||
assert server._resolve_min_interval("avito", fake_env) == 12.0
|
||
finally:
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = original_global
|
||
|
||
|
||
def test_no_per_provider_env_falls_back_to_global() -> None:
|
||
"""Нет BROWSER_MIN_PAGE_INTERVAL_S_YANDEX → фолбэк на глобал."""
|
||
fake_env: dict[str, str] = {}
|
||
original_global = server.BROWSER_MIN_PAGE_INTERVAL_S
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = 5.0
|
||
try:
|
||
result = server._resolve_min_interval("yandex", fake_env)
|
||
assert result == 5.0
|
||
finally:
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = original_global
|
||
|
||
|
||
def test_invalid_per_provider_env_falls_back_to_global() -> None:
|
||
"""Невалидный env (не float) → фолбэк на глобал, без исключения."""
|
||
fake_env = {"BROWSER_MIN_PAGE_INTERVAL_S_AVITO": "abc"}
|
||
original_global = server.BROWSER_MIN_PAGE_INTERVAL_S
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = 7.0
|
||
try:
|
||
result = server._resolve_min_interval("avito", fake_env)
|
||
assert result == 7.0
|
||
finally:
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = original_global
|
||
|
||
|
||
def test_empty_string_per_provider_env_falls_back_to_global() -> None:
|
||
"""Пустая строка env → float("") → ValueError → фолбэк на глобал."""
|
||
fake_env = {"BROWSER_MIN_PAGE_INTERVAL_S_GENERIC": ""}
|
||
original_global = server.BROWSER_MIN_PAGE_INTERVAL_S
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = 3.0
|
||
try:
|
||
result = server._resolve_min_interval("generic", fake_env)
|
||
assert result == 3.0
|
||
finally:
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = original_global
|
||
|
||
|
||
def test_zero_per_provider_env_disables_pacing() -> None:
|
||
"""BROWSER_MIN_PAGE_INTERVAL_S_AVITO=0 → 0.0 (без пейсинга для avito)."""
|
||
fake_env = {
|
||
"BROWSER_MIN_PAGE_INTERVAL_S": "12",
|
||
"BROWSER_MIN_PAGE_INTERVAL_S_AVITO": "0",
|
||
}
|
||
original_global = server.BROWSER_MIN_PAGE_INTERVAL_S
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = 12.0
|
||
try:
|
||
assert server._resolve_min_interval("avito", fake_env) == 0.0
|
||
# Другие провайдеры без override получают глобал.
|
||
assert server._resolve_min_interval("cian", fake_env) == 12.0
|
||
finally:
|
||
server.BROWSER_MIN_PAGE_INTERVAL_S = original_global
|
||
|
||
|
||
def test_all_providers_covered() -> None:
|
||
"""Все 4 провайдера (avito/cian/yandex/generic) обрабатываются резолвером."""
|
||
for provider in ("avito", "cian", "yandex", "generic"):
|
||
result = server._resolve_min_interval(provider, {})
|
||
assert isinstance(result, float), f"{provider}: ожидали float, получили {type(result)}"
|