294 lines
14 KiB
Python
294 lines
14 KiB
Python
"""test_server.py — юниты per-provider concurrency для tradein-browser (#1793).
|
||
|
||
Проверяет ядро per-provider модели:
|
||
1. _provider_for_url — корректный маппинг host → provider (avito/cian/yandex/generic);
|
||
2. _provider_proxy — BROWSER_PROXY_* primary с фолбэком на legacy *_PROXY_URL;
|
||
3. _resolve_provider — явный body["provider"] перебивает host, иначе host;
|
||
4. конкурентность: два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи),
|
||
два к ОДНОМУ — сериализованы (один лок). Через _do_fetch с asyncio.Event и
|
||
замером перекрытия по времени.
|
||
|
||
camoufox НЕ запускается: _ensure_browser / _do_fetch мокаются. Хендлер вызывается
|
||
напрямую (make_mocked_request не поднимает сокет).
|
||
|
||
Запуск (из tradein-mvp/browser/)::
|
||
|
||
python -m pytest test_server.py -q
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import importlib.util
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from aiohttp.test_utils import make_mocked_request
|
||
|
||
# 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)
|
||
|
||
|
||
def _json_body(response: Any) -> dict[str, Any]:
|
||
return json.loads(response.body.decode())
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Чистое per-provider состояние + инициализированный _locks_guard на каждый тест."""
|
||
monkeypatch.setattr(server, "_browsers", {})
|
||
monkeypatch.setattr(server, "_browser_cms", {})
|
||
monkeypatch.setattr(server, "_page_counters", {})
|
||
monkeypatch.setattr(server, "_locks", {})
|
||
monkeypatch.setattr(server, "_retry_tasks", {})
|
||
monkeypatch.setattr(server, "_locks_guard", asyncio.Lock())
|
||
|
||
|
||
# ── _provider_for_url ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_provider_for_url_avito() -> None:
|
||
assert server._provider_for_url("https://www.avito.ru/moskva/kvartiry") == "avito"
|
||
assert server._provider_for_url("https://m.avito.ru/item/123") == "avito"
|
||
|
||
|
||
def test_provider_for_url_cian() -> None:
|
||
assert server._provider_for_url("https://www.cian.ru/sale/flat/1/") == "cian"
|
||
assert server._provider_for_url("https://spb.cian.ru/x") == "cian"
|
||
|
||
|
||
def test_provider_for_url_yandex() -> None:
|
||
assert server._provider_for_url("https://realty.yandex.ru/offer/1/") == "yandex"
|
||
|
||
|
||
def test_provider_for_url_generic_fallback() -> None:
|
||
assert server._provider_for_url("https://example.com/x") == "generic"
|
||
assert server._provider_for_url("https://domclick.ru/x") == "generic"
|
||
# Битый / пустой URL — без хоста → generic, не падаем.
|
||
assert server._provider_for_url("not-a-url") == "generic"
|
||
assert server._provider_for_url("") == "generic"
|
||
|
||
|
||
# ── _provider_proxy ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_provider_proxy_uses_browser_proxy_primary(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""BROWSER_PROXY_* — primary источник для каждого поставщика."""
|
||
monkeypatch.setenv("BROWSER_PROXY_AVITO", "http://u:p@avito-proxy:8080")
|
||
monkeypatch.setenv("BROWSER_PROXY_CIAN", "http://u:p@cian-proxy:8080")
|
||
monkeypatch.setenv("BROWSER_PROXY_YANDEX", "http://u:p@yandex-proxy:8080")
|
||
assert server._provider_proxy("avito") == "http://u:p@avito-proxy:8080"
|
||
assert server._provider_proxy("cian") == "http://u:p@cian-proxy:8080"
|
||
assert server._provider_proxy("yandex") == "http://u:p@yandex-proxy:8080"
|
||
|
||
|
||
def test_provider_proxy_falls_back_to_legacy(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Без BROWSER_PROXY_* берётся legacy *_PROXY_URL фолбэк."""
|
||
monkeypatch.delenv("BROWSER_PROXY_AVITO", raising=False)
|
||
monkeypatch.delenv("BROWSER_PROXY_CIAN", raising=False)
|
||
monkeypatch.delenv("BROWSER_PROXY_YANDEX", raising=False)
|
||
monkeypatch.setenv("AVITO_PROXY_URL", "http://u:p@legacy-avito:8080")
|
||
monkeypatch.setenv("CIAN_PROXY_URL", "http://u:p@legacy-cian:8080")
|
||
monkeypatch.setenv("YANDEX_PROXY_URL", "http://u:p@legacy-yandex:8080")
|
||
assert server._provider_proxy("avito") == "http://u:p@legacy-avito:8080"
|
||
assert server._provider_proxy("cian") == "http://u:p@legacy-cian:8080"
|
||
assert server._provider_proxy("yandex") == "http://u:p@legacy-yandex:8080"
|
||
|
||
|
||
def test_provider_proxy_generic(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""generic: SCRAPER_PROXY_URL primary, AVITO_PROXY_URL фолбэк."""
|
||
monkeypatch.delenv("SCRAPER_PROXY_URL", raising=False)
|
||
monkeypatch.setenv("AVITO_PROXY_URL", "http://u:p@avito:8080")
|
||
assert server._provider_proxy("generic") == "http://u:p@avito:8080"
|
||
monkeypatch.setenv("SCRAPER_PROXY_URL", "http://u:p@scraper:8080")
|
||
assert server._provider_proxy("generic") == "http://u:p@scraper:8080"
|
||
|
||
|
||
def test_provider_proxy_none_when_unset(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Ни одной env → None (direct, как одиночный браузер без прокси)."""
|
||
for var in (
|
||
"BROWSER_PROXY_CIAN",
|
||
"CIAN_PROXY_URL",
|
||
):
|
||
monkeypatch.delenv(var, raising=False)
|
||
assert server._provider_proxy("cian") is None
|
||
|
||
|
||
def test_provider_proxy_browser_proxy_wins_over_legacy(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""BROWSER_PROXY_* перебивает legacy *_PROXY_URL (http vs возможный socks5)."""
|
||
monkeypatch.setenv("BROWSER_PROXY_CIAN", "http://u:p@http-cian:8080")
|
||
monkeypatch.setenv("CIAN_PROXY_URL", "socks5://u:p@socks-cian:1080")
|
||
assert server._provider_proxy("cian") == "http://u:p@http-cian:8080"
|
||
|
||
|
||
# ── _resolve_provider ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_resolve_provider_by_host() -> None:
|
||
assert server._resolve_provider({}, "https://www.avito.ru/x") == "avito"
|
||
assert server._resolve_provider({}, "https://www.cian.ru/x") == "cian"
|
||
|
||
|
||
def test_resolve_provider_explicit_overrides_host() -> None:
|
||
"""Явный валидный body["provider"] перебивает host."""
|
||
assert server._resolve_provider({"provider": "yandex"}, "https://www.avito.ru/x") == "yandex"
|
||
|
||
|
||
def test_resolve_provider_invalid_explicit_falls_back_to_host() -> None:
|
||
"""Невалидный body["provider"] игнорируется → берём по host."""
|
||
assert server._resolve_provider({"provider": "bogus"}, "https://www.cian.ru/x") == "cian"
|
||
assert server._resolve_provider({"provider": 123}, "https://www.cian.ru/x") == "cian"
|
||
|
||
|
||
def test_resolve_provider_uses_source_field() -> None:
|
||
"""browser_fetcher шлёт {"url","source"} — explicit берётся и из "source"."""
|
||
assert server._resolve_provider({"source": "avito"}, "https://example.com/x") == "avito"
|
||
assert server._resolve_provider({"source": "yandex"}, "https://www.avito.ru/x") == "yandex"
|
||
|
||
|
||
def test_resolve_provider_domclick_source_to_generic() -> None:
|
||
"""domclick scraper шлёт source='domklik' (нет в PROVIDERS) + хост domclick.ru.
|
||
|
||
'domklik' не валиден → host-detect domclick.ru → generic. Намеренно: в prod
|
||
нет выделенного BROWSER_PROXY_DOMCLICK.
|
||
"""
|
||
assert (
|
||
server._resolve_provider({"source": "domklik"}, "https://domclick.ru/x") == "generic"
|
||
)
|
||
|
||
|
||
# ── _lock_for: один лок на поставщика ────────────────────────────────────────────
|
||
|
||
|
||
def test_lock_for_is_stable_per_provider() -> None:
|
||
"""_lock_for возвращает один и тот же объект-лок для поставщика, разные — для разных."""
|
||
|
||
async def _run() -> tuple[Any, Any, Any]:
|
||
a1 = await server._lock_for("avito")
|
||
a2 = await server._lock_for("avito")
|
||
c1 = await server._lock_for("cian")
|
||
return a1, a2, c1
|
||
|
||
a1, a2, c1 = asyncio.run(_run())
|
||
assert a1 is a2, "лок одного поставщика стабилен"
|
||
assert a1 is not c1, "локи разных поставщиков различны"
|
||
|
||
|
||
# ── concurrency: параллель между поставщиками, сериализация внутри ───────────────
|
||
|
||
|
||
def _make_fetch_request(url: str) -> Any:
|
||
request = make_mocked_request("POST", "/fetch")
|
||
request.json = lambda: _coro({"url": url}) # type: ignore[method-assign]
|
||
return request
|
||
|
||
|
||
class _OverlapProbe:
|
||
"""Мок _do_fetch: меряет максимальное число одновременно активных fetch per-provider."""
|
||
|
||
def __init__(self, hold: float = 0.05) -> None:
|
||
self.hold = hold
|
||
self.active: dict[str, int] = {}
|
||
self.peak: dict[str, int] = {}
|
||
self.global_active = 0
|
||
self.global_peak = 0
|
||
|
||
async def __call__(self, provider: str, url: str) -> str:
|
||
self.active[provider] = self.active.get(provider, 0) + 1
|
||
self.peak[provider] = max(self.peak.get(provider, 0), self.active[provider])
|
||
self.global_active += 1
|
||
self.global_peak = max(self.global_peak, self.global_active)
|
||
try:
|
||
await asyncio.sleep(self.hold)
|
||
finally:
|
||
self.active[provider] -= 1
|
||
self.global_active -= 1
|
||
return f"<html>{provider}</html>"
|
||
|
||
|
||
def test_same_provider_fetches_are_serialized(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Два /fetch к ОДНОМУ поставщику сериализованы (один лок) — peak==1."""
|
||
probe = _OverlapProbe()
|
||
|
||
async def _ensure(provider: str) -> bool:
|
||
server._browsers[provider] = object()
|
||
return True
|
||
|
||
monkeypatch.setattr(server, "_ensure_browser", _ensure)
|
||
monkeypatch.setattr(server, "_do_fetch", probe)
|
||
|
||
async def _run() -> None:
|
||
await asyncio.gather(
|
||
server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
|
||
server.fetch_handler(_make_fetch_request("https://www.avito.ru/b")),
|
||
server.fetch_handler(_make_fetch_request("https://www.avito.ru/c")),
|
||
)
|
||
|
||
asyncio.run(_run())
|
||
assert probe.peak["avito"] == 1, "внутри поставщика строго ≤1 одновременный fetch"
|
||
|
||
|
||
def test_different_providers_fetch_concurrently(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи) — global peak>1."""
|
||
probe = _OverlapProbe()
|
||
|
||
async def _ensure(provider: str) -> bool:
|
||
server._browsers[provider] = object()
|
||
return True
|
||
|
||
monkeypatch.setattr(server, "_ensure_browser", _ensure)
|
||
monkeypatch.setattr(server, "_do_fetch", probe)
|
||
|
||
async def _run() -> list[Any]:
|
||
return await asyncio.gather(
|
||
server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
|
||
server.fetch_handler(_make_fetch_request("https://www.cian.ru/b")),
|
||
server.fetch_handler(_make_fetch_request("https://realty.yandex.ru/c")),
|
||
)
|
||
|
||
responses = asyncio.run(_run())
|
||
assert all(r.status == 200 for r in responses)
|
||
# >=2 (не ==3): на нагруженном CI планировщик может не вывести все три корутины
|
||
# в overlap одновременно; перекрытие хотя бы двух уже доказывает параллельность
|
||
# между поставщиками. Same-provider тест (peak==1) остаётся строгим.
|
||
assert probe.global_peak >= 2, "разные поставщики должны фетчить конкурентно"
|
||
# У каждого поставщика по одному запросу — внутренний peak == 1.
|
||
assert probe.peak == {"avito": 1, "cian": 1, "yandex": 1}
|
||
|
||
|
||
def test_mixed_load_serializes_within_parallelizes_across(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""Смешанная нагрузка: avito×2 сериализованы, но параллельны с cian×1."""
|
||
probe = _OverlapProbe()
|
||
|
||
async def _ensure(provider: str) -> bool:
|
||
server._browsers[provider] = object()
|
||
return True
|
||
|
||
monkeypatch.setattr(server, "_ensure_browser", _ensure)
|
||
monkeypatch.setattr(server, "_do_fetch", probe)
|
||
|
||
async def _run() -> None:
|
||
await asyncio.gather(
|
||
server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
|
||
server.fetch_handler(_make_fetch_request("https://www.avito.ru/b")),
|
||
server.fetch_handler(_make_fetch_request("https://www.cian.ru/c")),
|
||
)
|
||
|
||
asyncio.run(_run())
|
||
assert probe.peak["avito"] == 1, "avito сериализован сам с собой"
|
||
assert probe.peak["cian"] == 1
|
||
assert probe.global_peak >= 2, "avito и cian должны перекрываться во времени"
|
||
|
||
|
||
async def _coro(value: Any) -> Any:
|
||
return value
|