"""#2800 — браузерная проба спрашивает КАЖДУЮ площадку, вердикт пишется на ПАРУ.

Продолжение #2723 на другой оси. Там проба ходила не тем транспортом; здесь —
верным транспортом, но всегда на один адрес (robots.txt Авито), поэтому её зелёный
ответ означал «узел годен для Авито», а читался как «узел годен вообще».

Замер на проде 09.08.2026 (тот же тракт: сайдкар → camoufox с этим прокси → навигация):

    узел | affinity | avito | ekb.cian.ru         | realty.ya.ru | bff-search-web.domclick.ru
    -----+----------+-------+---------------------+--------------+---------------------------
      1  | domclick | 200   | 200 «Ошибка — Циан» | 200          | 500 NS_ERROR_PROXY_BAD_GATEWAY
     10  | any      | 200   | 200                 | 200          | 200

Что сторожится (каждый тест ниже падает на коде до фикса):

  1. ГЛАВНОЕ: узел, зелёный по Авито и мёртвый по Домклику, ОТБРАКОВЫВАЕТСЯ для
     Домклика и остаётся первосортным для Авито. До фикса `acquire('domclick')`
     выдавал его как ни в чём не бывало.
  2. Адрес пробы — рабочий хост КАЖДОЙ площадки, а не один зашитый и не apex-домен
     (`domclick.ru` через тот же узел отвечает 200 — проба по нему была бы зелёной).
  3. HTTP 200 с заглушкой вместо robots.txt — это отказ пары, а не успех.
  4. Успешная проба снимает ТОЛЬКО свою строку бана: robots.txt площадка отдаёт и
     забаненному IP, и гасить им вердикт живого сбора нельзя (та же ошибка, что
     «дешёвая ipify стирает браузерный вердикт» в #2723).
  5. Отказ ОДНОЙ площадки не пятнает узел глобально; отказ ВСЕХ — пятнает (узловой
     вердикт #2723 сохранён).
  6. Лежащий сайдкар не пишет ни одного бана пары (#2686-класс).
  7. Цена такта названа числом и закреплена: узлов × обслуживаемых источников.
"""

from __future__ import annotations

import os

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

from datetime import UTC, datetime, timedelta
from typing import Any

import pytest
import scraper_kit.browser_fetcher as bf

from app.services import proxy_pool
from app.services.proxy_pool import BROWSER_UNFIT_THRESHOLD, acquire, release
from tests.services.test_proxy_pool import FakeSession, _proxy

# Живой замер с прода 09.08.2026 — узел h1 мёртв для Домклика и годен остальным.
_LIVE_MATRIX: dict[tuple[str, str], tuple[bool, str | None, str]] = {
    ("http://u:p@h1:8080", "domclick"): (
        False,
        "proxy",
        '{"error": "Error: Page.goto: NS_ERROR_PROXY_BAD_GATEWAY"}',
    ),
}


def _patch_probes(
    monkeypatch: pytest.MonkeyPatch,
    matrix: dict[tuple[str, str], tuple[bool, str | None, str]],
    calls: list[tuple[str, str]] | None = None,
    *,
    default: tuple[bool, str | None, str] = (True, None, "html_len=16477"),
) -> None:
    """ipify всегда зелёная; браузерная проба отвечает по матрице (прокси, источник)."""

    async def _fake_http(url: str) -> tuple[bool, str | None, int | None, str | None]:
        return True, "1.2.3.4", 10, None

    async def _fake_browser(
        endpoint: str, proxy_url: str, **kw: Any
    ) -> tuple[bool, str | None, str]:
        source = str(kw.get("source", "avito"))
        if calls is not None:
            calls.append((proxy_url, source))
        return matrix.get((proxy_url, source), default)

    monkeypatch.setattr(proxy_pool, "_probe_proxy", _fake_http)
    monkeypatch.setattr(proxy_pool._settings, "use_proxy_pool_browser", True)
    monkeypatch.setattr(bf, "probe_proxy_via_browser", _fake_browser)


# ── 1. главное: вердикт разведён по источникам ───────────────────────────────


async def test_node_dead_for_domclick_is_not_issued_to_domclick(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Узел зелёный по Авито и мёртвый по Домклику: Домклику НЕ выдаём, Авито — выдаём.

    Именно этот сценарий четверо суток давал `domclick_city_sweep` ноль лотов при
    `browser_fail_streak=0` и свежей пробе.
    """
    _patch_probes(monkeypatch, _LIVE_MATRIX)
    db = FakeSession([_proxy(1), _proxy(2)])

    counters = await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    # Домклику достаётся только исправный узел…
    first = acquire(db, "domclick")  # type: ignore[arg-type]
    assert first is not None and first.id == 2

    # …а когда он занят, Домклик остаётся БЕЗ прокси, вместо того чтобы получить
    # заведомо мёртвый узел 1 и сжечь на нём прогон. ЭТО и есть поломка, ради которой
    # заведена задача: до фикса здесь выдавался узел 1 с browser_fail_streak=0.
    assert acquire(db, "domclick") is None, (  # type: ignore[arg-type]
        "Домклику выдан узел, у которого рабочий хост Домклика отвечает NS_ERROR_PROXY_BAD_GATEWAY"
    )

    # Для Авито тот же узел 1 — полноценный кандидат: бан у пары, не у узла.
    lease = acquire(db, "avito")  # type: ignore[arg-type]
    assert lease is not None and lease.id == 1

    ban = db._ban(1, "domclick")
    assert ban is not None
    assert ban["reason"] == "probe:browser", "строку должна опознавать сама проба"
    assert db._ban(1, "avito") is None and db._ban(2, "domclick") is None
    assert counters["pair_banned"] == 1


async def test_one_dead_platform_does_not_stain_the_node_globally(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Узловой вердикт остаётся про узел: одна мёртвая площадка его не помечает."""
    _patch_probes(monkeypatch, _LIVE_MATRIX)
    db = FakeSession([_proxy(1), _proxy(2)])

    await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    row = db._by_id(1)
    assert row["browser_unfit_since"] is None
    assert row["browser_fail_streak"] == 0
    assert row["enabled"] is True and row["consecutive_fails"] == 0


async def test_all_platforms_dead_still_marks_the_node(monkeypatch: pytest.MonkeyPatch) -> None:
    """Обратная сторона: транспорт не доходит НИКУДА → это уже диагноз узлу (#2723 цел)."""
    dead_everywhere = {
        ("http://u:p@h1:8080", src): (False, "proxy", "503 browser unavailable")
        for src in ("avito", "cian", "yandex", "domclick")
    }
    _patch_probes(monkeypatch, dead_everywhere)
    db = FakeSession([_proxy(1), _proxy(2)])

    for _ in range(BROWSER_UNFIT_THRESHOLD):
        db._by_id(1)["browser_check_at"] = None  # снять гейт такта, ускорить подтверждение
        await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    assert db._by_id(1)["browser_unfit_since"] is not None
    assert db._by_id(2)["browser_unfit_since"] is None


# ── 2-3. адрес пробы и «200 ≠ ответ площадки» ────────────────────────────────


@pytest.mark.parametrize(
    ("source", "must_contain"),
    [
        ("avito", "www.avito.ru"),
        ("cian", "ekb.cian.ru"),
        ("yandex", "realty.yandex.ru"),
        # apex-домен НЕ годится: через узел id=1 `domclick.ru/robots.txt` отдаёт 200,
        # а рабочий bff-хост — 500. Проба по apex была бы зелёной и бесполезной.
        ("domclick", "bff-search-web.domclick.ru"),
    ],
)
async def test_probe_asks_the_working_host_of_each_source(
    monkeypatch: pytest.MonkeyPatch, source: str, must_contain: str
) -> None:
    seen: dict[str, Any] = {}

    class _Resp:
        status_code = 200
        text = '{"html": "<pre>User-agent: *</pre>"}'

        @staticmethod
        def json() -> dict[str, str]:
            return {"html": "<pre>User-agent: *</pre>"}

    class _Client:
        def __init__(self, **_kw: Any) -> None: ...

        async def __aenter__(self) -> _Client:
            return self

        async def __aexit__(self, *_: object) -> None:
            return None

        async def post(self, url: str, json: dict[str, Any]) -> _Resp:
            seen["payload"] = json
            return _Resp()

    monkeypatch.setattr(bf.httpx, "AsyncClient", _Client)
    ok, _fail_kind, _detail = await bf.probe_proxy_via_browser(
        "http://tradein-browser:3000", "http://u:p@node:8080", source=source
    )

    assert ok is True
    assert must_contain in seen["payload"]["url"]
    assert seen["payload"]["url"].endswith("/robots.txt")  # нагрузки на площадку нет
    # Инстанс сайдкара остаётся 'generic' — проба не отбирает лок у боевой сессии.
    assert seen["payload"]["source"] == "generic"


async def test_stub_page_with_status_200_is_a_failure(monkeypatch: pytest.MonkeyPatch) -> None:
    """374 КБ «Ошибка — Циан» с кодом 200 — это отказ пары, а не успех пробы."""
    stub = "<!DOCTYPE html><html><head><title>Ошибка - Циан</title></head><body>…</body></html>"

    class _Resp:
        status_code = 200
        text = "{}"

        @staticmethod
        def json() -> dict[str, str]:
            return {"html": stub}

    class _Client:
        def __init__(self, **_kw: Any) -> None: ...

        async def __aenter__(self) -> _Client:
            return self

        async def __aexit__(self, *_: object) -> None:
            return None

        async def post(self, url: str, json: dict[str, Any]) -> _Resp:
            return _Resp()

    monkeypatch.setattr(bf.httpx, "AsyncClient", _Client)
    ok, fail_kind, detail = await bf.probe_proxy_via_browser(
        "http://tradein-browser:3000", "http://u:p@node:8080", source="cian"
    )

    assert ok is False
    # Тракт узла исправен — виновата ПАРА: площадка не отдала ресурс этому exit-IP.
    assert fail_kind == "page"
    assert "not robots.txt" in detail


async def test_stub_page_bans_the_pair(monkeypatch: pytest.MonkeyPatch) -> None:
    """«page» не принадлежит узлу (#2723), но принадлежит паре — собирать через неё нельзя."""
    _patch_probes(
        monkeypatch,
        {("http://u:p@h1:8080", "cian"): (False, "page", "not robots.txt (html_len=374168)")},
    )
    db = FakeSession([_proxy(1), _proxy(2)])

    await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    assert db._ban(1, "cian") is not None
    assert db._by_id(1)["browser_unfit_since"] is None  # узел не при чём


# ── 4. проба снимает только свою строку ──────────────────────────────────────


async def test_probe_clears_only_its_own_ban(monkeypatch: pytest.MonkeyPatch) -> None:
    """Зелёный robots.txt снимает вердикт ПРОБЫ и не трогает бан, распознанный сбором.

    robots.txt площадка отдаёт и забаненному IP — разрешить дешёвому успеху гасить
    дорогой вердикт значило бы повторить #2723 на паре.
    """
    later = datetime.now(UTC) + timedelta(hours=6)
    db = FakeSession(
        [_proxy(1), _proxy(2)],
        bans=[
            {
                "proxy_id": 1,
                "source": "avito",
                "banned_until": later,
                "ban_count": 1,
                "reason": "banned:avito",  # распознан боевым сбором (капча/QRATOR)
            },
            {
                "proxy_id": 1,
                "source": "cian",
                "banned_until": later,
                "ban_count": 1,
                "reason": "probe:browser",  # прошлый вердикт самой пробы
            },
        ],
    )
    _patch_probes(monkeypatch, {})  # все площадки отвечают

    counters = await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    avito = db._ban(1, "avito")
    assert avito is not None, "чужой бан проба снимать не имеет права"
    assert (avito["reason"], avito["ban_count"]) == ("banned:avito", 1), "и не переписывать"
    assert db._ban(1, "cian") is None, "свой вердикт проба обязана снять"
    assert counters["pair_cleared"] == 1


# ── 4b. …и не присваивает чужую (дефект #2803, реализовался на проде) ─────────


async def test_probe_does_not_steal_a_live_ban(monkeypatch: pytest.MonkeyPatch) -> None:
    """Упавшая проба НЕ переписывает активный бан, поставленный боевым сбором.

    Прод 09.08.2026, пара (1, cian): строка `banned:cian, ban_count=1, до 00:21` после
    упавшей пробы стала `probe:browser, ban_count=2, до 07:43`. Фильтр «снимаю только
    своё» при этом цел, но защищать перестаёт: присвоенная строка уже «своя», и
    следующая успешная проба сняла бы ею бан, который площадка поставила по-настоящему.
    Плюс сама метка перестаёт быть свидетельством («нас отбили» неотличимо от «мы не
    смогли», #2764), а ban_count складывает события разного рода в одну эскалацию —
    отдых пары вырос с 6 ч до 12 ч.
    """
    until = datetime.now(UTC) + timedelta(hours=6)
    db = FakeSession(
        [_proxy(1), _proxy(2)],
        bans=[
            {
                "proxy_id": 1,
                "source": "cian",
                "banned_until": until,
                "ban_count": 1,
                "reason": "banned:cian",  # боевой сбор: Циан отдал заглушку
            }
        ],
    )
    _patch_probes(
        monkeypatch,
        {("http://u:p@h1:8080", "cian"): (False, "page", "not robots.txt (html_len=374168)")},
    )

    counters = await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    ban = db._ban(1, "cian")
    assert ban is not None
    assert ban["reason"] == "banned:cian", "проба присвоила себе бан боевого сбора"
    assert ban["ban_count"] == 1, "два события разного рода посчитаны одной эскалацией"
    assert ban["banned_until"] == until, "чужой срок проба не пересчитывает (может и укоротить)"
    assert counters["pair_banned"] == 0, "счётчик не должен объявлять баном то, чего не записал"


async def test_live_ban_takes_over_the_probe_row(monkeypatch: pytest.MonkeyPatch) -> None:
    """Зеркало намеренно НЕ симметрично: боевой сбор строку пробы перехватывает.

    Его вердикт сильнее — площадка отбила нас именно сейчас, — пара остаётся забаненной,
    а метка становится точнее. Если запретить и ему, строка останется за пробой, и её же
    зелёный robots.txt снесёт настоящий бан площадки: тот же дефект, только зеркально.
    """
    db = FakeSession(
        [_proxy(1), _proxy(2)],
        bans=[
            {
                "proxy_id": 1,
                "source": "cian",
                "banned_until": datetime.now(UTC) + timedelta(hours=6),
                "ban_count": 1,
                "reason": "probe:browser",
            }
        ],
    )

    proxy_pool.mark_banned(db, 1, source="cian")  # type: ignore[arg-type]

    assert db._ban(1, "cian")["reason"] == "banned:cian"

    # …и с этой минуты зелёная проба его не снимет — ради чего перехват и нужен.
    _patch_probes(monkeypatch, {})
    await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]
    assert db._ban(1, "cian") is not None


# ── 5-6. чужие отказы ────────────────────────────────────────────────────────


async def test_sidecar_outage_bans_nothing(monkeypatch: pytest.MonkeyPatch) -> None:
    """Лежащий сайдкар не должен забанить разом все пары (#2686-класс)."""
    down = {
        (f"http://u:p@h{pid}:8080", src): (False, "sidecar", "ConnectError")
        for pid in (1, 2)
        for src in ("avito", "cian", "yandex", "domclick")
    }
    _patch_probes(monkeypatch, down)
    db = FakeSession([_proxy(1), _proxy(2)])

    counters = await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    assert counters["pair_banned"] == 0
    assert db.bans == []
    assert db._by_id(1)["browser_unfit_since"] is None


# ── 7. цена такта ────────────────────────────────────────────────────────────


async def test_probe_cost_is_nodes_times_servable_sources(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Крест ограничен источниками, которым узел вообще может достаться.

    Диагностика не должна превращаться в нагрузку: узел с выделенной affinity
    спрашивает ОДНУ площадку, 'any' — все четыре. На проде это 4 узла × 4 источника
    = 16 навигаций за такт (раз в BROWSER_PROBE_MINUTES), то есть 16 robots.txt на
    площадку в сутки против ~1000 боевых /fetch.
    """
    calls: list[tuple[str, str]] = []
    _patch_probes(monkeypatch, {}, calls)
    db = FakeSession([_proxy(1, affinity="any"), _proxy(2, affinity="domclick")])

    await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    assert sorted(s for url, s in calls if url.endswith("h1:8080")) == [
        "avito",
        "cian",
        "domclick",
        "yandex",
    ]
    assert [s for url, s in calls if url.endswith("h2:8080")] == ["domclick"]
    assert len(calls) == 5


async def test_confirmed_failure_needs_a_second_look(monkeypatch: pytest.MonkeyPatch) -> None:
    """Транспортный отказ пары подтверждается повтором — запуск camoufox бывает флаки.

    Повтор идёт по уже поднятому браузеру и только на отказах, поэтому цена такта из
    теста выше не меняется, пока всё зелено.
    """
    calls: list[tuple[str, str]] = []
    _patch_probes(monkeypatch, _LIVE_MATRIX, calls)
    db = FakeSession([_proxy(1), _proxy(2)])

    await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    assert calls.count(("http://u:p@h1:8080", "domclick")) == 2
    assert calls.count(("http://u:p@h1:8080", "avito")) == 1


async def test_flaky_failure_does_not_ban_the_pair(monkeypatch: pytest.MonkeyPatch) -> None:
    """Один провал, второй заход зелёный → бан пары не пишется."""
    attempts: dict[str, int] = {}

    async def _fake_http(url: str) -> tuple[bool, str | None, int | None, str | None]:
        return True, "1.2.3.4", 10, None

    async def _flaky(endpoint: str, proxy_url: str, **kw: Any) -> tuple[bool, str | None, str]:
        source = str(kw.get("source", "avito"))
        key = f"{proxy_url}|{source}"
        attempts[key] = attempts.get(key, 0) + 1
        if source == "domclick" and proxy_url.endswith("h1:8080") and attempts[key] == 1:
            return False, "proxy", "503 browser unavailable"
        return True, None, "html_len=150"

    monkeypatch.setattr(proxy_pool, "_probe_proxy", _fake_http)
    monkeypatch.setattr(proxy_pool._settings, "use_proxy_pool_browser", True)
    monkeypatch.setattr(bf, "probe_proxy_via_browser", _flaky)
    db = FakeSession([_proxy(1), _proxy(2)])

    counters = await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]

    assert counters["pair_banned"] == 0
    assert db._ban(1, "domclick") is None
    lease = acquire(db, "domclick")  # type: ignore[arg-type]
    assert lease is not None
    release(db, lease.id)  # type: ignore[arg-type]