"""Вид блока решает обрыв по доле и статус прогона (#3288, часть B).

Прогон 5425 оборвался по доле блоков (`abort_reason='ratio'`, статус `banned`) на
48 «блоках», из которых 41 был `AvitoSidecarUnavailableError` — отказ НАШЕГО тракта,
а не площадки: та же карточка не доехала до Авито, и 41 других карточек прогон
успел честно обогатить. `record_block()` вида не принимал вовсе, поэтому в числитель
скользящего окна (#3184) падало всё подряд.

Проверка ПО ЗНАЧЕНИЮ — сравниваются `abort_reason()`, счётчики и то, какой именно
финализатор дёрнут (`mark_banned` / `mark_failed` / `mark_done`), а не «не бросило
исключение». К каждому послаблению идёт контроль на противоположную ошибку: 14
настоящих `AvitoBlockedError` из 20 обязаны рвать прогон как раньше (#3184 не
меняется), а пустой пул опознаётся по ЦЕПОЧКЕ `__cause__`, не по подстроке
«no proxy available» — на подстроке уже обжёгся #3272.

Харнесс бэкфилла зеркалит tests/test_3338_backfill_counter_identity.py, оттуда же
тождество attempted = enriched + blocked + gone + failed.
"""

from __future__ import annotations

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

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

_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)

import pytest  # noqa: E402
from scraper_kit.avito_exceptions import (  # noqa: E402
    AvitoBlockedError,
    AvitoSidecarUnavailableError,
)
from scraper_kit.proxy_errors import NoProxyAvailableError  # noqa: E402

from app.services import scrape_runs as runs_svc  # noqa: E402
from app.services.backfill_block_breaker import BlockRatioBreaker  # noqa: E402
from app.services.scrape_runs import (  # noqa: E402
    BAN_KIND_INFRA,
    BAN_KIND_PLATFORM,
    BAN_KIND_UNKNOWN,
)
from app.tasks.avito_detail_backfill import (  # noqa: E402
    AvitoDetailBackfillResult,
    run_avito_detail_backfill,
)

_A_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
_A_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
_A_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
_A_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
_A_SETTINGS = "app.tasks.avito_detail_backfill.settings"
_A_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
_A_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
_A_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"
_A_ROTATE_PROXY = "app.tasks.avito_detail_backfill.rotate_proxy"

_WINDOW = 20
_THRESHOLD = 0.7


def _breaker(snapshot_size: int = 100) -> BlockRatioBreaker:
    return BlockRatioBreaker(
        window_size=_WINDOW,
        ratio_threshold=_THRESHOLD,
        safety_min=5,
        snapshot_size=snapshot_size,
    )


# ── брейкер: что попадает в числитель доли ────────────────────────────────────


def test_twenty_sidecar_blocks_do_not_abort_by_ratio() -> None:
    """20 подряд отказов сайдкара при снапшоте больше окна — не 'ratio' (приёмка)."""
    breaker = _breaker()
    for _ in range(20):
        breaker.record_block(BAN_KIND_INFRA)

    assert breaker.abort_reason() is None, (
        f"abort_reason={breaker.abort_reason()!r} при 20 infra-отказах: площадка "
        "не отказывала ни разу, рвать прогон не по чему"
    )
    assert (breaker.window_blocks, breaker.window_len) == (0, 20), (
        f"окно {breaker.window_blocks}/{breaker.window_len}: infra идёт в ЗНАМЕНАТЕЛЬ "
        "(как record_failure), а не в числитель"
    )
    assert breaker.consecutive_blocks == 0, (
        f"consecutive_blocks={breaker.consecutive_blocks}: infra не двигает серию, "
        "иначе на коротком снапшоте прогон рвал бы safety-net по тем же отказам"
    )


def test_fourteen_platform_blocks_of_twenty_still_abort_by_ratio() -> None:
    """Контроль на противоположную ошибку: #3184 не меняется (14/20 = порог 0.7)."""
    breaker = _breaker()
    for _ in range(6):
        breaker.record_success()
    for _ in range(14):
        breaker.record_block(BAN_KIND_PLATFORM)

    assert breaker.abort_reason() == "ratio", (
        f"abort_reason={breaker.abort_reason()!r} при 14 настоящих блоках из 20 — "
        "критерий #3184 обязан сработать ровно на пороге"
    )
    assert breaker.window_blocks == 14, f"числитель {breaker.window_blocks}, ожидали 14"


def test_infra_does_not_pad_the_numerator_to_the_threshold() -> None:
    """13 platform + 7 infra = 0.65, а не 1.0: infra числитель не добивает."""
    breaker = _breaker()
    for _ in range(7):
        breaker.record_block(BAN_KIND_INFRA)
    for _ in range(13):
        breaker.record_block(BAN_KIND_PLATFORM)

    assert breaker.abort_reason() is None, (
        f"abort_reason={breaker.abort_reason()!r}: 13/20 = 0.65 ниже порога 0.7, "
        "прогон рвётся только если infra зачли блоками"
    )


# ── бэкфилл: тот же вопрос сквозь задачу ──────────────────────────────────────


def _fake_avito_settings() -> MagicMock:
    return MagicMock(
        scraper_fetch_mode="browser",
        avito_detail_backfill_use_curl=False,
        detail_backfill_block_ratio_window=_WINDOW,
        detail_backfill_block_ratio_threshold=_THRESHOLD,
        browser_http_endpoint="http://browser:9000",
        avito_detail_backfill_rotate_after_attempts=15,
        avito_detail_backfill_rotate_on_ban_max=0,
        avito_detail_backfill_rotate_on_ban_min_gap=10,
    )


def _mock_avito_db(n: int) -> MagicMock:
    snapshot = [
        {
            "id": i + 1,
            "source_url": f"https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_{i + 1}",
        }
        for i in range(n)
    ]
    db = MagicMock()
    sel = MagicMock()
    sel.mappings.return_value.all.return_value = snapshot
    db.execute.return_value = sel
    return db


def _mock_browser_fetcher_cls() -> MagicMock:
    instance = AsyncMock()
    instance.__aenter__ = AsyncMock(return_value=instance)
    instance.__aexit__ = AsyncMock(return_value=False)
    instance.request_context_reset = MagicMock()
    instance.lease_id = 42
    return MagicMock(return_value=instance)


async def _run_avito(
    fetch_results: list[Any], *, snapshot_size: int
) -> tuple[AvitoDetailBackfillResult, MagicMock]:
    runs = MagicMock()
    db = _mock_avito_db(snapshot_size)
    with (
        patch(_A_SETTINGS, _fake_avito_settings()),
        patch(_A_SESSION),
        patch(_A_SCRAPER),
        patch(_A_RUNS, runs),
        patch(_A_BROWSER_FETCHER, _mock_browser_fetcher_cls()),
        patch(_A_FETCH, AsyncMock(side_effect=fetch_results)),
        patch(_A_SAVE, return_value=True),
        patch(_A_ROTATE_PROXY, AsyncMock()),
        patch(_A_SLEEP, new_callable=AsyncMock),
    ):
        counters = await run_avito_detail_backfill(
            db,
            run_id=3288,
            params={
                "batch_size": snapshot_size,
                "budget_sec": 3600,
                "max_consecutive_blocks": 10,
            },
        )
    return counters, runs


def _assert_identity(c: AvitoDetailBackfillResult, *, expected_attempted: int) -> None:
    outcomes = c.enriched + c.blocked + c.gone + c.failed
    assert (c.attempted, outcomes) == (expected_attempted, expected_attempted), (
        f"attempted={c.attempted} vs исходы={outcomes} "
        f"(enriched={c.enriched} blocked={c.blocked} gone={c.gone} failed={c.failed}), "
        f"ожидали {expected_attempted} — попытка без исхода (#3338)"
    )


@pytest.mark.asyncio
async def test_sidecar_run_walks_the_whole_snapshot_and_is_not_banned() -> None:
    """20 отказов сайдкара + 10 успехов: снапшот пройден целиком, обрыва нет."""
    fetches: list[Any] = [AvitoSidecarUnavailableError("sidecar 503") for _ in range(20)]
    fetches += [MagicMock() for _ in range(10)]

    counters, runs = await _run_avito(fetches, snapshot_size=30)

    finished = runs.mark_backfill_finished.call_args
    assert finished is not None, "прогон не финализирован mark_backfill_finished"
    # Приёмка #3288 первой строкой: величина, по которой прогон 5425 и оборвался.
    assert finished.args[2].get("abort_reason") is None, (
        f"abort_reason={finished.args[2].get('abort_reason')!r} при 20 отказах сайдкара: "
        "обрыва по доле быть не могло — площадка не отказала ни разу"
    )
    _assert_identity(counters, expected_attempted=30)
    assert (counters.blocked, counters.enriched) == (20, 10), (
        f"blocked={counters.blocked} enriched={counters.enriched}: прогон обязан дойти "
        "до конца снапшота, отказ нашего тракта его не рвёт"
    )
    assert finished.kwargs["aborted_by_blocks"] is False, (
        "aborted_by_blocks=True при нулевых отказах площадки — прогон уйдёт в 'banned'"
    )
    assert finished.kwargs["ban_kinds"] == {BAN_KIND_INFRA: 20}, (
        f"ban_kinds={finished.kwargs['ban_kinds']}: перепись диагнозов не должна "
        "теряться — она и решает статус"
    )


@pytest.mark.asyncio
async def test_done_run_keeps_ban_kinds_in_counters() -> None:
    """#3288 п.5: диагноз доезжает в counters и у прогона, который не 'banned'.

    Прогоны 7193/7264/7302 (1600 попыток, ~1599 отказов сайдкара, status='done') легли
    в БД без ban_kinds: перепись писалась в counters только на ветке бана.
    """
    fetches: list[Any] = [AvitoSidecarUnavailableError("sidecar 503") for _ in range(20)]
    fetches += [MagicMock() for _ in range(10)]

    _, runs = await _run_avito(fetches, snapshot_size=30)

    written = runs.mark_backfill_finished.call_args.args[2]
    assert written.get("ban_kinds") == {BAN_KIND_INFRA: 20}, (
        f"counters={written}: прогон без обрыва уходит в mark_done с этими counters — "
        "без переписи диагноз отказов теряется"
    )


@pytest.mark.asyncio
async def test_real_platform_bans_still_abort_by_ratio() -> None:
    """Контроль: 14 настоящих AvitoBlockedError из 20 рвут прогон как раньше."""
    fetches: list[Any] = [MagicMock() for _ in range(6)]
    fetches += [AvitoBlockedError("firewall/soft-block") for _ in range(14)]

    counters, runs = await _run_avito(fetches, snapshot_size=30)

    _assert_identity(counters, expected_attempted=20)
    finished = runs.mark_backfill_finished.call_args
    assert finished.args[2].get("abort_reason") == "ratio", (
        f"counters={finished.args[2]}: 14 отказов площадки из 20 — ровно порог #3184"
    )
    assert finished.kwargs["aborted_by_blocks"] is True, "обрыв по доле обязан быть помечен"


@pytest.mark.asyncio
async def test_empty_pool_stops_run_with_its_own_reason() -> None:
    """Пустой пул — не блок: прогон завершается 'нечем ходить', а не баном."""
    wrapped = AvitoSidecarUnavailableError("Avito detail browser fetch failed for https://...")
    wrapped.__cause__ = NoProxyAvailableError("avito")

    counters, runs = await _run_avito([wrapped], snapshot_size=30)

    _assert_identity(counters, expected_attempted=1)
    assert (counters.blocked, counters.failed) == (0, 1), (
        f"blocked={counters.blocked} failed={counters.failed}: к площадке не ходили — "
        "это отказ нашей стороны, а не блок"
    )
    assert runs.mark_backfill_finished.call_args is None, (
        "прогон с пустым пулом не должен идти через backfill-honest-status: у него своя причина"
    )
    failed = runs.mark_failed.call_args
    assert failed is not None, "прогон не финализирован mark_failed"
    assert failed.args[3]["no_proxy_stop"] == 1, f"counters={failed.args[3]}: нет no_proxy_stop"
    assert "пул прокси пуст" in failed.args[2], f"причина={failed.args[2]!r}"


@pytest.mark.asyncio
async def test_empty_pool_recognised_by_cause_not_by_text() -> None:
    """#3272: подстрока «no proxy available» в тексте — не доказательство пустого пула."""
    liar = AvitoBlockedError("firewall page: no proxy available for provider='avito'")

    counters, runs = await _run_avito([liar, MagicMock()], snapshot_size=2)

    _assert_identity(counters, expected_attempted=2)
    assert (counters.blocked, counters.enriched) == (1, 1), (
        f"blocked={counters.blocked} enriched={counters.enriched}: отказ площадки с "
        "«no proxy available» в тексте остаётся блоком — цепочка причин пуста"
    )
    assert runs.mark_failed.call_args is None, (
        "прогон остановлен как «пул пуст» по подстроке в тексте — ровно ошибка #3272"
    )


# ── статус прогона: диагноз infra ≠ 'banned' ──────────────────────────────────


def _finish(ban_kinds: dict[str, int], *, enriched: int) -> tuple[MagicMock, ...]:
    counters = {"attempted": 90, "enriched": enriched, "blocked": 48, "gone": 0, "failed": 1}
    with (
        patch.object(runs_svc, "mark_banned") as banned,
        patch.object(runs_svc, "mark_failed") as failed,
        patch.object(runs_svc, "mark_done") as done,
    ):
        runs_svc.mark_backfill_finished(
            MagicMock(),
            5425,
            counters,
            source="avito_detail_backfill",
            aborted_by_blocks=True,
            ban_kinds=ban_kinds,
        )
    return banned, failed, done


def test_infra_dominant_run_is_not_marked_banned() -> None:
    """41 infra из 48 при 41 обогащённой карточке — прогон 'done', не 'banned'."""
    banned, failed, done = _finish({BAN_KIND_INFRA: 41, BAN_KIND_PLATFORM: 7}, enriched=41)

    assert banned.call_count == 0, (
        "прогон помечен 'banned' при доминирующем infra — запись утверждает про "
        "площадку то, чего не было"
    )
    assert (failed.call_count, done.call_count) == (0, 1), (
        f"mark_failed={failed.call_count} mark_done={done.call_count}: 41 карточка "
        "обогащена, прогон работу сделал"
    )


def test_pure_infra_run_with_result_is_not_banned() -> None:
    """Основной кейс без примеси: перепись целиком infra, карточки есть → не 'banned'."""
    banned, failed, done = _finish({BAN_KIND_INFRA: 20}, enriched=10)

    assert (banned.call_count, failed.call_count, done.call_count) == (0, 0, 1), (
        f"mark_banned={banned.call_count} mark_failed={failed.call_count} "
        f"mark_done={done.call_count}: 20 отказов сайдкара при 10 обогащённых — 'done'"
    )


def test_infra_dominant_run_without_result_stays_banned() -> None:
    """Контроль на ОБРАТНУЮ ошибку: ноль результата — по-прежнему 'banned' (#2764/#3196).

    Диагноз infra на нулевом прогоне неотличим от бана площадки, опознанного по 5xx
    (yandex, #3196): статус остаётся 'banned', честность несёт ban_kind. Понижение
    до 'failed' прятало бы настоящий бан под «нашу поломку» — падало 2 теста CI.
    """
    banned, failed, done = _finish({BAN_KIND_INFRA: 41, BAN_KIND_PLATFORM: 7}, enriched=0)

    assert (banned.call_count, failed.call_count, done.call_count) == (1, 0, 0), (
        f"mark_banned={banned.call_count} mark_failed={failed.call_count} "
        f"mark_done={done.call_count}: нулевой прогон обязан остаться 'banned'"
    )
    assert banned.call_args.kwargs["ban_kind"] == BAN_KIND_INFRA, (
        f"ban_kind={banned.call_args.kwargs['ban_kind']!r}: диагноз обязан доехать"
    )


def test_run_without_census_stays_banned() -> None:
    """Источник видов не различает (ban_kinds пуст) → прежнее поведение: 'banned'."""
    banned, failed, done = _finish({}, enriched=41)

    assert (banned.call_count, failed.call_count, done.call_count) == (1, 0, 0), (
        f"mark_banned={banned.call_count} mark_failed={failed.call_count} "
        f"mark_done={done.call_count}: без переписи диагноза нет — статус не меняем"
    )
    assert banned.call_args.kwargs["ban_kind"] == BAN_KIND_UNKNOWN, (
        f"ban_kind={banned.call_args.kwargs['ban_kind']!r}, ожидали 'unknown'"
    )


def test_platform_dominant_run_is_still_banned() -> None:
    """Контроль на противоположную ошибку: настоящий бан площадки — по-прежнему 'banned'."""
    banned, failed, done = _finish({BAN_KIND_PLATFORM: 41, BAN_KIND_INFRA: 7}, enriched=41)

    assert banned.call_count == 1, (
        f"mark_banned={banned.call_count} (failed={failed.call_count} done={done.call_count}): "
        "площадка отказала в 41 блоке из 48 — это ровно 'banned'"
    )
    assert banned.call_args.kwargs["ban_kind"] == BAN_KIND_PLATFORM, (
        f"ban_kind={banned.call_args.kwargs['ban_kind']!r}, ожидали 'platform'"
    )


# ── #3439: серия отказов НАШЕГО тракта рвёт прогон своим критерием ─────────────
#
# Прогоны 7193/7264/7302: по 1600 попыток, 1-2 карточки, ~1599 отказов сайдкара
# (InvalidIP-шторм на узле), status='done'. Infra ушёл из доли (#3288), а своего
# предохранителя не получил. Порог — max_consecutive_failures (дефолт 25): тот же
# разряд «отказ нашей стороны», что и отказы-не-блоки.

_INFRA_LIMIT = 25


def _infra_breaker() -> BlockRatioBreaker:
    return BlockRatioBreaker(
        window_size=_WINDOW,
        ratio_threshold=_THRESHOLD,
        safety_min=5,
        snapshot_size=1600,
        infra_streak_limit=_INFRA_LIMIT,
    )


def test_infra_streak_aborts_exactly_at_limit() -> None:
    breaker = _infra_breaker()
    breaker.record_success()
    for _ in range(_INFRA_LIMIT - 1):
        breaker.record_block(BAN_KIND_INFRA)
    assert breaker.abort_reason() is None, "24 отказа подряд — ещё не порог 25"

    breaker.record_block(BAN_KIND_INFRA)
    assert breaker.abort_reason() == "infra_streak", (
        f"abort_reason={breaker.abort_reason()!r} на 25-м отказе сайдкара подряд"
    )


@pytest.mark.parametrize("answer", ["success", "platform", "gone"])
def test_any_platform_answer_resets_infra_streak(answer: str) -> None:
    """Площадка ответила (успех / её блок / 404) — тракт жив, серия с нуля."""
    breaker = _infra_breaker()
    for _ in range(_INFRA_LIMIT - 1):
        breaker.record_block(BAN_KIND_INFRA)
    {
        "success": breaker.record_success,
        "platform": lambda: breaker.record_block(BAN_KIND_PLATFORM),
        "gone": breaker.record_neutral,
    }[answer]()
    for _ in range(_INFRA_LIMIT - 1):
        breaker.record_block(BAN_KIND_INFRA)

    assert breaker.abort_reason() is None, (
        f"abort_reason={breaker.abort_reason()!r}: 24 + ответ площадки ({answer}) + 24 — "
        "серии в 25 подряд не было"
    )


@pytest.mark.asyncio
async def test_sidecar_storm_stops_run_as_failed_not_done() -> None:
    """Шторм сайдкара: 1 успех, дальше только отказы — обрыв на 25-м, 'failed'."""
    fetches: list[Any] = [MagicMock()]
    fetches += [AvitoSidecarUnavailableError("browser unavailable 503") for _ in range(1599)]

    counters, runs = await _run_avito(fetches, snapshot_size=1600)

    _assert_identity(counters, expected_attempted=1 + _INFRA_LIMIT)
    assert runs.mark_backfill_finished.call_args is None, (
        "прогон ушёл в backfill-honest-status: там 1 карточка превращает шторм в 'done'"
    )
    failed = runs.mark_failed.call_args
    assert failed is not None, "прогон не финализирован mark_failed"
    written = failed.args[3]
    assert (written["abort_reason"], written["ban_kinds"]) == (
        "infra_streak",
        {BAN_KIND_INFRA: _INFRA_LIMIT},
    ), f"counters={written}"
    assert "тракт недоступен" in failed.args[2], f"причина={failed.args[2]!r}"
    assert runs.mark_banned.call_args is None, "площадка не отказывала — не 'banned'"


@pytest.mark.asyncio
async def test_interleaved_platform_answers_do_not_trip_infra_stop() -> None:
    """Контроль: 24 отказа сайдкара + ответ площадки, трижды — снапшот пройден целиком."""
    fetches: list[Any] = []
    for answer in (AvitoBlockedError("firewall"), MagicMock(), MagicMock()):
        fetches += [AvitoSidecarUnavailableError("503") for _ in range(_INFRA_LIMIT - 1)]
        fetches.append(answer)

    counters, runs = await _run_avito(fetches, snapshot_size=len(fetches))

    _assert_identity(counters, expected_attempted=len(fetches))
    finished = runs.mark_backfill_finished.call_args
    assert finished is not None and finished.args[2].get("abort_reason") is None, (
        f"прогон оборван: counters={finished and finished.args[2]}, "
        f"mark_failed={runs.mark_failed.call_args}"
    )


@pytest.mark.asyncio
async def test_quarter_platform_blocks_run_is_not_aborted() -> None:
    """Контроль: здоровый прогон с 25% настоящих блоков не рвётся ни одним критерием."""
    fetches: list[Any] = []
    for _ in range(10):
        fetches += [MagicMock(), MagicMock(), MagicMock(), AvitoBlockedError("firewall")]

    counters, runs = await _run_avito(fetches, snapshot_size=40)

    _assert_identity(counters, expected_attempted=40)
    finished = runs.mark_backfill_finished.call_args
    assert finished is not None and finished.args[2].get("abort_reason") is None, (
        f"counters={finished and finished.args[2]}: 10 блоков из 40 — рвать не по чему"
    )