"""#3288: бан площадки снимает узел АДРЕСНО (узел×источник), а не глобально по здоровью.

Прод-замер 31.08–01.09: `resolve_proxy_url` отказывал ВСЕМ четырём источникам
(`ProxyPoolExhaustedError pool_total=7 banned_for_source=0 unhealthy_or_disabled=7`),
при том что живая проба тех же узлов проходила — они стояли ровно на потолке
`MAX_CONSECUTIVE_FAILS`. Механизм: сайдкар на бан-странице Авито («доступ ограничен:
проблема с IP») отвечает `ban_page: true` в теле (кодом 403 с #3288 п.4, кодом 500 у
сайдкара старой сборки), клиент поднимает
`SidecarBanPageError` — но она подкласс `httpx.HTTPStatusError`, и общий
`except Exception` в `_post_fetch` звал `_report_fetch_result(False)` →
`mark_health(ok=False)`. `mark_health` — решение ГЛОБАЛЬНОЕ для пула: три бан-страницы
Авито выбивали узел и у Яндекса, и у Циана, и у Домклика, которые не возражали вовсе.

Что закрепляем по ЗНАЧЕНИЮ:
  (а) подтверждённая бан-страница → `AvitoBlockedError`, `ban_kind_of_exception` ==
      'platform', `mark_banned(lease, source='avito')` вызван;
  (б) на бан-странице `mark_health` НЕ вызван вовсе, а на настоящем транспортном сбое
      (ReadTimeout) — вызван с ok=False, как и раньше;
  (в) `NoProxyAvailableError` (пул пуст, к площадке не ходили) — ни health-fail, ни бан;
      через обёртку `fetch_detail` он остаётся распознаваемым ПО ТИПУ в цепочке причин
      (приём `_iter_causes`), а не по подстроке «no proxy available» (#3272);
  (г) порядок веток `except`: ban-ветка стоит ДО общего `except Exception`. Пин по
      значению — тот же HTTP 500 с маркером и без него разводит судьбу узла;
  (д) при РОТАЦИИ после бана свежий узел не наказывается за чужой отказ: дедуп по
      `_banned_lease_id` сбрасывается взятием нового lease, поэтому единственный
      допустимый рапорт — из самого фетчера, до ротации. Второй такой же рапорт
      сверху (из провайдера) банил бы узел, который к площадке не ходил;
  (е) бан-страница стоит РОВНО один POST: `SidecarBanPageError` — подкласс
      `httpx.HTTPError`, и ретрай `fetch()` забирал её себе, удваивая fail-streak.

Зеркалит стиль tests/test_3196_domclick_ban_kind.py и
tests/test_3283_avito_sidecar_ban_is_platform_ban.py; харнесс пула (фейковый провайдер,
подмена `_client` после `__aenter__`) — из tests/test_kit_browser_fetcher_proxy_pool.py.
"""

from __future__ import annotations

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

import httpx
import pytest

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

from scraper_kit.avito_exceptions import AvitoBlockedError, AvitoSidecarUnavailableError
from scraper_kit.browser_fetcher import (
    _LEASE_ROTATE_AFTER_FAILS,
    BrowserFetcher,
    SidecarBanPageError,
)
from scraper_kit.contracts import ProxyLease
from scraper_kit.orchestration.pipeline import ban_kind_of_exception
from scraper_kit.providers.avito.detail import fetch_detail
from scraper_kit.proxy_errors import NoProxyAvailableError

from app.services.scrape_runs import BAN_KIND_INFRA, BAN_KIND_PLATFORM

_ITEM_URL = "https://www.avito.ru/ekaterinburg/kvartiry/x_1234567890"
_LEASE = ProxyLease(id=13, url="http://node13:8080", kind="http")


class _FakePool:
    """ProxyProvider-заглушка: пишет, ЧТО именно узнал пул об исходе (зеркало
    _FakeProxyProvider из test_kit_browser_fetcher_proxy_pool.py)."""

    def __init__(self, lease: ProxyLease | None = _LEASE) -> None:
        self._lease = lease
        self.health: list[tuple[int, bool]] = []
        self.banned: list[tuple[int, str]] = []
        self.touched: list[int] = []
        self.released: list[int] = []

    def acquire(self, provider: str) -> ProxyLease | None:
        return self._lease

    def release(self, lease: ProxyLease) -> None:
        self.released.append(lease.id)

    def mark_health(self, lease: ProxyLease, ok: bool, **_: Any) -> None:
        self.health.append((lease.id, ok))

    def touch(self, lease: ProxyLease) -> None:
        self.touched.append(lease.id)

    def mark_banned(self, lease: ProxyLease, *, source: str) -> None:
        self.banned.append((lease.id, source))


_SIDECAR_BAN_CODES = [403, 500]


def _ban_page_response(sidecar_status: int = 403) -> httpx.Response:
    """РОВНО то, что отдаёт сайдкар на распознанной бан-странице (browser/server.py):
    `ban_page: true` + статус целевой навигации. Ответ настоящий, а не мок, — чтобы
    проба шла тем же трактом, что и прод (_raise_for_sidecar_status разбирает тело).

    `sidecar_status` — код ОТВЕТА САЙДКАРА, а не площадки: 403 с #3288 п.4, 500 у
    сайдкара старой сборки. `tradein-browser` — отдельный образ со своим шагом
    деплоя, поэтому версии сайдкара и backend штатно разъезжаются на часы; оба кода
    обязаны давать один диагноз, и тесты ниже прогоняются по `_SIDECAR_BAN_CODES`.
    """
    request = httpx.Request("POST", "http://browser:3000/fetch")
    return httpx.Response(
        sidecar_status,
        json={
            "error": "BanPageDetectedError: бан-страница (проблема с IP)",
            "ban_page": True,
            "status": 403,
        },
        request=request,
    )


def _plain_500_response() -> httpx.Response:
    """Настоящая внутренняя ошибка сайдкара: тот же код 500, но БЕЗ маркера бана."""
    request = httpx.Request("POST", "http://browser:3000/fetch")
    return httpx.Response(
        500,
        json={"error": "Error: Page.goto: NS_ERROR_PROXY_BAD_GATEWAY"},
        request=request,
    )


async def _fetcher(post_result: Any, pool: _FakePool) -> BrowserFetcher:
    """Фетчер с живым lease из пула и замоканным httpx-клиентом.

    post_result — готовый httpx.Response (вернётся) либо Exception (будет брошен).
    """
    bf = BrowserFetcher(
        endpoint="http://browser:3000",
        source="avito",
        proxy_provider=pool,
        use_pool=True,
        environment="production",
    )
    await bf.__aenter__()
    client = MagicMock()
    if isinstance(post_result, Exception):
        client.post = AsyncMock(side_effect=post_result)
    else:
        client.post = AsyncMock(return_value=post_result)
    client.aclose = AsyncMock(return_value=None)
    bf._client = client
    return bf


def _iter_causes(exc: BaseException) -> list[BaseException]:
    """Цепочка причин без зацикливания — копия приёма из domclick_detail_backfill.py."""
    seen: set[int] = set()
    out: list[BaseException] = []
    cur: BaseException | None = exc
    while cur is not None and id(cur) not in seen:
        out.append(cur)
        seen.add(id(cur))
        cur = cur.__cause__ or cur.__context__
    return out


# ── (а) бан-страница → отказ ПЛОЩАДКИ + адресный бан пары «узел×источник» ──────


@pytest.mark.parametrize("sidecar_status", _SIDECAR_BAN_CODES)
async def test_ban_page_raises_platform_block_and_bans_node_for_avito_only(
    sidecar_status: int,
) -> None:
    """Бан-страница: AvitoBlockedError, диагноз 'platform', mark_banned(13, 'avito').

    Фальсификация: до фикса SidecarBanPageError доезжала до общего except в
    _post_fetch → mark_health(ok=False), а строки в scrape_proxy_source_bans не
    появлялось вовсе (за 7 суток 0 записей для source=avito при 58 прогонах banned).
    Вторая фальсификация (#3288 п.4): заведи в `_sidecar_ban_page_status` гейт
    «признавать ban_page только при 500» — 403-кейс покраснеет по значению
    (AvitoSidecarUnavailableError вместо бана), т.е. в час рассинхрона версий
    отказ площадки снова уезжал бы в инфра-ветку.
    """
    pool = _FakePool()
    bf = await _fetcher(_ban_page_response(sidecar_status), pool)

    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
        with pytest.raises(AvitoBlockedError) as ei:
            await fetch_detail(_ITEM_URL, browser_fetcher=bf)

    assert ban_kind_of_exception(ei.value) == BAN_KIND_PLATFORM
    assert pool.banned == [(13, "avito")], (
        "узел должен быть забанен РОВНО один раз и РОВНО для avito: одно событие "
        "доезжает до report_ban трижды (POST, ретрай fetch(), провайдер), а каждый "
        "mark_banned растит ban_count и удлиняет отдых пары"
    )


# ── (б) здоровье узла: бан-страница молчит, транспортный сбой — говорит ────────


@pytest.mark.parametrize("sidecar_status", _SIDECAR_BAN_CODES)
async def test_ban_page_does_not_mark_health_failure(sidecar_status: int) -> None:
    """Ключевое значение issue: на бан-странице mark_health НЕ вызывается вовсе.

    Именно этот вызов гнал узел к MAX_CONSECUTIVE_FAILS и выбивал его из выдачи
    ВСЕМ источникам (замер 01.09: узлы 9, 13, 14 на потолке при banned_for_source=0).
    """
    pool = _FakePool()
    bf = await _fetcher(_ban_page_response(sidecar_status), pool)

    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
        with pytest.raises(AvitoBlockedError):
            await fetch_detail(_ITEM_URL, browser_fetcher=bf)

    assert pool.health == [], f"бан площадки не должен трогать здоровье узла: {pool.health}"
    assert pool.touched, "heartbeat (touch) при этом обязан идти — аренда живая"


async def test_real_transport_failure_still_marks_health_failure() -> None:
    """Контроль к предыдущему: настоящий транспортный сбой (таймаут) health-fail'ит,
    как и до фикса. Без этой пары фикс неотличим от «выключили учёт здоровья»."""
    pool = _FakePool()
    request = httpx.Request("POST", "http://browser:3000/fetch")
    bf = await _fetcher(httpx.ReadTimeout("sidecar timeout", request=request), pool)

    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
        with pytest.raises(AvitoSidecarUnavailableError):
            await fetch_detail(_ITEM_URL, browser_fetcher=bf)

    assert pool.health == [(13, False), (13, False)], (
        f"таймаут — сбой НАШЕГО тракта, он обязан остаться в здоровье узла "
        f"(две записи: fetch() делает один ретрай): {pool.health}"
    )
    assert pool.banned == [], "таймаут не бан: площадка себя не показала"


# ── (в) пустой пул — ни бан, ни health-fail ───────────────────────────────────


async def test_empty_pool_is_neither_ban_nor_health_failure() -> None:
    """`NoProxyAvailableError` поднимается ДО POST'а: узла нет, площадки не видели."""
    pool = _FakePool(lease=None)
    bf = BrowserFetcher(
        endpoint="http://browser:3000",
        source="avito",
        proxy_provider=pool,
        use_pool=True,
        environment="production",
    )

    with pytest.raises(NoProxyAvailableError):
        await bf.__aenter__()

    assert pool.health == []
    assert pool.banned == []


async def test_empty_pool_through_wrapper_stays_infra_and_recognizable_by_type() -> None:
    """Обёртка `fetch_detail` даёт AvitoSidecarUnavailableError ('infra'), но исходный
    тип остаётся в цепочке причин — часть B (стоп «нечем ходить») ловит его через
    `_iter_causes`, а НЕ подстрокой «no proxy available» (на ней обжёгся #3272)."""
    bf = MagicMock()
    bf.fetch = AsyncMock(side_effect=NoProxyAvailableError("avito"))

    with pytest.raises(AvitoSidecarUnavailableError) as ei:
        await fetch_detail(_ITEM_URL, browser_fetcher=bf)

    assert ban_kind_of_exception(ei.value) == BAN_KIND_INFRA
    assert any(isinstance(c, NoProxyAvailableError) for c in _iter_causes(ei.value))
    bf.report_ban.assert_not_called()


# ── (г) порядок веток except ──────────────────────────────────────────────────


def test_sidecar_ban_page_is_a_subclass_of_httpx_error() -> None:
    """Причина, по которой порядок веток вообще имеет значение (#3287): широкая ловля
    забирает бан себе. Если это перестанет быть правдой — тесты выше проверяют не то,
    что думают."""
    assert issubclass(SidecarBanPageError, httpx.HTTPStatusError)


async def test_same_500_different_marker_gives_different_node_fate() -> None:
    """Пин порядка веток по значению: ОДИН и тот же HTTP 500 разводит судьбу узла
    ровно по маркеру `ban_page` в теле. Плоская 500 — health-fail без бана; поставь
    ban-ветку после `except Exception` — и бан-страница даст тот же исход, что здесь."""
    pool = _FakePool()
    bf = await _fetcher(_plain_500_response(), pool)

    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
        with pytest.raises(AvitoSidecarUnavailableError):
            await fetch_detail(_ITEM_URL, browser_fetcher=bf)

    assert pool.health == [(13, False), (13, False)]
    assert pool.banned == []


# ── (д) ротация после бана: наказан тот узел, который к площадке ходил ─────────


class _RotatingFakePool(_FakePool):
    """Пул с НЕСКОЛЬКИМИ узлами: `acquire()` выдаёт их по очереди.

    Отличие от `_FakePool` выше (вечный узел 13) несущее: дедуп рапортов в
    `BrowserFetcher.report_ban` держится на `_banned_lease_id`, а тот сбрасывается
    взятием нового lease — на неподвижном пуле дефект «второй рапорт банит свежий
    узел» физически не проявляется, и тест его не видел бы.
    """

    def __init__(self, leases: list[ProxyLease]) -> None:
        super().__init__(lease=None)
        self._queue = list(leases)
        self.acquired: list[int | None] = []

    def acquire(self, provider: str) -> ProxyLease | None:
        lease = self._queue.pop(0) if self._queue else None
        self.acquired.append(lease.id if lease is not None else None)
        return lease


async def test_rotation_after_ban_does_not_ban_the_fresh_node() -> None:
    """Узел 13 поймал бан-страницу, fail-streak сменил его на 14 — забанен ТОЛЬКО 13.

    Прод-сценарий: узел уже сыпался (streak на единицу ниже потолка), бан-страница
    добивает его до ротации. Фальсификация: верни `browser_fetcher.report_ban(...)`
    в ветку `SidecarBanPageError` в providers/avito/detail.py — к тому кадру lease
    уже сменился, дедуп по `_banned_lease_id` сброшен взятием нового lease, и
    значение станет [(13, 'avito'), (14, 'avito')]: свежий узел получает 6-часовой
    отдых (с эскалацией ban_count) за отказ, которого он не видел. При трёх узлах
    в пуле это выбивает две трети выдачи с одной бан-страницы.
    """
    pool = _RotatingFakePool([_LEASE, ProxyLease(id=14, url="http://node14:8080", kind="http")])
    bf = await _fetcher(_ban_page_response(), pool)
    bf._lease_fail_streak = _LEASE_ROTATE_AFTER_FAILS - 1

    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
        with pytest.raises(AvitoBlockedError):
            await fetch_detail(_ITEM_URL, browser_fetcher=bf)

    assert pool.acquired == [13, 14], f"ротация обязана была произойти: {pool.acquired}"
    assert pool.banned == [(13, "avito")], (
        f"забанен должен быть узел, который сходил на площадку и получил отказ, "
        f"а не тот, что пришёл ему на смену: {pool.banned}"
    )


# ── (е) одна бан-страница — один POST ─────────────────────────────────────────


@pytest.mark.parametrize("sidecar_status", _SIDECAR_BAN_CODES)
async def test_ban_page_costs_exactly_one_post(sidecar_status: int) -> None:
    """Ретрай `fetch()` не имеет права трогать бан-страницу: это ответ площадки.

    Фальсификация: убери `except SidecarBanPageError: raise` перед
    `except (httpx.HTTPError, ...)` в `fetch()` — станет 2 POST'а, а с ними и +2 к
    `_lease_fail_streak` вместо +1 (ротация вдвое раньше задуманного).
    """
    pool = _FakePool()
    bf = await _fetcher(_ban_page_response(sidecar_status), pool)

    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
        with pytest.raises(AvitoBlockedError):
            await fetch_detail(_ITEM_URL, browser_fetcher=bf)

    assert bf._client.post.await_count == 1, (
        f"бан-страницу ретраить нечем — площадка уже ответила: "
        f"{bf._client.post.await_count} POST'ов"
    )


# ── (ж) пустой пул при ротации не подменяет диагноз ───────────────────────────


async def test_empty_pool_during_post_ban_rotation_keeps_platform_diagnosis() -> None:
    """Ротация — best-effort: её `NoProxyAvailableError` не должна съесть бан.

    Второго узла в пуле нет, поэтому `_acquire_lease()` внутри ротации падает.
    Фальсификация: убери `except NoProxyAvailableError` в `report_platform_ban` —
    наверх уедет она вместо `SidecarBanPageError`, провайдер завернёт её в
    AvitoSidecarUnavailableError, и подтверждённый отказ площадки попадёт в
    `scrape_runs.ban_kind` как 'infra' (ровно та подмена, что чинил #3283).
    """
    pool = _RotatingFakePool([_LEASE])
    bf = await _fetcher(_ban_page_response(), pool)
    bf._lease_fail_streak = _LEASE_ROTATE_AFTER_FAILS - 1

    with patch("scraper_kit.browser_fetcher.asyncio.sleep", AsyncMock()):
        with pytest.raises(AvitoBlockedError) as ei:
            await fetch_detail(_ITEM_URL, browser_fetcher=bf)

    assert ban_kind_of_exception(ei.value) == BAN_KIND_PLATFORM
    assert pool.banned == [(13, "avito")], f"бан обязан быть отрапортован до ротации: {pool.banned}"