"""#2764: ban_kind перестаёт проставляться по умолчанию — метка либо доказана, либо 'unknown'.

Замер на проде 2026-08-07 (после миграции 218 и мержа #2711):
    SELECT ban_kind, count(*) FROM scrape_runs WHERE status='banned' GROUP BY 1;
      infra 92 / platform 41
Из них ДВЕ строки написаны уже в рантайме — и обе получили 'platform' по дефолту
сигнатуры mark_banned, из общего финализатора backfill-задач, который диагноз не
передаёт вовсе (3306 avito 12:42, 3313 domclick 15:20). Для avito причина 1600
отказов вообще не устанавливалась, и всё равно была записана блокировкой площадкой.

Проверяем четыре звена:
  1. дефолт mark_banned = 'unknown' (обе копии runs-модуля) — test_2686 §4;
  2. финализатор backfill'ов несёт диагноз, когда задача его знает, и 'unknown',
     когда нет или когда блоки разошлись в диагнозах;
  3. browser-режим fetch_detail больше не выдаёт отказ НАШЕГО сайдкара за
     блокировку площадкой (зеркало #2686 для SERP-ветки);
  4. 'unknown' проходит CHECK-констрейнт (миграция 234) — иначе запись упала бы
     на проде, а падение внутри задачи выглядело бы как обычный failed-прогон.
"""

from __future__ import annotations

import os
from collections import Counter
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:5432/test")

from scraper_kit.avito_exceptions import (
    AvitoBlockedError,
    AvitoRateLimitedError,
    AvitoSidecarUnavailableError,
)
from scraper_kit.orchestration.pipeline import ban_kind_of_exception
from scraper_kit.providers.avito.detail import fetch_detail

from app.services import scrape_runs as runs_mod

_BLOCKED_RUN = {"attempted": 5, "enriched": 0, "blocked": 5, "failed": 0}


def _ban_kind_of_finished(**kwargs: Any) -> str:
    """mark_backfill_finished на оборванном блоками прогоне → ban_kind, ушедший в mark_banned."""
    recorded: list[str] = []
    with patch.object(
        runs_mod,
        "mark_banned",
        lambda db, rid, err, c, *, ban_kind: recorded.append(ban_kind),
    ):
        runs_mod.mark_backfill_finished(
            MagicMock(),
            1,
            dict(_BLOCKED_RUN),
            source="avito_detail_backfill",
            aborted_by_blocks=True,
            **kwargs,
        )
    assert len(recorded) == 1, f"ожидался ровно один mark_banned, получено {recorded}"
    return recorded[0]


# ── 1. Финализатор backfill'ов: диагноз или честное «не знаю» ────────────────


def test_finalizer_without_diagnosis_writes_unknown() -> None:
    """Задача типы не различает → 'unknown'.

    Фальсификация: до правки этот вызывающий (единственный источник обеих
    рантайм-строк на проде) не передавал ничего и получал дефолт 'platform' —
    assert падает на старом коде.
    """
    assert _ban_kind_of_finished() == runs_mod.BAN_KIND_UNKNOWN


def test_finalizer_carries_the_diagnosis_when_all_blocks_agree() -> None:
    """Все блоки прогона — отказ нашего сайдкара → 'infra' доезжает до строки прогона."""
    assert _ban_kind_of_finished(ban_kinds={runs_mod.BAN_KIND_INFRA}) == runs_mod.BAN_KIND_INFRA
    assert (
        _ban_kind_of_finished(ban_kinds=[runs_mod.BAN_KIND_PLATFORM, runs_mod.BAN_KIND_PLATFORM])
        == runs_mod.BAN_KIND_PLATFORM
    )


def test_finalizer_with_mixed_diagnoses_writes_unknown() -> None:
    """Блоки разошлись → 'unknown': какой из них оборвал прогон, мы не знаем.

    Выбрать «последний попавшийся» значило бы вернуть назначенный диагноз под
    другим именем.
    """
    assert (
        _ban_kind_of_finished(ban_kinds={runs_mod.BAN_KIND_INFRA, runs_mod.BAN_KIND_PLATFORM})
        == runs_mod.BAN_KIND_UNKNOWN
    )


# ── 1b. Перепись (#3178): кратности решают, set() их терял ───────────────────


def test_finalizer_census_majority_wins_over_minority() -> None:
    """4×platform + 1×infra → 'platform': явное большинство, не 'unknown' (#3178).

    Фальсификация: до правки вызывающий терял кратности через set(ban_kinds) ДО
    решения — {'platform', 'infra'} от 4+1 был неотличим от настоящего 2+2, и оба
    давали 'unknown'. Прод: 5 прогонов подряд с одним и тем же AvitoBlockedError
    (firewall/soft-block, browser-mode) — 4 блока сошлись в 'platform', 1 в 'infra',
    строка прогона получала 'unknown' там, где явное большинство прямо говорило
    'platform'.
    """
    census = Counter({runs_mod.BAN_KIND_PLATFORM: 4, runs_mod.BAN_KIND_INFRA: 1})
    assert _ban_kind_of_finished(ban_kinds=census) == runs_mod.BAN_KIND_PLATFORM


def test_finalizer_census_tie_writes_unknown() -> None:
    """2×platform + 2×infra — ровно поровну, строгого большинства нет → 'unknown'."""
    census = Counter({runs_mod.BAN_KIND_PLATFORM: 2, runs_mod.BAN_KIND_INFRA: 2})
    assert _ban_kind_of_finished(ban_kinds=census) == runs_mod.BAN_KIND_UNKNOWN


def test_finalizer_census_all_same_kind() -> None:
    """5 из 5 одного вида → он же, вне зависимости от абсолютного счёта блоков."""
    census = Counter({runs_mod.BAN_KIND_PLATFORM: 5})
    assert _ban_kind_of_finished(ban_kinds=census) == runs_mod.BAN_KIND_PLATFORM


def test_finalizer_census_lands_in_counters() -> None:
    """Перепись (kind -> count) остаётся в counters['ban_kinds'] после выбора
    диагноза строки прогона — не теряется вместе с решением (#3178)."""
    counters = dict(_BLOCKED_RUN)
    with patch.object(runs_mod, "mark_banned", lambda *a, **k: None):
        runs_mod.mark_backfill_finished(
            MagicMock(),
            1,
            counters,
            source="avito_detail_backfill",
            aborted_by_blocks=True,
            ban_kinds=Counter({runs_mod.BAN_KIND_PLATFORM: 4, runs_mod.BAN_KIND_INFRA: 1}),
        )
    assert counters["ban_kinds"] == {"platform": 4, "infra": 1}


# ── 2. Диагноз не врёт там, где он передаётся: browser-ветка fetch_detail ────


def _browser(*, side_effect: BaseException | None = None, html: str = "") -> Any:
    bf = MagicMock()
    bf.fetch = AsyncMock(side_effect=side_effect, return_value=html)
    return bf


def _sidecar_503() -> httpx.HTTPStatusError:
    """Ровно то, что прод писал 90 раз с 05.07 по 03.08 (см. миграцию 218)."""
    request = httpx.Request("POST", "http://tradein-browser:3000/fetch")
    response = httpx.Response(
        503, json={"error": "browser unavailable (proxy may be down)"}, request=request
    )
    return httpx.HTTPStatusError("503", request=request, response=response)


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "sidecar_failure",
    [_sidecar_503(), httpx.ConnectError("connection refused"), httpx.ReadTimeout("read timeout")],
)
async def test_browser_detail_sidecar_failure_is_infra(sidecar_failure: BaseException) -> None:
    """Сайдкар не отдал страницу → диагноз 'infra', поведение прогона прежнее.

    Фальсификация: до правки ветка поднимала AvitoBlockedError на ЛЮБОЙ отказ
    сайдкара, то есть ban_kind_of_exception возвращал 'platform' — «площадка нас
    заблокировала» на странице, которая до площадки не доехала.
    Второй assert стережёт совместимость: все существующие
    `except (AvitoBlockedError, AvitoRateLimitedError)` обязаны продолжать ловить
    этот отказ, иначе прогон уйдёт в mark_failed и потеряет чекпоинт.
    """
    with pytest.raises(AvitoSidecarUnavailableError) as excinfo:
        await fetch_detail("/items/1", browser_fetcher=_browser(side_effect=sidecar_failure))

    assert isinstance(excinfo.value, AvitoRateLimitedError)
    assert ban_kind_of_exception(excinfo.value) == runs_mod.BAN_KIND_INFRA


@pytest.mark.asyncio
async def test_browser_detail_firewall_page_stays_platform() -> None:
    """Площадка ПОКАЗАЛА firewall → 'platform' остаётся: правка сужает метку, а не снимает."""
    firewall = "<html><body>Доступ ограничен: проблема с IP</body></html>"

    with pytest.raises(AvitoBlockedError) as excinfo:
        await fetch_detail("/items/1", browser_fetcher=_browser(html=firewall))

    assert not isinstance(excinfo.value, AvitoSidecarUnavailableError)
    assert ban_kind_of_exception(excinfo.value) == runs_mod.BAN_KIND_PLATFORM


# ── 3. 'unknown' проходит CHECK-констрейнт (миграция 234) ────────────────────


def _live_session() -> Any | None:
    """Session к живой Postgres, если она есть (в CI есть — ci-tradein.yml, #2745)."""
    try:
        from sqlalchemy import create_engine, text
        from sqlalchemy.orm import sessionmaker

        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
        if not dsn or "localhost:5432/test" in dsn:
            return None
        engine = create_engine(dsn, future=True)
        conn = engine.connect()
        conn.execute(text("SELECT 1"))
        conn.close()
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_default_ban_kind_survives_the_check_constraint() -> None:
    """mark_banned без диагноза пишет 'unknown' в живую схему.

    Фальсификация: без миграции 234 CHECK разрешает только platform/infra — UPDATE
    падает на IntegrityError. На проде это падение было бы неотличимо от обычного
    сбоя задачи (внешний `except Exception` backfill'а увёл бы прогон в
    mark_failed, потеряв done_buckets-чекпоинт).
    """
    from sqlalchemy import text

    db = _live_session()
    assert db is not None
    try:
        run_id = db.execute(
            text(
                "INSERT INTO scrape_runs (source, status) "
                "VALUES ('test_2764', 'running') RETURNING id"
            )
        ).scalar_one()
        db.commit()

        runs_mod.mark_banned(db, int(run_id), "test 2764", {"lots_fetched": 0})

        row = db.execute(
            text("SELECT status, ban_kind FROM scrape_runs WHERE id = :id"), {"id": run_id}
        ).one()
        assert row.status == "banned"
        assert row.ban_kind == runs_mod.BAN_KIND_UNKNOWN
    finally:
        db.execute(text("DELETE FROM scrape_runs WHERE source = 'test_2764'"))
        db.commit()
        db.close()