"""#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", "path_suffix"),
    [
        ("avito", "www.avito.ru", "/robots.txt"),
        ("cian", "ekb.cian.ru", "/robots.txt"),
        ("yandex", "realty.yandex.ru", "/robots.txt"),
        # apex-домен НЕ годится: через узел id=1 `domclick.ru/robots.txt` отдаёт 200,
        # а рабочий bff-хост — 500. Проба по apex была бы зелёной и бесполезной.
        #
        # #2855: хоста мало — нужен ЗАЩИЩЁННЫЙ ПУТЬ. QRATOR закрывает /api/offers/*,
        # robots.txt того же bff-хоста отдаётся свободно (это знал уже #2800 — см. п.4
        # в шапке: «robots.txt площадка отдаёт и забаненному IP»), поэтому проба по
        # нему зелёная ровно тогда, когда свип получает блок.
        ("domclick", "bff-search-web.domclick.ru", "/api/offers/count/v1"),
    ],
)
async def test_probe_asks_the_working_host_of_each_source(
    monkeypatch: pytest.MonkeyPatch, source: str, must_contain: str, path_suffix: str
) -> None:
    seen: dict[str, Any] = {}
    # Тело ответа зависит от площадки: у robots.txt-источников признак «ресурс отдан» —
    # 'User-agent', у Домклика — ключ 'snippetsCount' живого count-ответа.
    _body = (
        '{"result":{"snippetsCount":678}}' if source == "domclick" else "<pre>User-agent: *</pre>"
    )

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

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

    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"]
    # #2855: путь тоже сторожим — у robots.txt-источников он лёгкий, у Домклика это
    # боевой count-эндпоинт (одно число в ответе, без пагинации и выдачи), то есть
    # нагрузка на площадку остаётся минимальной, а защита — той же, что у работы.
    assert path_suffix in seen["payload"]["url"]
    # Инстанс сайдкара остаётся '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"
    # #2855: сообщение называет КОНКРЕТНЫЙ маркер, потому что он теперь зависит от
    # площадки ('User-agent' у robots.txt-источников, 'snippetsCount' у Домклика).
    # Прежний текст «not robots.txt» стал бы враньём там, где robots.txt и не просили.
    assert "no marker 'User-agent'" 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]


# ── 8. #2855: проба обязана делить с работой ЗАЩИЩЁННЫЙ ПУТЬ, а не только хост ──


@pytest.mark.parametrize(
    ("api_body", "expect_ok"),
    [
        # Блок: площадка отдаёт robots.txt свободно и закрывает /api/offers/* QRATOR'ом.
        # На старом коде проба спрашивала robots.txt → 'User-agent' на месте → ok=True,
        # то есть «пара здорова» ровно там, где свип получает блок. Красный на main.
        ("<html><title>Ошибка</title>qrator captcha</html>", False),
        # Контроль: живой ответ count-эндпоинта → ok=True. Держит двусторонность
        # ВНУТРИ ветки: реализация «маркер не найден никогда» прошла бы первый случай
        # и провалила бы этот, то есть тест не может позеленеть от глухого отказа.
        # На origin/main этот случай тоже красный, но по другой причине — там проба
        # спрашивает robots.txt, `ok` выходит True, и падает проверка АДРЕСА.
        ('{"result":{"snippetsCount":678}}', True),
    ],
)
async def test_domclick_probe_sees_the_block_that_robots_txt_hides(
    monkeypatch: pytest.MonkeyPatch, api_body: str, expect_ok: bool
) -> None:
    """Площадка: robots.txt отдаёт всем, /api/offers/* закрывает. Проба обязана увидеть блок.

    Замер прода 13.08.2026: 04:30 healthcheck `pair_banned=0` → 05:02 свип
    `QRATOR block during rooms='1'` и узел 11 в бан. За сутки 64 проверки пар и
    5 банов на все четыре площадки при ежедневном блоке Домклика.
    """
    seen: dict[str, Any] = {}

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

        def __init__(self, body: str) -> None:
            self._body = body

        def json(self) -> dict[str, str]:
            return {"html": self._body}

    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
            asked = str(json["url"])
            # Площадка ведёт себя как в проде: лёгкий путь открыт, боевой — закрыт.
            if asked.endswith("/robots.txt"):
                return _Resp("<pre>User-agent: *</pre>")
            return _Resp(api_body)

    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="domclick"
    )

    assert ok is expect_ok
    if not expect_ok:
        # Тракт узла исправен (200 пришёл) — негодна ПАРА, значит "page", не "proxy".
        assert fail_kind == "page"
    # Спрашивали именно боевой путь, а не robots.txt того же хоста.
    assert "/api/offers/" in seen["payload"]["url"]