From 77ae4ca779543698543b93afa09cf8d53eb31d2d Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 19 Jun 2026 18:52:55 +0300 Subject: [PATCH] =?UTF-8?q?feat(browser):=20per-provider=20concurrency=20?= =?UTF-8?q?=E2=80=94=20=D0=BF=D0=B0=D1=80=D0=B0=D0=BB=D0=BB=D0=B5=D0=BB?= =?UTF-8?q?=D1=8C=20=D0=BC=D0=B5=D0=B6=D0=B4=D1=83=20=D0=BF=D0=BE=D1=81?= =?UTF-8?q?=D1=82=D0=B0=D0=B2=D1=89=D0=B8=D0=BA=D0=B0=D0=BC=D0=B8,=20?= =?UTF-8?q?=E2=89=A41=20=D0=BA=D0=BE=D0=BD=D0=BD=D0=B5=D0=BA=D1=82=20?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=BF=D0=BE=D1=81=D1=82=D0=B0=D0=B2=D1=89=D0=B8?= =?UTF-8?q?=D0=BA=D0=B0=20(#1793)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/browser/server.py | 786 ++++++++--------------- tradein-mvp/browser/test_server.py | 274 ++++++++ tradein-mvp/browser/test_server_smoke.py | 259 ++------ 3 files changed, 625 insertions(+), 694 deletions(-) create mode 100644 tradein-mvp/browser/test_server.py diff --git a/tradein-mvp/browser/server.py b/tradein-mvp/browser/server.py index 8df8c923..05692e13 100644 --- a/tradein-mvp/browser/server.py +++ b/tradein-mvp/browser/server.py @@ -4,7 +4,7 @@ Он запускает camoufox **локально** внутри контейнера (AsyncCamoufox) и экспонирует простой HTTP API на базе aiohttp: - GET /health → {"status": "ok"} + GET /health → {"status": "ok", "browsers": {"avito": bool, ...}} POST /fetch → {"url": "..."} → {"html": "..."} POST /login → {"url": "...", "email": "...", "password": "...", ...} → {"cookies": [...]} @@ -14,10 +14,23 @@ WS-сервер) несовместим с современными версия с MODULE_NOT_FOUND / RuntimeError при запуске через ``launch_server``. Локальный запуск ``AsyncCamoufox`` работает стабильно — проверено. +Per-provider модель (#1793): + Вместо одного глобального браузера + одного egress-прокси держим ОТДЕЛЬНЫЙ + инстанс camoufox на каждого поставщика (avito/cian/yandex/generic). Поставщик + определяется по host URL запроса (или явному полю body["provider"]). Каждый + инстанс ходит через свой http-прокси (BROWSER_PROXY_AVITO/CIAN/YANDEX) и имеет + geoip=True — фингерпринт привязан к гео конкретного прокси, поэтому смешивать + источники в одном браузере нельзя. + + Конкурентность: + * МЕЖДУ поставщиками — параллельно (avito/cian/yandex гоняются одновременно); + * ВНУТРИ поставщика — строго ≤1 одновременный fetch (per-provider lock). + + Это решает проблему, когда источники клинили друг друга через единственный + egress-прокси / единственный сериализующий лок. + Переменные окружения: BROWSER_PORT — TCP-порт HTTP-сервера (default: 3000) - BROWSER_CONCURRENCY — сколько /fetch обрабатывается параллельно (default: 4, clamp >=1). - Один браузер, N параллельных страниц (по странице на запрос). BROWSER_RECYCLE_PAGES — страниц в одном сеансе браузера до перезапуска (default: 15) BROWSER_NAV_TIMEOUT_MS — таймаут page.goto в мс (default: 60000) BROWSER_WAIT_MS — ожидание гидрации listings после DOMContentLoaded, мс (default: 6000) @@ -30,8 +43,14 @@ WS-сервер) несовместим с современными версия BROWSER_BLOCKED_RESOURCE_TYPES — CSV типов ресурсов Playwright для блокировки (default: "image,media,font"). Менять консервативно: блокировка script/stylesheet/document/xhr сломает выдачу. - AVITO_PROXY_URL — прокси ``http://user:pass@host:port`` (mobileproxy, #623); - SCRAPER_PROXY_URL — generic-fallback. Без него — soft-block. + BROWSER_PROXY_AVITO — http-прокси для avito (fallback AVITO_PROXY_URL). + BROWSER_PROXY_CIAN — http-прокси для cian (fallback CIAN_PROXY_URL). + BROWSER_PROXY_YANDEX — http-прокси для yandex (fallback YANDEX_PROXY_URL). + SCRAPER_PROXY_URL — http-прокси для generic (fallback AVITO_PROXY_URL). + + ВАЖНО: для браузера используем именно BROWSER_PROXY_* (http-прокси). Legacy + CIAN_PROXY_URL/YANDEX_PROXY_URL могут быть socks5 с авторизацией — playwright + их не умеет, поэтому они лишь fallback (если кто-то положил http туда же). Контракт /login (провалидировано вживую 2026-05-31, Cian email+пароль без SMS): pre_click_selectors — список селекторов для последовательного клика до формы; @@ -55,30 +74,6 @@ logger = logging.getLogger(__name__) # ── конфигурация из env ──────────────────────────────────────────────────────── BROWSER_PORT: int = int(os.environ.get("BROWSER_PORT", "3000")) - - -def _parse_concurrency(raw: str | None) -> int: - """Парсит BROWSER_CONCURRENCY → int, clamp >=1, fallback 4 при мусоре. - - Ops поднимает значение в .env.runtime после смоука. Невалидный ввод - (нечисловой / пустой) → дефолт 4, чтобы кривой конфиг не валил сервис. - """ - if raw is None: - return 4 - try: - value = int(raw) - except (TypeError, ValueError): - logger.warning( - "tradein-browser: невалидный BROWSER_CONCURRENCY=%r — дефолт 4", raw - ) - return 4 - return max(1, value) - - -# Сколько /fetch обрабатывается параллельно. Один браузер, N параллельных -# страниц (по странице на запрос). Дефолт 4 — консервативно; ops поднимет до 8 -# после смоука (firefox-страницы прожорливы по RAM, mem_limit поднят до 5g). -BROWSER_CONCURRENCY: int = _parse_concurrency(os.environ.get("BROWSER_CONCURRENCY")) BROWSER_RECYCLE_PAGES: int = int(os.environ.get("BROWSER_RECYCLE_PAGES", "15")) BROWSER_NAV_TIMEOUT_MS: int = int(os.environ.get("BROWSER_NAV_TIMEOUT_MS", "60000")) # 6000 (не 2500): avito гидрирует listings client-side ПОСЛЕ domcontentloaded; @@ -86,6 +81,10 @@ BROWSER_NAV_TIMEOUT_MS: int = int(os.environ.get("BROWSER_NAV_TIMEOUT_MS", "6000 # выдача (~50 карточек, 3.2МБ). Подтверждено прод-дебагом 2026-05-31. BROWSER_WAIT_MS: int = int(os.environ.get("BROWSER_WAIT_MS", "6000")) +# Известные поставщики. "generic" — фолбэк для всех прочих хостов (один общий +# инстанс на неузнанные домены). Порядок задаёт детерминированный health-вывод. +PROVIDERS: tuple[str, ...] = ("avito", "cian", "yandex", "generic") + def _parse_bool(raw: str | None, default: bool) -> bool: """Парсит булев env-флаг ("true"/"1"/"yes"/"on" → True), fallback на default.""" @@ -106,53 +105,37 @@ _BLOCKED_TYPES: frozenset[str] = frozenset( if t.strip() ) -# Прокси: основной источник — AVITO_PROXY_URL (mobileproxy, #623, лежит в -# backend/.env.runtime, тот же что у curl_cffi-avito). SCRAPER_PROXY_URL — -# опциональный generic-fallback на будущее (cian/yandex). Без прокси camoufox -# фетчит с датацентр-IP → avito отдаёт soft-block заглушку. -SCRAPER_PROXY_URL: str | None = os.environ.get("AVITO_PROXY_URL") or os.environ.get( - "SCRAPER_PROXY_URL" -) -# ── per-source proxy pool (Phase 1, feature-flagged) ─────────────────────────── -# По умолчанию ВЫКЛЮЧЕН: при FEATURE_BROWSER_POOL_ENABLED!=true поведение байт-в-байт -# идентично текущему (один глобальный браузер на SCRAPER_PROXY_URL). Когда флаг -# включён, /fetch роутится по полю body["source"] на отдельный браузер+прокси из -# BROWSER_PROXY_MAP — источники (avito/cian/yandex/...) больше не клинят друг друга -# через единственный egress-прокси. -# -# Deploy-env (Phase 2, owner-gated — задаётся в .env.runtime): -# FEATURE_BROWSER_POOL_ENABLED — "true" включает per-source pool (default false) -# BROWSER_PROXY_AVITO — прокси для source=avito (fallback AVITO_PROXY_URL/SCRAPER_PROXY_URL) -# BROWSER_PROXY_CIAN — прокси для source=cian (fallback CIAN_PROXY_URL) -# BROWSER_PROXY_YANDEX — прокси для source=yandex (fallback YANDEX_PROXY_URL) -# BROWSER_PROXY_DOMCLICK — прокси для source=domclick (нет fallback → роутится на avito) -FEATURE_BROWSER_POOL_ENABLED: bool = ( - os.environ.get("FEATURE_BROWSER_POOL_ENABLED", "false").lower() == "true" -) +def _provider_for_url(url: str) -> str: + """Определяет поставщика по host URL. - -def _build_proxy_map() -> dict[str, str]: - """Строит source → proxy_url из env, отбрасывая пустые значения. - - Каждый источник берёт свою явную BROWSER_PROXY_, иначе падает на - legacy-переменную (AVITO_PROXY_URL/CIAN_PROXY_URL/...). Пустые/None НЕ - попадают в map — caller трактует отсутствие как «нет прокси → direct». + avito/cian/yandex → одноимённый провайдер; всё прочее → "generic". """ - raw: dict[str, str | None] = { - "avito": ( - os.environ.get("BROWSER_PROXY_AVITO") - or os.environ.get("AVITO_PROXY_URL") - or os.environ.get("SCRAPER_PROXY_URL") - ), - "cian": os.environ.get("BROWSER_PROXY_CIAN") or os.environ.get("CIAN_PROXY_URL"), - "yandex": os.environ.get("BROWSER_PROXY_YANDEX") or os.environ.get("YANDEX_PROXY_URL"), - "domclick": os.environ.get("BROWSER_PROXY_DOMCLICK"), + host = (urlparse(url).hostname or "").lower() + if "avito" in host: + return "avito" + if "cian" in host: + return "cian" + if "yandex" in host: + return "yandex" + return "generic" + + +def _provider_proxy(provider: str) -> str | None: + """Возвращает http-прокси для поставщика из env (с фолбэком), иначе None. + + Используем BROWSER_PROXY_* как primary — это http-прокси для браузера. + Legacy *_PROXY_URL — лишь fallback: playwright НЕ умеет socks5 с авторизацией, + поэтому socks5-варианты CIAN/YANDEX как primary не годятся. + """ + env_map: dict[str, tuple[str, str]] = { + "avito": ("BROWSER_PROXY_AVITO", "AVITO_PROXY_URL"), + "cian": ("BROWSER_PROXY_CIAN", "CIAN_PROXY_URL"), + "yandex": ("BROWSER_PROXY_YANDEX", "YANDEX_PROXY_URL"), + "generic": ("SCRAPER_PROXY_URL", "AVITO_PROXY_URL"), } - return {src: url for src, url in raw.items() if url} - - -BROWSER_PROXY_MAP: dict[str, str] = _build_proxy_map() + primary, fallback = env_map.get(provider, env_map["generic"]) + return os.environ.get(primary) or os.environ.get(fallback) def _parse_proxy(proxy_url: str | None) -> dict[str, str] | None: @@ -181,22 +164,23 @@ def _parse_proxy(proxy_url: str | None) -> dict[str, str] | None: return result -# ── состояние браузера ───────────────────────────────────────────────────────── +# ── per-provider состояние браузеров ─────────────────────────────────────────── +# Каждый словарь keyed по provider (str). Заполняется ЛЕНИВО: инстанс конкретного +# поставщика поднимается на первом /fetch (или /login) к этому поставщику. +# +# _locks[provider] сериализует ВЕСЬ доступ к инстансу этого поставщика +# (launch + relaunch + fetch + login) → внутри поставщика строго ≤1 операция +# одновременно. Разные поставщики держат РАЗНЫЕ локи → работают параллельно. +_browsers: dict[str, object] = {} # provider → Browser +_browser_cms: dict[str, object] = {} # provider → AsyncCamoufox CM (для __aexit__) +_page_counters: dict[str, int] = {} # provider → страниц с момента launch'а +_locks: dict[str, asyncio.Lock] = {} # provider → Lock (весь доступ к инстансу) +_retry_tasks: dict[str, asyncio.Task[None]] = {} # provider → фоновая retry-задача -_browser: object | None = None # текущий экземпляр Browser -_browser_cm: object | None = None # AsyncCamoufox context manager (для __aexit__) -_page_counter: int = 0 # страниц с момента последнего (пере)запуска -# _launch_lock сериализует ТОЛЬКО launch/relaunch/recycle браузера (один __aenter__ -# за раз на общий глобал _browser — иначе два launch'а затрут друг друга). НЕ -# сериализует сами /fetch — это делает семафор ниже. -_launch_lock: asyncio.Lock | None = None -# _fetch_sem ограничивает число ПАРАЛЛЕЛЬНЫХ /fetch (и /login) до BROWSER_CONCURRENCY. -# Один браузер, N параллельных страниц. Заменяет старый _fetch_lock (сериализация 1-за-раз). -_fetch_sem: asyncio.Semaphore | None = None -# Число fetch/login в полёте прямо сейчас. Нужно чтобы recycle/crash-relaunch НЕ -# рвал общий браузер из-под живых параллельных страниц (teardown под сиблингами → -# каскад крашей). Relaunch выполняется только когда in-flight == 1 (мы единственные). -_inflight: int = 0 +# Guard на ленивое создание per-provider локов: setdefault на обычном dict из +# разных корутин гонок не даёт (нет await между read-modify-write), но держим +# явный guard на случай будущей сложной инициализации. Создаётся в _on_startup. +_locks_guard: asyncio.Lock | None = None # Resilience (#crash-storm): если прокси (mproxy.site) лежит, camoufox делает # eager public_ip(proxy) для geoip-fingerprint и бросает InvalidProxy. Раньше это @@ -205,29 +189,32 @@ _inflight: int = 0 # launch с backoff пока прокси не вернётся. proxy+geoip СОХРАНЕНЫ (анти-бан). _BROWSER_RETRY_MIN_S: float = 30.0 # стартовый интервал retry launch'а _BROWSER_RETRY_MAX_S: float = 60.0 # потолок backoff'а -_browser_retry_task: asyncio.Task[None] | None = None # handle фоновой retry-задачи - -# ── per-source pool state (Phase 1) ────────────────────────────────────────────── -# Параллельные структуры к одиночным глобалам выше, но keyed по proxy_url. Заполняются -# ЛЕНИВО (только при FEATURE_BROWSER_POOL_ENABLED и первом /fetch на данный proxy_url), -# поэтому OFF-путь их вообще не трогает. Один браузер+lock на каждый уникальный прокси. -_browser_pool: dict[str, object | None] = {} # proxy_url → Browser -_browser_cm_pool: dict[str, object | None] = {} # proxy_url → AsyncCamoufox CM -_page_counter_pool: dict[str, int] = {} # proxy_url → страниц с launch'а -_pool_locks: dict[str, asyncio.Lock] = {} # proxy_url → Lock (launch+fetch сериализация) -# Guard на создание per-proxy локов: setdefault на обычном dict из разных корутин -# гонок не даёт (нет await между read-modify-write), но держим явный guard на случай -# будущей сложной инициализации. Создаётся в _on_startup. -_pool_locks_guard: asyncio.Lock | None = None -async def _launch_browser() -> None: - """Запускает AsyncCamoufox и сохраняет browser + CM в модульных переменных.""" - global _browser, _browser_cm, _page_counter +async def _lock_for(provider: str) -> asyncio.Lock: + """Возвращает (создавая лениво под guard'ом) per-provider лок. + Лок сериализует launch+fetch+login на инстансе поставщика: один поставщик = + один браузер = ≤1 операция за раз. Guard защищает создание лока от гонки + параллельных корутин на первом запросе к новому поставщику. + """ + existing = _locks.get(provider) + if existing is not None: + return existing + assert _locks_guard is not None, "_locks_guard not initialised" + async with _locks_guard: + return _locks.setdefault(provider, asyncio.Lock()) + + +async def _launch_browser(provider: str) -> None: + """Запускает AsyncCamoufox для поставщика и кладёт в *_…[provider]. + + Использует _provider_proxy(provider) + geoip=True (фингерпринт привязан к гео + IP конкретного прокси — поэтому отдельный инстанс на каждого поставщика). + """ from camoufox.async_api import AsyncCamoufox - proxy = _parse_proxy(SCRAPER_PROXY_URL) + proxy = _parse_proxy(_provider_proxy(provider)) kwargs: dict[str, object] = { "headless": True, "os": "windows", @@ -239,381 +226,203 @@ async def _launch_browser() -> None: kwargs["proxy"] = proxy logger.info( - "tradein-browser: запуск AsyncCamoufox (proxy=%s, recycle_pages=%d)", + "tradein-browser[%s]: запуск AsyncCamoufox (proxy=%s, recycle_pages=%d)", + provider, proxy is not None, BROWSER_RECYCLE_PAGES, ) cm = AsyncCamoufox(**kwargs) # type: ignore[arg-type] browser = await cm.__aenter__() - _browser_cm = cm - _browser = browser - _page_counter = 0 - logger.info("tradein-browser: браузер запущен") + _browser_cms[provider] = cm + _browsers[provider] = browser + _page_counters[provider] = 0 + logger.info("tradein-browser[%s]: браузер запущен", provider) -async def _close_browser() -> None: - """Закрывает текущий экземпляр браузера.""" - global _browser, _browser_cm, _page_counter - - cm = _browser_cm +async def _close_browser(provider: str) -> None: + """Закрывает per-provider браузер и чистит состояние для поставщика.""" + cm = _browser_cms.get(provider) if cm is not None: try: await cm.__aexit__(None, None, None) # type: ignore[attr-defined] - logger.info("tradein-browser: браузер закрыт") + logger.info("tradein-browser[%s]: браузер закрыт", provider) except Exception as exc: - logger.warning("tradein-browser: ошибка при закрытии браузера: %s", exc) - finally: - _browser = None - _browser_cm = None - _page_counter = 0 + logger.warning( + "tradein-browser[%s]: ошибка при закрытии браузера: %s", + provider, + type(exc).__name__, + ) + _browsers.pop(provider, None) + _browser_cms.pop(provider, None) + _page_counters[provider] = 0 -async def _relaunch_browser() -> None: - """Закрывает текущий браузер и запускает новый (recycle / crash-recovery).""" - logger.info("tradein-browser: перезапуск браузера") - await _close_browser() - await _launch_browser() +async def _relaunch_browser(provider: str) -> None: + """Закрывает и заново запускает per-provider браузер (recycle / crash-recovery). + + Caller держит _locks[provider], поэтому параллельных страниц на этом инстансе + нет — teardown безопасен. + """ + logger.info("tradein-browser[%s]: перезапуск браузера", provider) + await _close_browser(provider) + await _launch_browser(provider) -async def _try_launch_browser() -> bool: - """Одна попытка launch'а с санитизированным логом фейла. +async def _try_launch_browser(provider: str) -> bool: + """Одна попытка launch'а поставщика с санитизированным логом фейла. НЕ пробрасывает исключение наружу — возвращает True/False. Критично для resilience: фейл launch'а (прокси лёг → InvalidProxy) не должен валить startup или фоновую задачу. Креды прокси НЕ логируются: camoufox светит ``user:pass@host`` в тексте своего исключения, поэтому мы логируем только - тип + санитизированное сообщение, без ``str(exc)``. + тип, без ``str(exc)``. При частичном фейле (CM создан, но __aenter__ не довёл) чистит состояние через _close_browser, чтобы следующий retry стартовал с чистого листа. """ try: - await _launch_browser() + await _launch_browser(provider) return True except Exception as exc: logger.warning( - "tradein-browser: browser launch failed: proxy unreachable " - "(exc_type=%s) — контейнер остаётся Up, retry в фоне", + "tradein-browser[%s]: browser launch failed: proxy unreachable " + "(exc_type=%s) — контейнер остаётся Up, retry на следующем запросе/в фоне", + provider, type(exc).__name__, ) - # Подчистить возможный частичный CM перед следующей попыткой. - await _close_browser() + await _close_browser(provider) return False -async def _ensure_browser() -> bool: - """Гарантирует поднятый браузер; одна lazy-попытка launch'а под launch-локом. +async def _ensure_browser(provider: str) -> bool: + """Гарантирует поднятый инстанс поставщика; одна lazy-попытка launch'а. - Идемпотентна и конкурентно-безопасна: несколько параллельных /fetch могут - вызвать её одновременно — лок сериализует launch, а double-check внутри лока - не даёт второму launch'у затереть уже поднятый первым браузер. + Caller держит _locks[provider] (весь доступ к инстансу сериализован), поэтому + здесь своей сериализации нет — двух одновременных launch'ей на один provider + быть не может. Returns: True если браузер доступен (был или поднялся), False если launch не удался (прокси лежит) — caller отдаёт 503. """ - if _browser is not None: + if _browsers.get(provider) is not None: return True - assert _launch_lock is not None, "_launch_lock not initialised" - async with _launch_lock: - if _browser is not None: - return True - await _try_launch_browser() - return _browser is not None + await _try_launch_browser(provider) + return _browsers.get(provider) is not None -async def _safe_relaunch(reason: str) -> None: - """Перезапуск браузера под launch-локом ТОЛЬКО когда мы единственный in-flight. +async def _browser_retry_loop(provider: str) -> None: + """Фоновый retry launch'а поставщика с backoff, пока прокси не вернётся. - Под конкуренцией recycle/crash-relaunch не должен рвать общий браузер из-под - живых параллельных страниц (teardown под сиблингами → каскад крашей). Поэтому: - relaunch выполняется лишь если ``_inflight <= 1`` (мы — единственный активный - запрос). Иначе relaunch ПРОПУСКАЕТСЯ — браузер переиспользуется, а recycle - случится позже, когда параллельная нагрузка спадёт. Crash-recovery при этом не - теряется: per-page краш изолирован (своя страница), а реально мёртвый браузер - поднимет _ensure_browser на следующем запросе. - - Сброс ``_page_counter = 0`` происходит АТОМАРНО с реальным relaunch'ем (под - launch-локом, только когда guard inflight<=1 пройден). Иначе под конкуренцией - N>1 несколько корутин видят counter >= порога, все откладывают relaunch (guard - inflight>1), но counter НИКОГДА не сбрасывается → recycle-check срабатывает на - каждом fetch вечно, а реальный recycle голодает. При отложенном relaunch'е - counter НЕ трогаем — следующий запрос (когда нагрузка спадёт) дожмёт recycle. - """ - global _page_counter - assert _launch_lock is not None, "_launch_lock not initialised" - async with _launch_lock: - if _inflight > 1: - logger.info( - "tradein-browser: relaunch (%s) отложен — %d параллельных запросов в полёте", - reason, - _inflight, - ) - return - logger.info("tradein-browser: relaunch (%s)", reason) - _page_counter = 0 - await _relaunch_browser() - - -async def _browser_retry_loop() -> None: - """Фоновый retry launch'а браузера с backoff, пока прокси не вернётся. - - Идемпотентна: если браузер уже поднят (_browser не None) — сразу выходит. - Иначе ретраит _try_launch_browser с возрастающим интервалом + Идемпотентна: если инстанс уже поднят — сразу выходит. Иначе ретраит + _try_launch_browser под _locks[provider] с возрастающим интервалом (_BROWSER_RETRY_MIN_S → _BROWSER_RETRY_MAX_S). Завершается после первого успешного launch'а. Отменяется в _on_cleanup (CancelledError пробрасывается). """ - if _browser is not None: + if _browsers.get(provider) is not None: return delay = _BROWSER_RETRY_MIN_S - while _browser is None: + while _browsers.get(provider) is None: await asyncio.sleep(delay) - # Launch-лок сериализует launch с in-flight lazy-launch и recycle — иначе - # два __aenter__ на один глобал _browser затрут друг друга. - assert _launch_lock is not None, "_launch_lock not initialised" - async with _launch_lock: - if _browser is not None: - return # handler уже поднял браузер — задача отработала - logger.info("tradein-browser: фоновый retry launch'а браузера") - launched = await _try_launch_browser() + lock = await _lock_for(provider) + async with lock: + if _browsers.get(provider) is not None: + return # handler уже поднял инстанс — задача отработала + logger.info("tradein-browser[%s]: фоновый retry launch'а браузера", provider) + launched = await _try_launch_browser(provider) if launched: - logger.info("tradein-browser: браузер поднят после retry, задача завершена") + logger.info( + "tradein-browser[%s]: браузер поднят после retry, задача завершена", provider + ) return delay = min(delay * 1.5, _BROWSER_RETRY_MAX_S) -def _start_retry_task(app: web.Application) -> None: - """Запускает (если ещё нет) фоновую retry-задачу и кладёт handle в app.""" - global _browser_retry_task - - existing = _browser_retry_task +def _start_retry_task(provider: str) -> None: + """Запускает (если ещё нет) фоновую retry-задачу для поставщика.""" + existing = _retry_tasks.get(provider) if existing is not None and not existing.done(): return # задача уже бежит — не плодим дубли - - task = asyncio.create_task(_browser_retry_loop()) - _browser_retry_task = task - app["browser_retry_task"] = task - - -# ── per-source pool helpers (Phase 1) ──────────────────────────────────────────── - - -async def _get_pool_lock(proxy_url: str) -> asyncio.Lock: - """Возвращает (создавая лениво под guard'ом) per-proxy лок для proxy_url. - - Лок сериализует launch+fetch на конкретном браузере: один прокси = один - браузер = N последовательных страниц. Guard защищает создание лока от гонки - параллельных корутин на первом запросе к новому проксе. - """ - existing = _pool_locks.get(proxy_url) - if existing is not None: - return existing - assert _pool_locks_guard is not None, "_pool_locks_guard not initialised" - async with _pool_locks_guard: - return _pool_locks.setdefault(proxy_url, asyncio.Lock()) - - -async def _launch_browser_for_proxy(proxy_url: str) -> None: - """Запускает AsyncCamoufox для конкретного proxy_url и кладёт в *_pool[proxy_url]. - - Зеркало _launch_browser, но параметризовано прокси и пишет в pool-словари - (keyed по proxy_url), а не в одиночные глобалы. Пустой proxy_url → _parse_proxy - вернёт None → direct (совпадает с поведением одиночного браузера без прокси). - """ - from camoufox.async_api import AsyncCamoufox - - proxy = _parse_proxy(proxy_url) - kwargs: dict[str, object] = { - "headless": True, - "os": "windows", - "locale": "ru-RU", - "geoip": True, - "humanize": True, - } - if proxy is not None: - kwargs["proxy"] = proxy - - logger.info( - "tradein-browser[pool]: запуск AsyncCamoufox (proxy=%s, recycle_pages=%d)", - proxy is not None, - BROWSER_RECYCLE_PAGES, - ) - cm = AsyncCamoufox(**kwargs) # type: ignore[arg-type] - browser = await cm.__aenter__() - _browser_cm_pool[proxy_url] = cm - _browser_pool[proxy_url] = browser - _page_counter_pool[proxy_url] = 0 - logger.info("tradein-browser[pool]: браузер запущен (proxy=%s)", proxy is not None) - - -async def _close_pool_browser(proxy_url: str) -> None: - """Закрывает per-proxy браузер и чистит pool-состояние для proxy_url.""" - cm = _browser_cm_pool.get(proxy_url) - if cm is not None: - try: - await cm.__aexit__(None, None, None) # type: ignore[attr-defined] - logger.info("tradein-browser[pool]: браузер закрыт") - except Exception as exc: - logger.warning( - "tradein-browser[pool]: ошибка при закрытии браузера: %s", type(exc).__name__ - ) - _browser_pool[proxy_url] = None - _browser_cm_pool[proxy_url] = None - _page_counter_pool[proxy_url] = 0 - - -async def _relaunch_pool_browser(proxy_url: str) -> None: - """Закрывает и заново запускает per-proxy браузер (recycle / crash-recovery).""" - logger.info("tradein-browser[pool]: перезапуск браузера") - await _close_pool_browser(proxy_url) - await _launch_browser_for_proxy(proxy_url) - - -async def _get_or_launch_pool_browser(proxy_url: str) -> object | None: - """Возвращает поднятый per-proxy браузер, лениво запуская при необходимости. - - Caller держит соответствующий _pool_locks[proxy_url], поэтому здесь нет своей - сериализации. При фейле launch'а (прокси лежит) лог санитизирован (только - type(exc).__name__, без str(exc) — camoufox светит user:pass@host в тексте) и - возвращается None → caller отдаёт 503 (как одиночный путь при недоступном прокси). - """ - existing = _browser_pool.get(proxy_url) - if existing is not None: - return existing - try: - await _launch_browser_for_proxy(proxy_url) - except Exception as exc: - logger.warning( - "tradein-browser[pool]: browser launch failed: proxy unreachable " - "(exc_type=%s) — отдаём 503, retry на следующем запросе", - type(exc).__name__, - ) - await _close_pool_browser(proxy_url) - return None - return _browser_pool.get(proxy_url) - - -async def _do_fetch_pool(url: str, proxy_url: str) -> str: - """Pool-вариант _do_fetch: при краше браузера — relaunch и один retry. - - Caller держит _pool_locks[proxy_url] (launch+fetch сериализованы на этот прокси), - поэтому relaunch безопасен — нет параллельных страниц на этом браузере. - """ - try: - return await _fetch_once_pool(url, proxy_url) - except Exception as exc: - if _is_browser_crash(exc): - logger.warning( - "tradein-browser[pool]: краш браузера (%s), перезапуск + retry: %s", - type(exc).__name__, - url, - ) - await _relaunch_pool_browser(proxy_url) - browser = _browser_pool.get(proxy_url) - if browser is None: - raise - return await _fetch_once_pool(url, proxy_url) - raise - - -async def _fetch_once_pool(url: str, proxy_url: str) -> str: - """Открывает страницу на per-proxy браузере, грузит URL, возвращает HTML. - - Caller держит _pool_locks[proxy_url], поэтому страницы на этом браузере не - параллелятся — recycle через _relaunch_pool_browser безопасен прямо здесь. - """ - browser = _browser_pool.get(proxy_url) - assert browser is not None, "pool browser not launched" - - page = await browser.new_page() # type: ignore[attr-defined] - try: - await _apply_resource_block(page) - await page.goto(url, timeout=BROWSER_NAV_TIMEOUT_MS, wait_until="domcontentloaded") # type: ignore[attr-defined] - if BROWSER_WAIT_MS > 0: - await page.wait_for_timeout(BROWSER_WAIT_MS) # type: ignore[attr-defined] - html: str = await page.content() # type: ignore[attr-defined] - finally: - await page.close() # type: ignore[attr-defined] - - _page_counter_pool[proxy_url] = _page_counter_pool.get(proxy_url, 0) + 1 - if _page_counter_pool[proxy_url] >= BROWSER_RECYCLE_PAGES: - logger.info( - "tradein-browser[pool]: recycle threshold (%d) достигнут, перезапуск браузера", - BROWSER_RECYCLE_PAGES, - ) - await _relaunch_pool_browser(proxy_url) - - return html + _retry_tasks[provider] = asyncio.create_task(_browser_retry_loop(provider)) # ── aiohttp lifecycle hooks ──────────────────────────────────────────────────── async def _on_startup(app: web.Application) -> None: - global _launch_lock, _fetch_sem, _pool_locks_guard - _launch_lock = asyncio.Lock() - _fetch_sem = asyncio.Semaphore(BROWSER_CONCURRENCY) - _pool_locks_guard = asyncio.Lock() - if FEATURE_BROWSER_POOL_ENABLED: - logger.info( - "tradein-browser: per-source proxy pool ВКЛЮЧЁН (sources=%s)", - sorted(BROWSER_PROXY_MAP.keys()), - ) + global _locks_guard + _locks_guard = asyncio.Lock() logger.info( - "tradein-browser: BROWSER_CONCURRENCY=%d (параллельных /fetch)", BROWSER_CONCURRENCY + "tradein-browser: per-provider concurrency — параллель между %s, ≤1 на поставщика", + list(PROVIDERS), ) - # Launch НЕ фатален: если прокси лежит, camoufox бросит InvalidProxy. Раньше - # это валило startup → exit(1) → restart-loop. Теперь ловим, остаёмся Up и - # ретраим в фоне. proxy+geoip сохранены — никакого direct-fetch fallback. - if not await _try_launch_browser(): - logger.warning( - "tradein-browser: старт без браузера (launch failed) — запуск фонового retry" - ) - _start_retry_task(app) + # Eager-launch каждого поставщика. Launch НЕ фатален: если прокси лежит, + # camoufox бросит InvalidProxy. Раньше это валило startup → exit(1) → + # restart-loop. Теперь ловим, остаёмся Up и ретраим в фоне. Поставщики + # независимы: лёгший avito-прокси не мешает подняться cian/yandex. + for provider in PROVIDERS: + lock = await _lock_for(provider) + async with lock: + launched = await _try_launch_browser(provider) + if not launched: + logger.warning( + "tradein-browser[%s]: старт без браузера (launch failed) — запуск фонового retry", + provider, + ) + _start_retry_task(provider) async def _on_cleanup(app: web.Application) -> None: - global _browser_retry_task + # Отменяем все фоновые retry-задачи. + for provider, task in list(_retry_tasks.items()): + if task is not None and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + except Exception as exc: + logger.warning( + "tradein-browser[%s]: ошибка при отмене retry-задачи: %s", + provider, + type(exc).__name__, + ) + _retry_tasks.clear() - task = _browser_retry_task - if task is not None and not task.done(): - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - except Exception as exc: - logger.warning( - "tradein-browser: ошибка при отмене retry-задачи: %s", type(exc).__name__ - ) - _browser_retry_task = None - await _close_browser() - # Закрыть все per-source pool браузеры (если pool использовался). - for proxy_url in list(_browser_pool.keys()): - if _browser_pool.get(proxy_url) is not None: - await _close_pool_browser(proxy_url) + # Закрываем все поднятые инстансы. + for provider in list(_browsers.keys()): + await _close_browser(provider) # ── handlers ─────────────────────────────────────────────────────────────────── +def _resolve_provider(body: dict, url: str) -> str: + """Выбирает поставщика: явный body["provider"] (если валиден) иначе по host URL.""" + explicit = body.get("provider") + if isinstance(explicit, str) and explicit in PROVIDERS: + return explicit + return _provider_for_url(url) + + async def health_handler(request: web.Request) -> web.Response: # Всегда 200 — иначе compose healthcheck (если появится) убьёт контейнер, - # хотя процесс жив и ретраит браузер. browser_ready — для observability: - # видно лежит ли прокси, не теряя контейнер. Текущий prod compose НЕ имеет - # healthcheck на browser (depends_on: condition: service_started), так что - # 200-без-браузера никого не рестартит. - return web.json_response({"status": "ok", "browser_ready": _browser is not None}) + # хотя процесс жив и ретраит браузеры. browsers — per-provider readiness для + # observability: видно какой поставщик лежит, не теряя контейнер. Текущий prod + # compose НЕ имеет healthcheck на browser (depends_on: service_started). + browsers = {p: _browsers.get(p) is not None for p in PROVIDERS} + return web.json_response({"status": "ok", "browsers": browsers}) async def fetch_handler(request: web.Request) -> web.Response: """POST /fetch {"url": "..."} → {"html": "..."} - Выполняет навигацию в браузере и возвращает полный HTML страницы. - До BROWSER_CONCURRENCY запросов обрабатываются параллельно (семафор), у - каждого — своя страница (изоляция крашей). + Выполняет навигацию в браузере поставщика и возвращает полный HTML страницы. + Поставщик определяется по host URL (или явному body["provider"]). Берётся + ТОЛЬКО лок этого поставщика → разные поставщики работают параллельно, внутри + поставщика — строго ≤1 fetch. """ try: body = await request.json() @@ -624,73 +433,30 @@ async def fetch_handler(request: web.Request) -> web.Response: if not url: return web.json_response({"error": "missing 'url' field"}, status=400) - if FEATURE_BROWSER_POOL_ENABLED: - source: str | None = body.get("source") - return await _fetch_via_pool(url, source) - return await _fetch_via_single(url) + provider = _resolve_provider(body, url) - -async def _fetch_via_single(url: str) -> web.Response: - """Текущий (OFF-флаг) путь: один глобальный браузер + семафор. Без изменений.""" - global _inflight - - assert _fetch_sem is not None, "_fetch_sem not initialised" - - async with _fetch_sem: - # Браузер мог не подняться на старте (прокси лежал). Одна lazy-попытка - # под launch-локом — вдруг прокси уже вернулся. НИКОГДА не фетчим без - # прокси/браузера: если всё ещё None → 503 (caller словит как - # HTTPStatusError и обработает как failed fetch, а не как crash). - if not await _ensure_browser(): - logger.warning("tradein-browser: /fetch 503 — браузер недоступен (proxy may be down)") - return web.json_response( - {"error": "browser unavailable (proxy may be down)"}, status=503 - ) - - _inflight += 1 - try: - html = await _do_fetch(url) - except Exception as exc: - logger.error( - "tradein-browser: fetch error url=%r: %s: %s", url, type(exc).__name__, exc - ) - return web.json_response({"error": f"{type(exc).__name__}: {exc}"}, status=500) - finally: - _inflight -= 1 - - return web.json_response({"html": html}) - - -async def _fetch_via_pool(url: str, source: str | None) -> web.Response: - """Per-source путь (ON-флаг): роутит на отдельный браузер+прокси по source. - - Выбор прокси: BROWSER_PROXY_MAP[source] → fallback на avito-прокси → "" (direct, - как одиночный путь при пустом env). Один lock на proxy_url сериализует launch+fetch - на этот браузер — источники с разными проксями работают независимо, не клиня друг - друга. Lazy-launch на первом запросе к данному проксе; при недоступном прокси → 503. - """ - proxy_url = BROWSER_PROXY_MAP.get(source or "avito") or BROWSER_PROXY_MAP.get("avito") - if not proxy_url: - # Ни одного прокси не сконфигурировано → direct (как одиночный путь при пустом env). - proxy_url = "" - - lock = await _get_pool_lock(proxy_url) + lock = await _lock_for(provider) async with lock: - browser = await _get_or_launch_pool_browser(proxy_url) - if browser is None: + # Браузер мог не подняться на старте (прокси лежал). Одна lazy-попытка — + # вдруг прокси уже вернулся. НИКОГДА не фетчим без прокси/браузера: если + # всё ещё None → 503 (caller словит как HTTPStatusError → failed fetch, не + # crash). + if not await _ensure_browser(provider): logger.warning( - "tradein-browser[pool]: /fetch 503 — браузер недоступен (proxy may be down)" + "tradein-browser[%s]: /fetch 503 — браузер недоступен (proxy may be down)", + provider, ) return web.json_response( {"error": "browser unavailable (proxy may be down)"}, status=503 ) + try: - html = await _do_fetch_pool(url, proxy_url) + html = await _do_fetch(provider, url) except Exception as exc: logger.error( - "tradein-browser[pool]: fetch error url=%r source=%r: %s: %s", + "tradein-browser[%s]: fetch error url=%r: %s: %s", + provider, url, - source, type(exc).__name__, exc, ) @@ -732,38 +498,36 @@ async def _apply_resource_block(page: object) -> None: await page.route("**/*", _route_block) # type: ignore[attr-defined] -async def _do_fetch(url: str) -> str: - """Одна попытка навигации; при краше браузера — перезапускает и повторяет один раз. +async def _do_fetch(provider: str, url: str) -> str: + """Одна попытка навигации; при краше браузера — relaunch и один retry. - Per-page краш изолирован (своя страница на запрос); _safe_relaunch рвёт общий - браузер лишь когда нет параллельных in-flight запросов, иначе пропускает (мёртвый - браузер поднимет _ensure_browser на следующем запросе). + Caller держит _locks[provider] (нет параллельных страниц на этом инстансе), + поэтому relaunch безопасен — никого не рвёт из-под живых страниц. """ try: - return await _fetch_once(url) + return await _fetch_once(provider, url) except Exception as exc: if _is_browser_crash(exc): logger.warning( - "tradein-browser: краш браузера (%s), перезапуск + retry: %s", + "tradein-browser[%s]: краш браузера (%s), перезапуск + retry: %s", + provider, type(exc).__name__, url, ) - await _safe_relaunch("crash-recovery") - if not await _ensure_browser(): + await _relaunch_browser(provider) + if _browsers.get(provider) is None: raise - return await _fetch_once(url) + return await _fetch_once(provider, url) raise -async def _fetch_once(url: str) -> str: +async def _fetch_once(provider: str, url: str) -> str: """Открывает СОБСТВЕННУЮ страницу, переходит по URL, ждёт JS, возвращает HTML. - Каждый параллельный /fetch держит свою page (никакого шеринга — общая page под - конкуренцией ломает навигацию), закрывает её в finally. + Caller держит _locks[provider], поэтому страницы на этом инстансе не + параллелятся — recycle через _relaunch_browser безопасен прямо здесь. """ - global _page_counter - - browser = _browser + browser = _browsers.get(provider) assert browser is not None, "browser not launched" page = await browser.new_page() # type: ignore[attr-defined] @@ -776,15 +540,21 @@ async def _fetch_once(url: str) -> str: finally: await page.close() # type: ignore[attr-defined] - _page_counter += 1 - logger.debug("tradein-browser: fetch OK url=%r pages_since_launch=%d", url, _page_counter) + _page_counters[provider] = _page_counters.get(provider, 0) + 1 + logger.debug( + "tradein-browser[%s]: fetch OK url=%r pages_since_launch=%d", + provider, + url, + _page_counters[provider], + ) - if _page_counter >= BROWSER_RECYCLE_PAGES: + if _page_counters[provider] >= BROWSER_RECYCLE_PAGES: logger.info( - "tradein-browser: recycle threshold (%d) достигнут, перезапуск браузера", + "tradein-browser[%s]: recycle threshold (%d) достигнут, перезапуск браузера", + provider, BROWSER_RECYCLE_PAGES, ) - await _safe_relaunch("recycle") + await _relaunch_browser(provider) return html @@ -819,10 +589,9 @@ async def login_handler(request: web.Request) -> web.Response: Выполняет вход на указанную страницу (email + пароль), ждёт появления success_cookie, возвращает полный список cookies браузерного контекста. - Делит тот же семафор/браузер что и /fetch (своя страница на запрос). + Делит тот же per-provider браузер/лок что и /fetch (поставщик по host URL + или явному body["provider"]). """ - global _inflight - try: body = await request.json() except Exception: @@ -833,52 +602,58 @@ async def login_handler(request: web.Request) -> web.Response: if missing: return web.json_response({"error": f"missing required fields: {missing}"}, status=400) - assert _fetch_sem is not None, "_fetch_sem not initialised" + provider = _resolve_provider(body, body["url"]) - async with _fetch_sem: + lock = await _lock_for(provider) + async with lock: # Та же resilience что и в /fetch: браузер мог не подняться (прокси лежал). - # Lazy-попытка под launch-локом, иначе 503 — без прокси не логинимся. - if not await _ensure_browser(): - logger.warning("tradein-browser: /login 503 — браузер недоступен (proxy may be down)") + # Lazy-попытка, иначе 503 — без прокси не логинимся. + if not await _ensure_browser(provider): + logger.warning( + "tradein-browser[%s]: /login 503 — браузер недоступен (proxy may be down)", + provider, + ) return web.json_response( {"error": "browser unavailable (proxy may be down)"}, status=503 ) - _inflight += 1 try: - cookies = await _do_login(body) + cookies = await _do_login(provider, body) except LoginError as exc: return web.json_response(exc.payload, status=502) except Exception as exc: logger.error( - "tradein-browser: login error url=%r: %s: %s", + "tradein-browser[%s]: login error url=%r: %s: %s", + provider, body.get("url"), type(exc).__name__, exc, ) return web.json_response({"error": f"{type(exc).__name__}: {exc}"}, status=502) - finally: - _inflight -= 1 return web.json_response({"cookies": cookies}) -async def _do_login(params: dict) -> list[dict]: - """Одна попытка логина; при краше браузера — перезапускает и повторяет.""" +async def _do_login(provider: str, params: dict) -> list[dict]: + """Одна попытка логина; при краше браузера — relaunch и повтор. + + Caller держит _locks[provider], поэтому relaunch безопасен. + """ try: - return await _login_once(params) + return await _login_once(provider, params) except LoginError: raise # диагностический payload идёт напрямую в хендлер except Exception as exc: if _is_browser_crash(exc): logger.warning( - "tradein-browser: краш браузера (%s) при логине, перезапуск + retry", + "tradein-browser[%s]: краш браузера (%s) при логине, перезапуск + retry", + provider, type(exc).__name__, ) - await _safe_relaunch("login-crash-recovery") - if not await _ensure_browser(): + await _relaunch_browser(provider) + if _browsers.get(provider) is None: raise - return await _login_once(params) + return await _login_once(provider, params) raise @@ -896,16 +671,14 @@ async def _wait_cookie(page: object, name: str, tries: int, interval_ms: int) -> return False -async def _login_once(params: dict) -> list[dict]: +async def _login_once(provider: str, params: dict) -> list[dict]: """Открывает страницу, заполняет форму, нажимает submit, ждёт success_cookie. Поддерживает список pre_click_selectors (клик по очереди, каждый non-fatal) и 2-шаговый submit Cian: после первого сабмита появляется экран «Введите пароль» — дозаполняем password и кликаем повторно (макс. 2 попытки). """ - global _page_counter - - browser = _browser + browser = _browsers.get(provider) assert browser is not None, "browser not launched" url: str = params["url"] @@ -965,14 +738,16 @@ async def _login_once(params: dict) -> list[dict]: break logger.info( - "tradein-browser: login success_cookie=%r present=%s", + "tradein-browser[%s]: login success_cookie=%r present=%s", + provider, success_cookie, cookie_found, ) cookies: list[dict] = await page.context.cookies() # type: ignore[attr-defined] logger.info( - "tradein-browser: login OK url=%r email_selector=%r cookie_count=%d", + "tradein-browser[%s]: login OK url=%r email_selector=%r cookie_count=%d", + provider, url, email_selector, len(cookies), @@ -997,13 +772,14 @@ async def _login_once(params: dict) -> list[dict]: finally: await page.close() # type: ignore[attr-defined] - _page_counter += 1 - if _page_counter >= BROWSER_RECYCLE_PAGES: + _page_counters[provider] = _page_counters.get(provider, 0) + 1 + if _page_counters[provider] >= BROWSER_RECYCLE_PAGES: logger.info( - "tradein-browser: recycle threshold (%d) достигнут после login, перезапуск браузера", + "tradein-browser[%s]: recycle threshold (%d) достигнут после login, перезапуск", + provider, BROWSER_RECYCLE_PAGES, ) - await _safe_relaunch("recycle-after-login") + await _relaunch_browser(provider) return cookies diff --git a/tradein-mvp/browser/test_server.py b/tradein-mvp/browser/test_server.py new file mode 100644 index 00000000..39243537 --- /dev/null +++ b/tradein-mvp/browser/test_server.py @@ -0,0 +1,274 @@ +"""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" + + +# ── _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"{provider}" + + +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) + assert probe.global_peak == 3, "три разных поставщика должны фетчить конкурентно" + # У каждого поставщика по одному запросу — внутренний 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 diff --git a/tradein-mvp/browser/test_server_smoke.py b/tradein-mvp/browser/test_server_smoke.py index 40896b54..f0a40345 100644 --- a/tradein-mvp/browser/test_server_smoke.py +++ b/tradein-mvp/browser/test_server_smoke.py @@ -2,9 +2,13 @@ Полный браузер (camoufox subprocess) юнит-тестить непрактично, поэтому здесь только смоук: (1) модуль импортируется; (2) /fetch и /login отдают 503 когда -браузер не поднят (прокси лежит) — критичный resilience-инвариант, защищающий -от crash-storm. camoufox в окружении теста НЕ нужен: launch замокан no-op'ом, -имитируя «прокси всё ещё недоступен». +браузер поставщика не поднят (прокси лежит) — критичный resilience-инвариант, +защищающий от crash-storm. camoufox в окружении теста НЕ нужен: launch замокан +no-op'ом, имитируя «прокси всё ещё недоступен». + +Per-provider модель (#1793): состояние keyed по provider (avito/cian/yandex/ +generic), доступ к инстансу сериализован _locks[provider]; разные поставщики +работают параллельно. Конкурентность проверяется в test_server.py. Запуск (из tradein-mvp/browser/):: @@ -38,6 +42,17 @@ 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()) + + def test_module_imports() -> None: """Смоук: модуль импортируется и экспонирует ключевые хендлеры.""" assert callable(server.fetch_handler) @@ -45,40 +60,44 @@ def test_module_imports() -> None: assert callable(server.health_handler) -def test_health_reports_browser_not_ready(monkeypatch: pytest.MonkeyPatch) -> None: - """/health всегда 200, browser_ready=False когда браузер не поднят.""" - monkeypatch.setattr(server, "_browser", None) +def test_health_reports_browsers_not_ready() -> None: + """/health всегда 200, browsers[]=False когда инстансы не подняты.""" request = make_mocked_request("GET", "/health") response = asyncio.run(server.health_handler(request)) assert response.status == 200 body = _json_body(response) assert body["status"] == "ok" - assert body["browser_ready"] is False + assert body["browsers"] == {"avito": False, "cian": False, "yandex": False, "generic": False} + + +def test_health_reports_partial_readiness() -> None: + """/health отражает per-provider readiness: один поднят, остальные нет.""" + server._browsers["avito"] = object() + request = make_mocked_request("GET", "/health") + response = asyncio.run(server.health_handler(request)) + body = _json_body(response) + assert body["browsers"]["avito"] is True + assert body["browsers"]["cian"] is False def test_fetch_returns_503_when_browser_unavailable( monkeypatch: pytest.MonkeyPatch, ) -> None: - """/fetch отдаёт 503 (не 500/краш) когда браузер не поднят и launch не удался. + """/fetch отдаёт 503 (не 500/краш) когда инстанс поставщика не поднят. - Имитирует «прокси лежит»: _browser=None + _ensure_browser no-op, - оставляющий None. Никакого реального camoufox-launch'а / direct-fetch. + Имитирует «прокси лежит»: _ensure_browser no-op, оставляющий инстанс None. + Никакого реального camoufox-launch'а / direct-fetch. """ - monkeypatch.setattr(server, "_browser", None) - monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4)) - async def _no_ensure() -> bool: + async def _no_ensure(provider: str) -> bool: return False # прокси всё ещё недоступен → браузер остаётся None monkeypatch.setattr(server, "_ensure_browser", _no_ensure) - request = make_mocked_request( - "POST", "/fetch", payload=b'{"url": "https://example.com"}' - ) + request = make_mocked_request("POST", "/fetch") async def _run() -> Any: - # make_mocked_request не парсит тело — подменяем .json() на нужный payload. - request.json = lambda: _coro({"url": "https://example.com"}) # type: ignore[method-assign] + request.json = lambda: _coro({"url": "https://www.avito.ru/x"}) # type: ignore[method-assign] return await server.fetch_handler(request) response = asyncio.run(_run()) @@ -90,17 +109,15 @@ def test_fetch_returns_503_when_browser_unavailable( def test_login_returns_503_when_browser_unavailable( monkeypatch: pytest.MonkeyPatch, ) -> None: - """/login тоже отдаёт 503 когда браузер не поднят (без credential-логирования).""" - monkeypatch.setattr(server, "_browser", None) - monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4)) + """/login тоже отдаёт 503 когда инстанс не поднят (без credential-логирования).""" - async def _no_ensure() -> bool: + async def _no_ensure(provider: str) -> bool: return False monkeypatch.setattr(server, "_ensure_browser", _no_ensure) login_body = { - "url": "https://cian.example/login", + "url": "https://www.cian.ru/login", "email": "u@example.com", "password": "secret", "email_selector": "#email", @@ -119,54 +136,7 @@ def test_login_returns_503_when_browser_unavailable( assert "browser unavailable" in body["error"] -# ── BROWSER_CONCURRENCY config parse ──────────────────────────────────────────── - - -def test_concurrency_default_is_4() -> None: - """BROWSER_CONCURRENCY не задан → дефолт 4.""" - assert server._parse_concurrency(None) == 4 - - -def test_concurrency_env_override() -> None: - """Валидное число из env применяется как есть.""" - assert server._parse_concurrency("8") == 8 - assert server._parse_concurrency("1") == 1 - - -def test_concurrency_clamps_to_min_1() -> None: - """0 / отрицательное зажимается до >=1 (семафор с 0 завесил бы сервис).""" - assert server._parse_concurrency("0") == 1 - assert server._parse_concurrency("-5") == 1 - - -def test_concurrency_garbage_falls_back_to_4() -> None: - """Нечисловой мусор → дефолт 4 (кривой конфиг не валит сервис).""" - assert server._parse_concurrency("abc") == 4 - assert server._parse_concurrency("") == 4 - - -def test_startup_creates_semaphore_with_concurrency( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """_on_startup создаёт _fetch_sem с ёмкостью BROWSER_CONCURRENCY + launch-лок.""" - monkeypatch.setattr(server, "BROWSER_CONCURRENCY", 3) - monkeypatch.setattr(server, "_fetch_sem", None) - monkeypatch.setattr(server, "_launch_lock", None) - - async def _no_launch() -> bool: - return True # не запускаем реальный camoufox в тесте - - monkeypatch.setattr(server, "_try_launch_browser", _no_launch) - - app = server.web.Application() - asyncio.run(server._on_startup(app)) - - assert isinstance(server._fetch_sem, asyncio.Semaphore) - assert server._fetch_sem._value == 3 # стартовая ёмкость == concurrency - assert isinstance(server._launch_lock, asyncio.Lock) - - -# ── semaphore limits concurrency + per-page isolation ──────────────────────────── +# ── per-provider state isolation ───────────────────────────────────────────────── class _FakePage: @@ -176,13 +146,9 @@ class _FakePage: self._tracker = tracker async def route(self, pattern: str, handler: Any) -> None: - # _apply_resource_block навешивает route-interception перед goto (perf: - # блокировка image/media/font). Заглушка no-op — поведение route проверяется - # отдельно в test_server_resource_block.py. return None async def goto(self, url: str, **kwargs: Any) -> None: - # Внутри goto держим «активную» страницу — здесь меряем параллелизм. self._tracker.active += 1 self._tracker.peak = max(self._tracker.peak, self._tracker.active) await asyncio.sleep(0.02) @@ -215,47 +181,19 @@ class _Tracker: self.peak = 0 -def _make_fetch_request() -> Any: +def _make_fetch_request(url: str = "https://www.avito.ru/x") -> Any: request = make_mocked_request("POST", "/fetch") - request.json = lambda: _coro({"url": "https://example.com"}) # type: ignore[method-assign] + request.json = lambda: _coro({"url": url}) # type: ignore[method-assign] return request -def test_semaphore_limits_concurrency_to_n(monkeypatch: pytest.MonkeyPatch) -> None: - """Семафор(N) не даёт больше N одновременных goto, даже при M>N запросах.""" - tracker = _Tracker() - browser = _FakeBrowser(tracker) - monkeypatch.setattr(server, "_browser", browser) - monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(2)) - monkeypatch.setattr(server, "_inflight", 0) - # recycle отключаем — высокий порог, чтобы relaunch не вмешивался. - monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) - - async def _ensure() -> bool: - return True - - monkeypatch.setattr(server, "_ensure_browser", _ensure) - - async def _run() -> None: - await asyncio.gather(*[server.fetch_handler(_make_fetch_request()) for _ in range(6)]) - - asyncio.run(_run()) - - assert tracker.peak <= 2, f"параллелизм {tracker.peak} превысил лимит семафора 2" - assert tracker.opened == 6 - assert tracker.closed == 6 # каждая страница своя, закрыта в finally - - def test_fetch_opens_and_closes_own_page(monkeypatch: pytest.MonkeyPatch) -> None: """Один /fetch открывает ровно одну собственную страницу и закрывает её.""" tracker = _Tracker() - browser = _FakeBrowser(tracker) - monkeypatch.setattr(server, "_browser", browser) - monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4)) - monkeypatch.setattr(server, "_inflight", 0) + server._browsers["avito"] = _FakeBrowser(tracker) monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) - async def _ensure() -> bool: + async def _ensure(provider: str) -> bool: return True monkeypatch.setattr(server, "_ensure_browser", _ensure) @@ -267,82 +205,31 @@ def test_fetch_opens_and_closes_own_page(monkeypatch: pytest.MonkeyPatch) -> Non assert tracker.closed == 1 -def test_ensure_browser_serialized_by_launch_lock( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """_ensure_browser сериализует launch: при N параллельных вызовах launch один раз.""" - monkeypatch.setattr(server, "_browser", None) - monkeypatch.setattr(server, "_launch_lock", asyncio.Lock()) - calls = {"n": 0} +# ── _relaunch resets per-provider page counter (recycle) ───────────────────────── - async def _launch_once() -> bool: - calls["n"] += 1 - await asyncio.sleep(0.01) - # имитируем успешный launch — выставляем браузер - monkeypatch.setattr(server, "_browser", object()) + +def test_fetch_recycles_after_threshold(monkeypatch: pytest.MonkeyPatch) -> None: + """Достижение BROWSER_RECYCLE_PAGES → _relaunch_browser + сброс счётчика.""" + tracker = _Tracker() + server._browsers["avito"] = _FakeBrowser(tracker) + server._page_counters["avito"] = 0 + monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 1) + + relaunched = {"n": 0} + + async def _fake_relaunch(provider: str) -> None: + relaunched["n"] += 1 + server._page_counters[provider] = 0 + + async def _ensure(provider: str) -> bool: return True - monkeypatch.setattr(server, "_try_launch_browser", _launch_once) - - async def _run() -> list[bool]: - return await asyncio.gather(*[server._ensure_browser() for _ in range(5)]) - - results = asyncio.run(_run()) - assert all(results) - assert calls["n"] == 1, "launch должен произойти ровно один раз (double-check под локом)" - - -# ── _safe_relaunch resets _page_counter (recycle starvation fix) ───────────────── - - -def test_safe_relaunch_resets_page_counter_when_proceeds( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """_safe_relaunch сбрасывает _page_counter=0 когда relaunch проходит (inflight<=1). - - Регрессия: без сброса под конкуренцией counter копился вечно (>= порога на - каждом fetch) → recycle голодал. Сброс атомарен с relaunch'ем под локом. - """ - monkeypatch.setattr(server, "_launch_lock", asyncio.Lock()) - monkeypatch.setattr(server, "_inflight", 1) # мы единственный in-flight - monkeypatch.setattr(server, "_page_counter", 999) - - relaunched = {"n": 0} - - async def _fake_relaunch() -> None: - relaunched["n"] += 1 - monkeypatch.setattr(server, "_relaunch_browser", _fake_relaunch) + monkeypatch.setattr(server, "_ensure_browser", _ensure) - asyncio.run(server._safe_relaunch("recycle")) - - assert relaunched["n"] == 1, "relaunch должен выполниться при inflight<=1" - assert server._page_counter == 0, "counter должен сброситься при реальном relaunch" - - -def test_safe_relaunch_keeps_page_counter_when_deferred( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """_safe_relaunch НЕ трогает _page_counter когда relaunch отложен (inflight>1). - - Отложенный relaunch оставляет counter, чтобы следующий запрос (когда нагрузка - спадёт) дожал recycle — иначе recycle потерялся бы навсегда. - """ - monkeypatch.setattr(server, "_launch_lock", asyncio.Lock()) - monkeypatch.setattr(server, "_inflight", 3) # есть параллельные запросы - monkeypatch.setattr(server, "_page_counter", 42) - - relaunched = {"n": 0} - - async def _fake_relaunch() -> None: - relaunched["n"] += 1 - - monkeypatch.setattr(server, "_relaunch_browser", _fake_relaunch) - - asyncio.run(server._safe_relaunch("recycle")) - - assert relaunched["n"] == 0, "relaunch должен быть отложен при inflight>1" - assert server._page_counter == 42, "counter НЕ трогаем при отложенном relaunch" + asyncio.run(server.fetch_handler(_make_fetch_request())) + assert relaunched["n"] == 1, "recycle должен сработать на пороге" + assert server._page_counters["avito"] == 0 # ── _login_once closes page on CancelledError (BaseException leak fix) ──────────── @@ -353,7 +240,7 @@ class _CancellingLoginPage: def __init__(self, tracker: "_Tracker") -> None: self._tracker = tracker - self.url = "https://cian.example/login" + self.url = "https://www.cian.ru/login" async def goto(self, url: str, **kwargs: Any) -> None: return None @@ -362,7 +249,6 @@ class _CancellingLoginPage: return None async def fill(self, selector: str, value: str, **kwargs: Any) -> None: - # Отмена таска (BaseException, не Exception) — обходит except Exception. raise asyncio.CancelledError() async def close(self) -> None: @@ -381,17 +267,12 @@ class _CancellingLoginBrowser: def test_login_closes_page_on_cancelled_error( monkeypatch: pytest.MonkeyPatch, ) -> None: - """_login_once закрывает страницу даже при CancelledError (finally, не except). - - CancelledError — BaseException, обходит except Exception, без finally страница - утекала бы при отмене таска (shutdown / dropped client). - """ + """_login_once закрывает страницу даже при CancelledError (finally, не except).""" tracker = _Tracker() - browser = _CancellingLoginBrowser(tracker) - monkeypatch.setattr(server, "_browser", browser) + server._browsers["cian"] = _CancellingLoginBrowser(tracker) params = { - "url": "https://cian.example/login", + "url": "https://www.cian.ru/login", "email": "u@example.com", "password": "secret", "email_selector": "#email", @@ -400,7 +281,7 @@ def test_login_closes_page_on_cancelled_error( } with pytest.raises(asyncio.CancelledError): - asyncio.run(server._login_once(params)) + asyncio.run(server._login_once("cian", params)) assert tracker.opened == 1 assert tracker.closed == 1, "страница должна закрыться в finally при CancelledError"