gendesign/tradein-mvp/backend/tests/test_2687_pool_exhaustion_keeps_checkpoint.py

"""#2687: опустевший пул прокси mid-run стирал чекпоинт полного обхода.

Найдено при разборе шести `ban_kind='infra'` у `avito_full_load` (29.07-03.08). Сами
шесть объяснены и починены раньше (#2637 подключил браузерный путь Авито к пулу,
#2634/#2640 добили запасной путь), но рядом с ними на main живёт соседний дефект того
же класса, что и #2686: наш собственный отказ финализируется веткой, которая теряет
чекпоинт.

Механизм. `NoProxyAvailableError` (proxy_errors.py) — наследник `RuntimeError`, и его
собственный докстринг говорит прямо: «это НАША инфраструктура (нет живого прокси), не
внешний блок». В `run_*_full_load` он попадал в общую ветку `except RuntimeError` →
`mark_failed`, а `mark_failed`, в отличие от `mark_banned`, НЕ пишет `done_buckets`
(pipeline.py, ветка блока пишет его явно). Плюс `ban_kind_of_exception` возвращал для
него `unknown` — диагноз «наша инфраструктура» терялся дважды: и как метка, и как
сохранённый прогресс.

Достижимость на проде (замер 2026-08-12, read-only):
  - под `avito` доступны три узла (id 9/10/11; id 1 забанен парой до 13.08);
  - `BrowserFetcher` меняет lease после `_LEASE_ROTATE_AFTER_FAILS=3` подряд неудачных
    `/fetch`, а `proxy_pool.acquire` не выдаёт узел с
    `consecutive_fails >= MAX_CONSECUTIVE_FAILS=3` → девять неудачных POST'ов
    опустошают пул под источник;
  - лестница ОДНОГО полного обхода при отказывающем сайдкаре делает до 24 POST'ов
    (4 бакета `_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED` × 3 попытки
    `_AVITO_SIDECAR_TRANSIENT_RETRIES`+1 × 2 внутренних httpx-ретрая `fetch`),
    то есть заведомо проходит через это состояние.
Цена потери чекпоинта — прогон 3547 (09.08): 35 бакетов, 5496 объявлений за 2ч58м.

Фальсификация (красный прогон на коде до правки):
  - `ban_kind_of_exception` → `'unknown'` вместо `'infra'`;
  - `run_*_full_load` → `mark_failed` вместо `mark_banned`, исключение улетает наружу
    (тест падает на непойманном `NoProxyAvailableError`), `done_buckets` нигде нет.
"""

from __future__ import annotations

import os
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")

from scraper_kit.orchestration import runs as kit_runs
from scraper_kit.orchestration.pipeline import (
    ban_kind_of_exception,
    run_avito_full_load,
    run_cian_full_load,
    run_yandex_full_load,
)
from scraper_kit.proxy_errors import NoProxyAvailableError

PFX = "scraper_kit.orchestration.pipeline"

_FULL_LOAD = {
    "avito": (run_avito_full_load, f"{PFX}.AvitoScraper"),
    "cian": (run_cian_full_load, f"{PFX}.CianScraper"),
    "yandex": (run_yandex_full_load, f"{PFX}.YandexRealtyScraper"),
}


class _Recorder:
    """Двойник scrape_runs: интересны имя финализатора, ban_kind и counters."""

    def __init__(self) -> None:
        self.calls: list[tuple[str, str, dict[str, Any]]] = []

    def is_cancelled(self, db: Any, run_id: int) -> bool:
        return False

    def update_heartbeat(self, db: Any, run_id: int, counters: dict[str, Any]) -> None:
        pass

    def mark_done(self, db: Any, run_id: int, counters: dict[str, Any]) -> None:
        self.calls.append(("mark_done", "", dict(counters)))

    def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None:
        self.calls.append(("mark_failed", "", dict(counters)))

    def mark_banned(
        self,
        db: Any,
        run_id: int,
        error: str,
        counters: dict[str, Any],
        *,
        ban_kind: str = kit_runs.BAN_KIND_UNKNOWN,
    ) -> None:
        self.calls.append(("mark_banned", ban_kind, dict(counters)))


def _config() -> SimpleNamespace:
    return SimpleNamespace(
        scraper_fetch_mode="curl_cffi",
        browser_http_endpoint="http://browser.test/fetch",
        scraper_proxy_url=None,
        avito_proxy_max_rotations=0,
        avito_serp_ok_not_banned=True,
        avito_proxy_rotate_settle_s=0.0,
        cian_proxy_max_rotations=0,
        yandex_proxy_max_rotations=0,
        cian_full_load_per_fetch_timeout_s=0.0,
        scraper_skip_seen_today=False,
    )


def _scraper_that_saves_one_bucket_then_runs_out_of_proxies() -> MagicMock:
    """Двойник скрапера: один бакет доехал до on_bucket, затем пул опустел."""

    async def _fetch(*_a: Any, on_bucket: Any = None, **_k: Any) -> None:
        on_bucket("2к:0-5m", [MagicMock(source_id="a1")])
        raise NoProxyAvailableError("avito")

    m = MagicMock()
    m.__aenter__ = AsyncMock(return_value=m)
    m.__aexit__ = AsyncMock(return_value=None)
    m.fetch_all_secondary = _fetch
    m.state_extraction_attempts = 0
    m.state_extraction_failures = 0
    m.gate_fetch_attempts = 0
    m.gate_fetch_failures = 0
    m._browser = None
    return m


def test_ban_kind_of_no_proxy_is_infra() -> None:
    """Пустой пул — наша инфраструктура по определению самого исключения.

    Раньше классификатор отдавал 'unknown' (дефолт #2764): для НЕизвестной причины
    это верно, но здесь причина известна и названа в типе.
    """
    assert ban_kind_of_exception(NoProxyAvailableError("avito")) == kit_runs.BAN_KIND_INFRA


@pytest.mark.parametrize("source", ["avito", "cian", "yandex"])
async def test_full_load_pool_exhaustion_keeps_checkpoint(source: str) -> None:
    """Пул опустел mid-run → 'banned'/'infra' и done_buckets целы, а не mark_failed.

    Требование #2686 («чекпоинт сохраняется и при нашем сбое, и при блокировке
    площадкой») до этой правки держалось только для отказа сайдкара; отказ пула шёл
    мимо него и обнулял прогресс следующего прогона.
    """
    fn, scraper_target = _FULL_LOAD[source]
    recorder = _Recorder()
    extra: dict[str, Any] = {}
    if source == "yandex":
        enrichment = MagicMock()
        enrichment.record_yandex_price_history = MagicMock(return_value=0)
        extra["enrichment"] = enrichment

    scraper = _scraper_that_saves_one_bucket_then_runs_out_of_proxies()
    with (
        patch(scraper_target, return_value=scraper),
        patch(f"{PFX}.save_listings", MagicMock(return_value=(1, 0))),
        patch(f"{PFX}.runs", recorder),
    ):
        await fn(MagicMock(), run_id=1, config=_config(), matcher=MagicMock(), **extra)

    assert [c[0] for c in recorder.calls] == ["mark_banned"]
    _method, ban_kind, counters = recorder.calls[0]
    assert ban_kind == kit_runs.BAN_KIND_INFRA
    assert counters["done_buckets"] == ["2к:0-5m"]