"""test_server_http_status.py — HTTP-статус целевой навигации в /fetch (#3196).
Проблема, установленная замером на проде 28.08.2026: ДомКлик отдаёт статическую
страницу «403 | Домклик» ровно на 26 624 байта, в которой НЕТ ни startpow, ни
qrator, ни капчи, ни PoW-скрипта. Ни один маркер из _CHALLENGE_MARKERS/_BAN_MARKERS
(оба сняты с авитовских страниц) в неё не попадает, поэтому отказ уезжал наверх как
валидный контент, парсер не находил состояние страницы, и прогон получал блок
неизвестной природы — 14 прогонов domclick_detail_backfill подряд с ban_kind=unknown.
_fetch_once игнорировал Response, который возвращает page.goto. Теперь статус
целевой навигации сохраняется в _last_response_status[provider] и отдаётся в теле
/fetch рядом с html — АДДИТИВНО, ключ "html" не меняется.
camoufox НЕ запускается: _browsers[provider] — поддельный browser/page (зеркалит
test_server_pow_challenge.py). wait_for_timeout на фейковой page — no-op.
Запуск (из tradein-mvp/browser/)::
python -m pytest test_server_http_status.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)
@pytest.fixture(autouse=True)
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
"""Чистое per-provider состояние на каждый тест (зеркалит соседние тесты)."""
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, "_last_goto_at", {})
monkeypatch.setattr(server, "_last_response_status", {})
monkeypatch.setattr(server, "_launched_proxy", {})
monkeypatch.setattr(server, "_locks_guard", asyncio.Lock())
monkeypatch.delenv("SCRAPER_PROXY_URL", raising=False)
# Заглушка ДомКлика: 403 и НИ ОДНОГО маркера из _CHALLENGE_MARKERS/_BAN_MARKERS —
# ровно то, на чём ломался старый детектор.
_DOMCLICK_403_HTML = "403 | Домклик"
_REAL_HTML = "listing card content"
# Челлендж Авито: маркер startPow при HTTP 200 — путь, который трогать нельзя.
_CHALLENGE_HTML = ""
class _Response:
"""Поддельный playwright Response — интересует только .status."""
def __init__(self, status: int) -> None:
self.status = status
class _Page:
"""Поддельная page: goto отдаёт заданный Response, content() — заданный HTML.
goto_results — по одному элементу на вызов goto (None = «playwright вернул None»,
навигационная гонка/редирект). После исчерпания повторяет последний.
"""
def __init__(
self,
html_sequence: list[str],
goto_results: list[_Response | None],
) -> None:
self._html_sequence = html_sequence
self._goto_results = goto_results
self._content_calls = 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) -> _Response | None:
idx = min(len(self.goto_urls), len(self._goto_results) - 1)
self.goto_urls.append(url)
return self._goto_results[idx]
async def wait_for_timeout(self, ms: int) -> None:
self.wait_for_timeout_calls.append(ms)
async def content(self) -> str:
idx = min(self._content_calls, len(self._html_sequence) - 1)
self._content_calls += 1
return self._html_sequence[idx]
async def close(self) -> None:
self.closed += 1
class _Browser:
def __init__(self, page: _Page) -> None:
self._page = page
async def new_page(self) -> _Page:
return self._page
def _install(monkeypatch: pytest.MonkeyPatch, page: _Page, provider: str = "avito") -> None:
server._browsers[provider] = _Browser(page)
monkeypatch.setattr(
server, "_RECYCLE_PAGES_BY_PROVIDER",
dict.fromkeys(server.PROVIDERS, 10_000),
)
monkeypatch.setattr(server, "BROWSER_WAIT_MS", 0)
monkeypatch.setattr(server, "_MIN_PAGE_INTERVAL_BY_PROVIDER", {})
monkeypatch.setattr(server, "BROWSER_MIN_PAGE_INTERVAL_S", 0.0)
def _json_body(response: Any) -> dict[str, Any]:
return json.loads(response.body.decode())
async def _coro(value: Any) -> Any:
return value
def _make_request(body: dict[str, Any]) -> Any:
request = make_mocked_request("POST", "/fetch")
request.json = lambda: _coro(body) # type: ignore[method-assign]
return request
# ── _status_of: чтение статуса ────────────────────────────────────────────────
def test_status_of_reads_response_status() -> None:
assert server._status_of(_Response(403)) == 403
def test_status_of_none_response_is_none() -> None:
"""goto вернул None (редирект/навигационная гонка) — не падаем, статуса нет."""
assert server._status_of(None) is None
# ── _fetch_once: статус пишется в per-provider состояние ──────────────────────
def test_fetch_once_records_target_status(monkeypatch: pytest.MonkeyPatch) -> None:
page = _Page([_DOMCLICK_403_HTML], [_Response(403)])
_install(monkeypatch, page)
html = asyncio.run(server._fetch_once("avito", "https://domclick.ru/card/1"))
assert html == _DOMCLICK_403_HTML # контент отдан как раньше — правка аддитивна
assert server._last_response_status["avito"] == 403
def test_fetch_once_status_none_when_goto_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
page = _Page([_REAL_HTML], [None])
_install(monkeypatch, page)
html = asyncio.run(server._fetch_once("avito", "https://www.avito.ru/x"))
assert html == _REAL_HTML
assert server._last_response_status["avito"] is None
def test_fetch_once_status_taken_from_target_not_origin(monkeypatch: pytest.MonkeyPatch) -> None:
"""Прогрев origin отдал 200, целевая навигация — 403. Наверх идёт 403."""
page = _Page([_DOMCLICK_403_HTML], [_Response(200), _Response(403)])
_install(monkeypatch, page)
asyncio.run(
server._fetch_once("avito", "https://domclick.ru/card/1", origin="https://domclick.ru/")
)
assert page.goto_urls == ["https://domclick.ru/", "https://domclick.ru/card/1"]
assert server._last_response_status["avito"] == 403
def test_fetch_once_resets_stale_status_before_navigation(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Статус прошлого запроса не должен пережить следующую навигацию."""
server._last_response_status["avito"] = 403
page = _Page([_REAL_HTML], [None])
_install(monkeypatch, page)
asyncio.run(server._fetch_once("avito", "https://www.avito.ru/x"))
assert server._last_response_status["avito"] is None
# ── статус в диагностике: 403/429 не ждут PoW ─────────────────────────────────
def test_fetch_once_skips_pow_wait_on_403(monkeypatch: pytest.MonkeyPatch) -> None:
"""403 + маркеры челленджа — отказ, а не проверка: бюджет ожидания не тратим.
Заглушку наверх НЕ отдаём: вернув её как валидный HTML, воскресили бы #3045 —
парсер Авито получил бы страницу челленджа вместо карточки.
"""
page = _Page([_CHALLENGE_HTML], [_Response(403)])
_install(monkeypatch, page)
monkeypatch.setattr(server, "BROWSER_CHALLENGE_WAIT_MS", 30_000)
with pytest.raises(server.BanPageDetectedError):
asyncio.run(server._fetch_once("avito", "https://www.avito.ru/x"))
assert page.wait_for_timeout_calls == [] # ни одного тика ожидания челленджа
def test_fetch_once_skips_pow_wait_on_429(monkeypatch: pytest.MonkeyPatch) -> None:
page = _Page([_CHALLENGE_HTML], [_Response(429)])
_install(monkeypatch, page)
with pytest.raises(server.BanPageDetectedError):
asyncio.run(server._fetch_once("avito", "https://www.avito.ru/x"))
assert page.wait_for_timeout_calls == []
def test_fetch_once_still_waits_out_challenge_on_200(monkeypatch: pytest.MonkeyPatch) -> None:
"""HTTP 200 с PoW-заглушкой (путь Авито) не задет: ждём и получаем контент."""
page = _Page([_CHALLENGE_HTML, _CHALLENGE_HTML, _REAL_HTML], [_Response(200)])
_install(monkeypatch, page)
html = asyncio.run(server._fetch_once("avito", "https://www.avito.ru/x"))
assert html == _REAL_HTML
assert page.wait_for_timeout_calls # ожидание челленджа реально шло
def test_fetch_once_clears_stale_challenge_status_once_settled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""#3283: снятый челлендж не должен отдать наверх статус challenge-страницы.
Замер прода (прогон 5606): сайдкар пишет `_last_response_status` ОДИН раз, из
ответа на challenge-страницу (тут — 439), и никогда его не обновляет после
self-reload'а, которым PoW чинит себя сам. Клиент (avito/detail.py) трактует
403/439 как блок площадки НЕ глядя в HTML и выбрасывает уже вытащенную карточку.
После этой правки статус обязан стать None («неизвестен») — клиент такой статус
блоком не считает.
"""
page = _Page([_CHALLENGE_HTML, _REAL_HTML], [_Response(439)])
_install(monkeypatch, page)
html = asyncio.run(server._fetch_once("avito", "https://www.avito.ru/x"))
assert html == _REAL_HTML
assert server._last_response_status["avito"] is None
def test_fetch_once_challenge_timeout_keeps_status_as_is(monkeypatch: pytest.MonkeyPatch) -> None:
"""Челлендж НЕ снялся → ChallengeTimeoutError, статус ответа НЕ обнуляем.
Отказ остаётся отказом (требование задачи #3283): return settled — единственное
место, где мы решаем, что status протух; на ветке таймаута до него не доходит.
"""
monkeypatch.setattr(server, "BROWSER_CHALLENGE_WAIT_MS", 1000)
page = _Page([_CHALLENGE_HTML], [_Response(439)]) # челлендж висит вечно
_install(monkeypatch, page)
with pytest.raises(server.ChallengeTimeoutError):
asyncio.run(server._fetch_once("avito", "https://www.avito.ru/x"))
assert server._last_response_status["avito"] == 439
def test_fetch_once_ban_marker_still_raises_on_200(monkeypatch: pytest.MonkeyPatch) -> None:
"""Ветка _is_ban_page осталась: «проблема с IP» с кодом 200 — по-прежнему бан."""
page = _Page(["Доступ ограничен: проблема с IP"], [_Response(200)])
_install(monkeypatch, page)
with pytest.raises(server.BanPageDetectedError):
asyncio.run(server._fetch_once("avito", "https://www.avito.ru/x"))
# ── /fetch: статус доезжает в тело ответа ─────────────────────────────────────
def test_fetch_handler_returns_status_next_to_html(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(server, "IS_PROD", False)
async def _ensure(provider: str, proxy_override: str | None = None) -> bool:
return True
async def _fake_do_fetch(provider: str, url: str, **_kw: Any) -> str:
server._last_response_status[provider] = 403
return _DOMCLICK_403_HTML
monkeypatch.setattr(server, "_ensure_browser", _ensure)
monkeypatch.setattr(server, "_do_fetch", _fake_do_fetch)
response = asyncio.run(server.fetch_handler(_make_request({"url": "https://domclick.ru/x"})))
assert response.status == 200
body = _json_body(response)
assert body["html"] == _DOMCLICK_403_HTML # старый ключ не тронут
assert body["status"] == 403
def test_fetch_handler_status_null_without_response(monkeypatch: pytest.MonkeyPatch) -> None:
"""Статуса нет (goto вернул None) — ответ не ломается, status=null."""
monkeypatch.setattr(server, "IS_PROD", False)
async def _ensure(provider: str, proxy_override: str | None = None) -> bool:
return True
async def _fake_do_fetch(provider: str, url: str, **_kw: Any) -> str:
server._last_response_status[provider] = None
return _REAL_HTML
monkeypatch.setattr(server, "_ensure_browser", _ensure)
monkeypatch.setattr(server, "_do_fetch", _fake_do_fetch)
response = asyncio.run(server.fetch_handler(_make_request({"url": "https://www.avito.ru/x"})))
body = _json_body(response)
assert body["html"] == _REAL_HTML
assert body["status"] is None
# ── #3239: 500-ка бан-страницы отличима от 500-ки транспорта ──────────────────
def _fetch_handler_error_body(
monkeypatch: pytest.MonkeyPatch, exc: Exception, upstream_status: int | None
) -> tuple[int, dict]:
monkeypatch.setattr(server, "IS_PROD", False)
async def _ensure(provider: str, proxy_override: str | None = None) -> bool:
return True
async def _fake_do_fetch(provider: str, url: str, **_kw: Any) -> str:
server._last_response_status[provider] = upstream_status
raise exc
monkeypatch.setattr(server, "_ensure_browser", _ensure)
monkeypatch.setattr(server, "_do_fetch", _fake_do_fetch)
response = asyncio.run(server.fetch_handler(_make_request({"url": "https://domclick.ru/x"})))
return response.status, _json_body(response)
def test_fetch_handler_marks_ban_page_in_error_body(monkeypatch: pytest.MonkeyPatch) -> None:
"""Бан-страница несёт ban_page + апстрим-статус И ОТЛИЧИМЫЙ КОД 403.
Без признака в теле клиент видит обычную 500-ку, уводит отказ в транспортную
ветку (report_ban там не зовётся) и перестаёт ротировать отказавший узел —
регрессия, которую #3237 внёс, а #3239 чинит.
Код 403 (#3288 п.4): один и тот же 500 означал «площадка забанила» и
«сайдкар упал», и разбор инцидента начинался с ложного следа. Тело не
меняется — клиент опознаёт бан по нему (образы деплоятся врозь).
"""
status, body = _fetch_handler_error_body(
monkeypatch, server.BanPageDetectedError("статический отказ площадки"), 401
)
assert status == 403
assert body["ban_page"] is True
assert body["status"] == 401
assert "BanPageDetectedError" in body["error"]
def test_fetch_handler_transport_error_has_no_ban_page_flag(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Обычный сбой — признака нет вовсе, клиент трактует его как раньше."""
status, body = _fetch_handler_error_body(
monkeypatch, RuntimeError("NS_ERROR_PROXY_BAD_GATEWAY"), None
)
assert status == 500
assert "ban_page" not in body
def test_fetch_handler_ban_page_without_status_stays_null(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Апстрим-статуса нет — отдаём null, а не выдуманный код (#2764)."""
_status, body = _fetch_handler_error_body(
monkeypatch, server.BanPageDetectedError("бан-страница"), None
)
assert body["ban_page"] is True
assert body["status"] is None