"""test_server_domclick_challenge.py — QRATOR-рукопожатие DomClick в /fetch (#3196).
Проблема, установленная замером прода 29.08.2026: у DomClick, в отличие от
Авито, промежуточная (нерендеренная) PoW-страница НЕ содержит ни одного
стабильного маркера вообще — ни `__qrator`, ни `startpow`, ни текста отказа.
`_CHALLENGE_MARKERS` (сняты живьём с Авито, #3045) на DomClick никогда не
матчатся, поэтому ветка `_wait_out_pow_challenge` для него не включалась —
недосчитанная страница уезжала наверх как валидный контент, парсер не находил
`__SSR_STATE__` и поднимал ложный "блок" (двухнедельный
`attempted=3, blocked=3, enriched=0` у `domclick_detail_backfill`).
Логика для DomClick инвертирована: опознаём положительно ДВА крайних
состояния — «точно успех» (`_DOMCLICK_SUCCESS_MARKER`, JS-глобаль
`__SSR_STATE__`) и «точно отказ» (`_DOMCLICK_REFUSAL_MARKERS`, статическая
страница «403 | Домклик» ~26.6 КБ) — а всё остальное (загрузчик рукопожатия
279 байт, нерендеренная PoW-страница ~7 КБ) трактуется как «рукопожатие ещё
идёт» и уходит в существующий `_wait_out_pow_challenge` с кастомным
`is_pending`.
Тестируется _fetch_once целиком (зеркалит test_server_pow_challenge.py),
camoufox не запускается — фейковые browser/page возвращают заранее заданную
последовательность content(). wait_for_timeout на фейковой page — no-op.
Запуск (из tradein-mvp/browser/)::
python -m pytest test_server_domclick_challenge.py -q
"""
from __future__ import annotations
import asyncio
import importlib.util
from pathlib import Path
from typing import Any
import pytest
# 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)
@pytest.fixture(autouse=True)
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
"""Чистое per-provider состояние на каждый тест (зеркалит test_server_smoke.py)."""
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())
# Приблизительные образцы трёх состояний, снятых живьём 29.08.2026 (см. докстринг
# задачи) — точный размер в байтах для теста не важен, важно отсутствие/наличие
# опознаваемых строк.
_HANDSHAKE_LOADER_HTML = (
''
""
)
_RENDERED_POW_HTML = (
""
"ожидание...
"
)
_REFUSAL_HTML = (
"403 | Домклик"
"Похоже, ваш запрос выглядит необычно. Мы усилили защиту."
)
_SUCCESS_HTML = ''
class _DomclickPage:
"""Поддельная page: отдаёт заданную последовательность content() по вызовам.
После исчерпания списка повторяет последний элемент (имитирует «страница
осталась в этом состоянии»). Зеркалит _ChallengePage из
test_server_pow_challenge.py.
"""
def __init__(self, html_sequence: list[str]) -> None:
self._html_sequence = html_sequence
self._call_count = 0
self.goto_urls: list[str] = []
self.wait_for_timeout_calls: list[int] = []
self.closed = 0
async def route(self, pattern: str, handler: Any) -> None:
return None
async def goto(self, url: str, **kwargs: Any) -> None:
self.goto_urls.append(url)
async def wait_for_timeout(self, ms: int) -> None:
self.wait_for_timeout_calls.append(ms)
async def content(self) -> str:
idx = min(self._call_count, len(self._html_sequence) - 1)
item = self._html_sequence[idx]
self._call_count += 1
return item
async def close(self) -> None:
self.closed += 1
class _DomclickBrowser:
def __init__(self, page: _DomclickPage) -> None:
self._page = page
async def new_page(self) -> _DomclickPage:
return self._page
def _install(
monkeypatch: pytest.MonkeyPatch, page: _DomclickPage, provider: str = "domclick"
) -> None:
server._browsers[provider] = _DomclickBrowser(page)
monkeypatch.setattr(
server, "_RECYCLE_PAGES_BY_PROVIDER",
dict.fromkeys(server.PROVIDERS, 10_000),
)
# ── детекторы: чистые функции ────────────────────────────────────────────────────
def test_is_domclick_success_true_on_ssr_state_marker() -> None:
assert server._is_domclick_success(_SUCCESS_HTML) is True
def test_is_domclick_success_false_on_handshake_loader() -> None:
assert server._is_domclick_success(_HANDSHAKE_LOADER_HTML) is False
def test_is_domclick_success_false_on_rendered_pow_page() -> None:
"""Ключевой случай задачи: у рендеренной PoW-страницы НЕТ стабильных маркеров
вообще — детектор успеха не должен ложно сработать на ней."""
assert server._is_domclick_success(_RENDERED_POW_HTML) is False
def test_is_domclick_success_false_on_refusal_page() -> None:
assert server._is_domclick_success(_REFUSAL_HTML) is False
def test_is_domclick_refusal_true_on_title_and_text_markers() -> None:
assert server._is_domclick_refusal(_REFUSAL_HTML) is True
def test_is_domclick_refusal_false_on_handshake_loader() -> None:
assert server._is_domclick_refusal(_HANDSHAKE_LOADER_HTML) is False
def test_is_domclick_refusal_false_on_rendered_pow_page() -> None:
assert server._is_domclick_refusal(_RENDERED_POW_HTML) is False
def test_is_domclick_refusal_false_on_success_page() -> None:
assert server._is_domclick_refusal(_SUCCESS_HTML) is False
# ── _fetch_once: сценарии DomClick ───────────────────────────────────────────────
def test_fetch_once_refusal_page_raises_immediately_without_waiting(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Страница отказа → BanPageDetectedError СРАЗУ, без ухода в опрос (пункт 1)."""
page = _DomclickPage([_REFUSAL_HTML])
_install(monkeypatch, page)
with pytest.raises(server.BanPageDetectedError):
asyncio.run(server._fetch_once("domclick", "https://ekaterinburg.domclick.ru/card/1"))
assert page.closed == 1
# Единственная пауза — обычная гидро-пауза ДО детекта; цикла опроса (доп.
# паузы по 1000мс) не было — бюджет ожидания не потрачен.
assert page.wait_for_timeout_calls == [server.BROWSER_WAIT_MS]
def test_fetch_once_page_without_markers_waits_then_returns_success(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Ни успеха, ни отказа (загрузчик/рендеренная PoW-страница) → уходит в
ожидание, а не отдаёт недосчитанную страницу как контент (пункт 2)."""
page = _DomclickPage([_HANDSHAKE_LOADER_HTML, _RENDERED_POW_HTML, _SUCCESS_HTML])
_install(monkeypatch, page)
html = asyncio.run(
server._fetch_once("domclick", "https://ekaterinburg.domclick.ru/card/1")
)
assert html == _SUCCESS_HTML
assert page.closed == 1
assert server.BROWSER_WAIT_MS in page.wait_for_timeout_calls
assert 1000 in page.wait_for_timeout_calls # шаг опроса — бюджет реально тратился
def test_fetch_once_success_page_returned_immediately_without_waiting(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Успех с первого раза → отдаётся сразу, БЕЗ ухода в polling-ветку (пункт 3)."""
page = _DomclickPage([_SUCCESS_HTML])
_install(monkeypatch, page)
html = asyncio.run(
server._fetch_once("domclick", "https://ekaterinburg.domclick.ru/card/1")
)
assert html == _SUCCESS_HTML
assert page.closed == 1
assert page.wait_for_timeout_calls == [server.BROWSER_WAIT_MS]
def test_fetch_once_settles_into_refusal_during_wait_raises_ban_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Рукопожатие домоталось до отказа уже ПОСЛЕ начала опроса → тоже BanPageDetectedError,
а не отдача отказной страницы как валидного контента."""
page = _DomclickPage([_HANDSHAKE_LOADER_HTML, _REFUSAL_HTML])
_install(monkeypatch, page)
with pytest.raises(server.BanPageDetectedError):
asyncio.run(server._fetch_once("domclick", "https://ekaterinburg.domclick.ru/card/1"))
assert page.closed == 1
def test_fetch_once_raises_challenge_timeout_when_budget_exhausted(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Ни успех, ни отказ так и не наступают → ChallengeTimeoutError по истечении бюджета."""
monkeypatch.setattr(server, "BROWSER_CHALLENGE_WAIT_MS", 2000)
page = _DomclickPage([_HANDSHAKE_LOADER_HTML]) # повторяется бесконечно
_install(monkeypatch, page)
with pytest.raises(server.ChallengeTimeoutError):
asyncio.run(server._fetch_once("domclick", "https://ekaterinburg.domclick.ru/card/1"))
assert page.closed == 1
def test_fetch_once_401_status_is_not_used_as_refusal_signal_for_domclick(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""HTTP 401 — статус здорового рукопожатия DomClick (замер 29.08.2026), НЕ должен
трактоваться как отказ: только тело страницы решает."""
class _Response:
status = 401
class _StatusPage(_DomclickPage):
async def goto(self, url: str, **kwargs: Any) -> _Response | None:
self.goto_urls.append(url)
return _Response()
page = _StatusPage([_HANDSHAKE_LOADER_HTML, _SUCCESS_HTML])
_install(monkeypatch, page)
html = asyncio.run(
server._fetch_once("domclick", "https://ekaterinburg.domclick.ru/card/1")
)
assert html == _SUCCESS_HTML
# ── не-домклик провайдеры: поведение не изменилось (пункт 4) ────────────────────
def test_fetch_once_non_domclick_provider_ignores_domclick_markers(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""avito с HTML, который матчил бы _is_domclick_success/_is_domclick_refusal,
не должен уходить ни в какую специальную ветку — DomClick-логика провайдер-
скопирована строго под `provider == "domclick"`."""
page = _DomclickPage([_REFUSAL_HTML]) # содержит "похоже, ваш запрос..." — не avito-маркер
_install(monkeypatch, page, provider="avito")
html = asyncio.run(server._fetch_once("avito", "https://www.avito.ru/card/1"))
assert html == _REFUSAL_HTML # отдан как обычный контент — avito его не распознаёт
assert page.wait_for_timeout_calls == [server.BROWSER_WAIT_MS]