from __future__ import annotations

import asyncio
import fnmatch
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 app.core import shutdown as _sd  # noqa: E402
from app.tasks.avito_detail_backfill import (  # noqa: E402
    _OBLAST_AVITO_URL_PATTERNS,
    AvitoDetailBackfillResult,
    run_avito_detail_backfill,
)


@pytest.fixture(autouse=True)
def _reset_shutdown() -> None:
    """shutdown — module-global Event: чистим вокруг каждого теста (изоляция #1182)."""
    _sd.reset_shutdown()
    yield
    _sd.reset_shutdown()


@pytest.fixture(autouse=True)
def _patch_build_warmed() -> object:
    """use_curl-ветка (#1551) строит прогретую сессию через build_warmed_session и
    освежает куки через research_in_session (реальный yandex/avito GET + sleep).
    fetch_detail в этих тестах всё равно замокан — мокаем обе не-сетевыми AsyncMock,
    чтобы setup / periodic re-warm / on-block cooldown не ходили в сеть."""
    with (
        patch(
            "app.tasks.avito_detail_backfill.build_warmed_session",
            AsyncMock(return_value=AsyncMock()),
        ),
        patch(
            "app.tasks.avito_detail_backfill.research_in_session",
            AsyncMock(return_value=True),
        ) as research,
    ):
        yield research


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_snapshot(n: int) -> list[dict]:
    return [{"id": i + 1, "source_url": f"/items/{i + 1}"} for i in range(n)]


def _mock_db(snapshot: list[dict]) -> MagicMock:
    """Fake Session: first execute() returns snapshot via .mappings().all()."""
    db = MagicMock()
    sel = MagicMock()
    sel.mappings.return_value.all.return_value = snapshot
    db.execute.return_value = sel
    return db


_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
_SETTINGS = "app.tasks.avito_detail_backfill.settings"
_SHUTDOWN = "app.tasks.avito_detail_backfill.shutdown_requested"
_BUILD_WARM = "app.tasks.avito_detail_backfill.build_warmed_session"
_RESEARCH = "app.tasks.avito_detail_backfill.research_in_session"
# #2825: settings.scraper_proxy_url в "elif not use_curl" (legacy curl_cffi) branch
# заменён на resolve_proxy_url(db, "avito") (пул scrape_proxies с учётом банов,
# fallback на settings.scraper_proxy_url внутри app.services.proxy_egress) -- эти
# тесты про block/ban/rotate-логику, не про подбор прокси (см.
# tests/services/test_proxy_egress.py), поэтому мокаем сам резолвер.
_RESOLVE_PROXY_URL = "app.tasks.avito_detail_backfill.resolve_proxy_url"

# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_backfill_empty_snapshot_marks_done() -> None:
    """Empty snapshot -> mark_done immediately, no fetch calls."""
    db = _mock_db([])
    runs = MagicMock()
    fake_session = AsyncMock()
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=fake_session),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH) as mock_fetch,
    ):
        result = await run_avito_detail_backfill(
            db, run_id=1, params={"batch_size": 10, "budget_sec": 60}
        )

    assert isinstance(result, AvitoDetailBackfillResult)
    assert result.attempted == 0
    assert result.enriched == 0
    mock_fetch.assert_not_called()
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()


@pytest.mark.asyncio
async def test_backfill_processes_snapshot_to_completion() -> None:
    """3 listings -> all fetched and enriched, mark_done called."""
    from app.services.scraper_adapters import RealScraperConfig

    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER) as mock_scraper_cls,
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=2, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.attempted == 3
    assert result.enriched == 3
    assert result.blocked == 0
    assert result.failed == 0
    assert mock_fetch.call_count == 3
    runs.mark_backfill_finished.assert_called_once()
    runs.mark_failed.assert_not_called()
    # #2310 regression guard: kit fetch_detail silently drops the backconnect-
    # on-403 retry (and kit AvitoScraper can't read scraper_proxy_url at all)
    # unless config=RealScraperConfig() is passed/injected at the call
    # site — assert_called()/call_count alone wouldn't catch someone dropping
    # that kwarg later (mirrors #2306's test_backfill_wave2.py:282-286 pattern).
    _, fetch_call_kwargs = mock_fetch.call_args
    assert isinstance(fetch_call_kwargs.get("config"), RealScraperConfig)
    assert isinstance(mock_scraper_cls.call_args.args[0], RealScraperConfig)


@pytest.mark.asyncio
async def test_backfill_build_warmed_session_receives_config() -> None:
    """#2397 Part D1 / #2330 regression guard: kit's build_warmed_session() now accepts
    config=ScraperConfig and silently drops settings.scraper_proxy_url (sticky МГТС-прокси)
    -- NO crash, just a proxy-less session -- unless config=RealScraperConfig() is passed
    at the call site. use_curl=True is the prod-default path (avito_detail_backfill_use_curl),
    so this is the setup-time build_warmed_session() call (~line 177). assert_awaited_once()
    alone wouldn't catch someone dropping the kwarg later (mirrors the fetch_detail config=
    guard above, lines 141-148)."""
    from app.services.scraper_adapters import RealScraperConfig

    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    fake_settings = MagicMock(
        scraper_fetch_mode="cffi",
        avito_detail_backfill_use_curl=True,
    )
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
        patch(_BUILD_WARM, AsyncMock(return_value=AsyncMock())) as mock_build,
    ):
        result = await run_avito_detail_backfill(
            db, run_id=15, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.enriched == 1
    mock_build.assert_awaited_once()
    _, build_call_kwargs = mock_build.call_args
    assert isinstance(build_call_kwargs.get("config"), RealScraperConfig)
    runs.mark_backfill_finished.assert_called_once()
    runs.mark_failed.assert_not_called()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("exc_factory", "expected_kinds"),
    [
        (lambda: AvitoBlockedError("ip blocked"), {"platform"}),
        (lambda: AvitoSidecarUnavailableError("browser unavailable"), {"infra"}),
    ],
)
async def test_backfill_reports_ban_kind_of_the_blocks_it_saw(
    exc_factory: Any, expected_kinds: set[str]
) -> None:
    """Диагноз блоков доезжает до финализатора по ТИПУ исключения (#2764).

    Фальсификация: до правки задача не передавала ничего, и обе серии — отказ
    площадки и отказ нашего сайдкара — давали в scrape_runs.ban_kind одинаковое
    'platform' по умолчанию (прод, прогон 3306: blocked=5, причина не установлена).
    """
    snapshot = _make_snapshot(10)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_scraper = MagicMock()
    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER, mock_scraper),
        patch(_RUNS, runs),
        patch(_RESOLVE_PROXY_URL, MagicMock(return_value="http://test-proxy.local:8080")),
        patch(_FETCH, AsyncMock(side_effect=exc_factory())),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        await run_avito_detail_backfill(
            db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
        )

    runs.mark_backfill_finished.assert_called_once()
    assert set(runs.mark_backfill_finished.call_args.kwargs["ban_kinds"]) == expected_kinds


@pytest.mark.asyncio
async def test_backfill_blocked_abort_after_max_consecutive() -> None:
    """5 consecutive AvitoBlockedError -> abort с пометкой aborted_by_blocks (#2674).

    Раньше — mark_done; на проде 13 прогонов attempted=5 blocked=5 enriched=0
    назывались успехом. Теперь флаг обрыва → статус 'banned'.

    #1950 abort-reorder: abort-check ПЕРЕД recovery → на 5-м (аборт-)блоке rotate_ip
    НЕ дёргается (не тратим recovery на финальном блоке). rotate_ip x4 (блоки 1-4).
    """
    from scraper_kit.avito_exceptions import AvitoBlockedError

    snapshot = _make_snapshot(10)
    db = _mock_db(snapshot)
    runs = MagicMock()
    blocked_exc = AvitoBlockedError("ip blocked")
    mock_fetch = AsyncMock(side_effect=blocked_exc)
    mock_scraper = MagicMock()
    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
    # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
    # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER, mock_scraper),
        patch(_RUNS, runs),
        patch(_RESOLVE_PROXY_URL, MagicMock(return_value="http://test-proxy.local:8080")),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
        )

    assert result.blocked == 5
    assert result.attempted == 5
    assert result.enriched == 0
    runs.mark_backfill_finished.assert_called_once()
    assert runs.mark_backfill_finished.call_args.kwargs["aborted_by_blocks"] is True
    runs.mark_failed.assert_not_called()
    # abort-check до recovery → 5-й блок абортит без rotate; rotate только на блоках 1-4.
    assert mock_scraper.return_value._rotate_ip.call_count == 4


@pytest.mark.asyncio
async def test_backfill_sigterm_drain_breaks_and_marks_done_partial() -> None:
    """#1182 Phase 2: shutdown_requested() True → loop выходит на границе карточки,
    mark_done вызывается с ЧАСТИЧНЫМИ счётчиками (не mark_failed, не mark_cancelled).

    side_effect [False, True]: 1-я карточка обрабатывается (attempted=1, enriched=1),
    перед 2-й приходит SIGTERM-drain → break. Snapshot — pending-query, остаток
    до-резюмит следующий run (resume_cursor не нужен).
    """
    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
        patch(_SHUTDOWN, side_effect=[False, True]),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=13, params={"batch_size": 10, "budget_sec": 3600}
        )

    # Обработана только 1-я карточка, на 2-й — drain-break.
    assert result.attempted == 1
    assert result.enriched == 1
    assert mock_fetch.call_count == 1
    runs.mark_backfill_finished.assert_called_once()
    runs.mark_failed.assert_not_called()
    runs.mark_cancelled.assert_not_called()
    # mark_done получил ЧАСТИЧНЫЕ счётчики (attempted=1, а не весь snapshot=3).
    done_counters = runs.mark_backfill_finished.call_args.args[2]
    assert done_counters["attempted"] == 1


@pytest.mark.asyncio
async def test_backfill_budget_guard_stops_loop() -> None:
    """Budget expired before first listing -> fetch_detail not called."""
    snapshot = _make_snapshot(5)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mono_values = iter([0.0, 999.0, 999.0])
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch("app.tasks.avito_detail_backfill.time.monotonic", side_effect=mono_values),
        patch(_FETCH) as mock_fetch,
    ):
        await run_avito_detail_backfill(db, run_id=4, params={"batch_size": 5, "budget_sec": 1})

    mock_fetch.assert_not_called()
    runs.mark_backfill_finished.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_top_level_exception_marks_failed() -> None:
    """db.execute raises -> mark_failed called, exception re-raised."""
    db = MagicMock()
    db.execute.side_effect = RuntimeError("DB connection lost")
    runs = MagicMock()
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
    ):
        with pytest.raises(RuntimeError, match="DB connection lost"):
            await run_avito_detail_backfill(
                db, run_id=5, params={"batch_size": 5, "budget_sec": 60}
            )

    runs.mark_failed.assert_called_once()
    runs.mark_backfill_finished.assert_not_called()


@pytest.mark.asyncio
async def test_backfill_rotate_ip_called_on_each_block() -> None:
    """1 block + 1 success -> rotate_ip called once, enriched=1."""
    from scraper_kit.avito_exceptions import AvitoBlockedError

    snapshot = _make_snapshot(2)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    blocked_exc = AvitoBlockedError("x")
    mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment])
    mock_scraper = MagicMock()
    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
    # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
    # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER, mock_scraper),
        patch(_RUNS, runs),
        patch(_RESOLVE_PROXY_URL, MagicMock(return_value="http://test-proxy.local:8080")),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=6, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
        )

    assert result.enriched == 1
    assert result.blocked == 1
    assert mock_scraper.return_value._rotate_ip.call_count == 1
    runs.mark_backfill_finished.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_snapshot_filters_ekb_active_only() -> None:
    """Снапшот-SELECT (#1814, расширено #2576) фильтрует активные ЕКБ- И
    известные oblast-листинги (region 66), НЕ всё подряд.

    Проверяем, что текст запроса содержит `is_active = TRUE`, `LIKE
    '%/ekaterinburg/%'` (ekb CTE, LIMIT batch_size НЕ сокращён) и `LIKE ANY(...)`
    по oblast-паттернам (oblast CTE, отдельный LIMIT oblast_batch_size) — legacy
    не-ЕКБ/не-область (moskva/spb/tyumen) и мёртвые листинги не попадают в фетч,
    иначе browser спотыкается → curl-бан 429.
    """
    db = _mock_db([])
    runs = MagicMock()
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH),
    ):
        await run_avito_detail_backfill(db, run_id=8, params={"batch_size": 10, "budget_sec": 60})

    # Первый (и единственный при пустом снапшоте) execute — это SELECT-снапшот.
    snapshot_call = db.execute.call_args_list[0]
    sql_text = str(snapshot_call.args[0])
    assert "is_active = TRUE" in sql_text
    assert "/ekaterinburg/" in sql_text
    assert "LIKE ANY(CAST(:oblast_patterns AS text[]))" in sql_text
    assert "detail_enriched_at IS NULL" in sql_text
    assert "(lat IS NULL) DESC" in sql_text
    assert "CAST(:batch_size AS int)" in sql_text
    assert "CAST(:oblast_batch_size AS int)" in sql_text

    # ekb-квота передаётся batch_size БЕЗ урезания (#2576 требование "ЕКБ не
    # деградирует") — oblast получает отдельный (не вычтенный) bind-параметр.
    bind_params = snapshot_call.args[1]
    assert bind_params["batch_size"] == 10
    assert bind_params["oblast_batch_size"] == 100  # default
    assert set(bind_params["oblast_patterns"]) == set(_OBLAST_AVITO_URL_PATTERNS)


def test_oblast_avito_url_patterns_cover_region66_cities() -> None:
    """#2576: _OBLAST_AVITO_URL_PATTERNS строится из CITY_LOCATIONS.avito_slug —
    список должен покрывать реальные Avito-слаги oblast-городов (в т.ч. те, что
    ОТЛИЧАЮТСЯ от нашего city_slug: kamensk-uralskiy через дефис, а не
    kamensk_uralskiy).

    #2578 review: '_' в слаге -- LIKE wildcard, экранируем при построении паттерна
    ('_' -> '\\_') -- nizhniy_tagil/verhnyaya_pyshma здесь ожидаются С обратным
    слэшем перед '_', НЕ голым подчёркиванием."""
    assert "%/kamensk-uralskiy/%" in _OBLAST_AVITO_URL_PATTERNS
    assert "%/nizhniy\\_tagil/%" in _OBLAST_AVITO_URL_PATTERNS
    assert "%/pervouralsk/%" in _OBLAST_AVITO_URL_PATTERNS
    assert "%/verhnyaya\\_pyshma/%" in _OBLAST_AVITO_URL_PATTERNS
    assert "%/serov/%" in _OBLAST_AVITO_URL_PATTERNS
    # ЕКБ обрабатывается отдельным жёстко закодированным паттерном (ekb CTE),
    # НЕ через этот oblast-список — не должен в него затесаться.
    assert not any("ekaterinburg" in p for p in _OBLAST_AVITO_URL_PATTERNS)


def _like_pattern_to_fnmatch(pattern: str) -> str:
    """Точный перевод семантики Postgres `LIKE` (default `ESCAPE '\\'`) в fnmatch-
    паттерн -- посимвольно, а НЕ наивным `.replace()`.

    LIKE: `%` = любая последовательность символов, `_` = РОВНО один любой символ,
    `\\%`/`\\_`/`\\\\` = литералы (экранирование). fnmatch: `*` = любая
    последовательность, `?` = один любой символ; голые `_`/`%` в fnmatch не
    специальны (можно вставлять как литерал без экранирования).

    #2578 review: наивный `pat.replace("%", "*")` (как было раньше) НЕ отражал бы
    семантику `_` вообще -- fnmatch трактует `_` как литерал, LIKE -- как wildcard.
    Из-за этого расхождения прежний тест не поймал бы латентный баг (нет
    экранирования `_` в продовых паттернах). Посимвольный разбор здесь корректно
    различает голый `_` (-> `?` wildcard) и экранированный `\\_` (-> литерал `_`).
    """
    out: list[str] = []
    i = 0
    n = len(pattern)
    while i < n:
        ch = pattern[i]
        if ch == "\\" and i + 1 < n and pattern[i + 1] in ("%", "_", "\\"):
            out.append(pattern[i + 1])  # экранированный символ -> литерал as-is
            i += 2
            continue
        if ch == "%":
            out.append("*")
        elif ch == "_":
            out.append("?")
        else:
            out.append(ch)
        i += 1
    return "".join(out)


def _in_oblast_or_ekb_scope(source_url: str) -> bool:
    """Локальная реплика WHERE-условия snapshot-запроса (ekb CTE OR oblast CTE)
    через корректную LIKE-эмуляцию -- без поднятия БД."""
    if fnmatch.fnmatchcase(source_url, _like_pattern_to_fnmatch("%/ekaterinburg/%")):
        return True
    return any(
        fnmatch.fnmatchcase(source_url, _like_pattern_to_fnmatch(pat))
        for pat in _OBLAST_AVITO_URL_PATTERNS
    )


def test_oblast_avito_url_patterns_include_oblast_and_ekb_exclude_foreign_region() -> None:
    """#2576 DoD: листинг города области и екатеринбургский листинг проходят
    scope-фильтр; листинг чужого региона (Москва/СПб) — нет.

    Использует корректную LIKE-эмуляцию (_like_pattern_to_fnmatch), а не наивный
    `%` -> `*` replace (#2578 review — тот не различал бы `_`-семантику)."""
    # Область (Каменск-Уральский, #2576 — реальный кейс из тикета) -- проходит.
    assert _in_oblast_or_ekb_scope("https://www.avito.ru/kamensk-uralskiy/kvartiry/prodam_123")
    # ЕКБ — по-прежнему проходит (не деградировал).
    assert _in_oblast_or_ekb_scope("https://www.avito.ru/ekaterinburg/kvartiry/prodam_456")
    # Чужой регион — НЕ проходит (иначе поехали бы Москва/СПб/Тюмень legacy-строки).
    assert not _in_oblast_or_ekb_scope("https://www.avito.ru/moskva/kvartiry/prodam_789")
    assert not _in_oblast_or_ekb_scope("https://www.avito.ru/sankt-peterburg/kvartiry/prodam_000")


def test_like_underscore_wildcard_regression_caught_by_escaped_patterns() -> None:
    """#2578 deep-review latent bug: Postgres `LIKE` трактует `_` как wildcard
    "ровно один любой символ", а НЕ литерал. Два слага из пяти (nizhniy_tagil,
    verhnyaya_pyshma) содержат `_` -- БЕЗ экранирования 'nizhniy_tagil' молча
    совпал бы с 'nizhniyXtagil' (X = любой символ), т.е. коллизия слагов при
    появлении похожего города. Сегодня коллизий нет (проверено на проде: raw vs
    escaped паттерны дают одинаковые 776 совпадений), но дыра латентная.

    Этот тест ДОЛЖЕН падать на RAW (неэкранированном) варианте паттерна -- именно
    так выглядели продовые паттерны ДО фикса #2578 (`%/nizhniy_tagil/%`, без
    `\\`). Экранированный прод-паттерн (_OBLAST_AVITO_URL_PATTERNS, ПОСЛЕ фикса)
    коллизию отклоняет, точный слаг по-прежнему матчит (позитивный кейс жив).
    """
    raw_pattern = "%/nizhniy_tagil/%"  # как было бы БЕЗ фикса #2578 (голый '_')
    escaped_pattern = next(p for p in _OBLAST_AVITO_URL_PATTERNS if "nizhniy" in p)
    # Сам факт экранирования: прод-паттерн ДОЛЖЕН отличаться от raw ('_' -> '\_').
    assert escaped_pattern != raw_pattern, "фикс #2578 должен экранировать '_' в avito_slug"

    collision_url = "https://www.avito.ru/nizhniyXtagil/kvartiry/prodam_1"
    exact_url = "https://www.avito.ru/nizhniy_tagil/kvartiry/prodam_1"

    # RAW: '_' -- wildcard -> ложное совпадение с ЛЮБЫМ символом на его месте.
    assert fnmatch.fnmatchcase(collision_url, _like_pattern_to_fnmatch(raw_pattern))
    # Экранированный прод-паттерн (после фикса): '_' -- литерал -> коллизия отклонена.
    assert not fnmatch.fnmatchcase(collision_url, _like_pattern_to_fnmatch(escaped_pattern))
    # Позитивный кейс не сломан: точный слаг матчит ОБА варианта паттерна.
    assert fnmatch.fnmatchcase(exact_url, _like_pattern_to_fnmatch(raw_pattern))
    assert fnmatch.fnmatchcase(exact_url, _like_pattern_to_fnmatch(escaped_pattern))


@pytest.mark.asyncio
async def test_backfill_fetch_exception_continues() -> None:
    """RuntimeError on one listing -> failed++, db.rollback(), loop continues for next."""
    snapshot = _make_snapshot(2)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(side_effect=[RuntimeError("parse error"), mock_enrichment])
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=7, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.failed == 1
    assert result.enriched == 1
    assert result.attempted == 2
    db.rollback.assert_called()
    runs.mark_backfill_finished.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_fetch_timeout_skips_and_continues() -> None:
    """#1950: один fetch_detail зависает дольше hard-timeout → этот листинг failed,
    loop НЕ зависает, переходит к следующему, run завершается mark_done (не zombie).

    Регрессия run 423 (завис 7.7ч → reaped): fetch_detail без timeout блокировал
    loop навсегда. asyncio.wait_for(timeout) отменяет зависший fetch → TimeoutError.
    """
    snapshot = _make_snapshot(2)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()

    call_urls: list[str] = []

    async def _fetch(
        url: str,
        *,
        cffi_session: object = None,
        browser_fetcher: object = None,
        referer: object = None,
        reconnect_on_block: bool = True,
        config: object = None,
    ):
        call_urls.append(url)
        if len(call_urls) == 1:
            await asyncio.Event().wait()  # висит вечно → wait_for отменит по timeout
        return mock_enrichment

    # Короткий timeout (50ms) — тест не ждёт реальные 90s; rotate-settle тоже 0.
    fake_settings = MagicMock(
        scraper_fetch_mode="cffi",
        avito_detail_fetch_timeout_s=0.05,
        avito_proxy_rotate_settle_s=0.0,
    )
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, _fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=11, params={"batch_size": 10, "budget_sec": 3600}
        )

    # Первый листинг — failed (timeout), второй — enriched. Run завершён mark_done.
    assert result.failed == 1
    assert result.enriched == 1
    assert result.attempted == 2
    assert len(call_urls) == 2, "loop должен дойти до второго листинга, а не зависнуть"
    db.rollback.assert_called()
    runs.mark_backfill_finished.assert_called_once()
    runs.mark_failed.assert_not_called()


@pytest.mark.asyncio
async def test_backfill_listing_gone_marks_inactive_no_breaker() -> None:
    """#2034: AvitoListingGoneError (мёртвый 404-листинг) → counters.gone++, blocked
    НЕ растёт, consecutive-block breaker НЕ абортит, листинг помечается is_active=FALSE.

    Регрессия run 458 (attempted=5 enriched=0 blocked=5 → abort): lat-null очередь
    состоит из dead-листингов; раньше 404 ловился как soft-block → breaker абортил run
    до live-листингов. Теперь 404 нейтрален к breaker'у и метит листинг inactive.
    """
    from scraper_kit.avito_exceptions import AvitoListingGoneError

    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(side_effect=AvitoListingGoneError("404 gone"))
    mock_scraper = MagicMock()
    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
    # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
    # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER, mock_scraper),
        patch(_RUNS, runs),
        patch(_RESOLVE_PROXY_URL, MagicMock(return_value="http://test-proxy.local:8080")),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=12,
            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
        )

    assert result.gone == 1
    assert result.blocked == 0
    assert result.attempted == 1
    assert result.enriched == 0
    # breaker НЕ абортил: rotate_ip НЕ дёргался (gone ≠ block), run завершён mark_done.
    mock_scraper.return_value._rotate_ip.assert_not_called()
    runs.mark_backfill_finished.assert_called_once()
    runs.mark_failed.assert_not_called()
    # UPDATE listings SET is_active = FALSE по row id=1 выполнен (мок db.execute).
    update_calls = [
        c
        for c in db.execute.call_args_list
        if "UPDATE listings SET is_active = FALSE" in str(c.args[0])
    ]
    assert len(update_calls) == 1
    assert update_calls[0].args[1] == {"id": 1}


_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"


@pytest.mark.asyncio
async def test_backfill_use_curl_flag_skips_browser_fetcher() -> None:
    """avito_detail_backfill_use_curl=True: BrowserFetcher не создаётся,
    fetch_detail вызывается с browser_fetcher=None (curl/backconnect путь).
    """
    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    # Флаг use_curl=True, fetch_mode=browser (но флаг перекрывает)
    fake_settings = MagicMock(
        scraper_fetch_mode="browser",
        avito_detail_backfill_use_curl=True,
    )
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_BROWSER_FETCHER) as mock_bf_cls,
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=9, params={"batch_size": 10, "budget_sec": 3600}
        )

    # BrowserFetcher не должен быть создан
    mock_bf_cls.assert_not_called()
    # fetch_detail вызван с browser_fetcher=None (curl путь)
    assert mock_fetch.call_count == 1
    _, kwargs = mock_fetch.call_args
    assert kwargs.get("browser_fetcher") is None
    assert result.enriched == 1
    runs.mark_backfill_finished.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_use_curl_false_creates_browser_fetcher() -> None:
    """avito_detail_backfill_use_curl=False + scraper_fetch_mode='browser':
    BrowserFetcher создаётся (legacy browser/auv путь).
    """
    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    fake_settings = MagicMock(
        scraper_fetch_mode="browser",
        avito_detail_backfill_use_curl=False,
    )
    mock_bf_instance = AsyncMock()
    mock_bf_instance.__aenter__ = AsyncMock(return_value=mock_bf_instance)
    mock_bf_instance.__aexit__ = AsyncMock(return_value=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_BROWSER_FETCHER, return_value=mock_bf_instance) as mock_bf_cls,
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=10, params={"batch_size": 10, "budget_sec": 3600}
        )

    # BrowserFetcher должен быть создан (source="avito", endpoint из settings — #2310
    # kit BrowserFetcher требует endpoint= обязательным keyword-only параметром)
    mock_bf_cls.assert_called_once_with(
        source="avito", endpoint=fake_settings.browser_http_endpoint
    )
    # fetch_detail вызван с browser_fetcher установленным (не None)
    assert mock_fetch.call_count == 1
    _, kwargs = mock_fetch.call_args
    assert kwargs.get("browser_fetcher") is not None
    assert result.enriched == 1
    runs.mark_backfill_finished.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_use_curl_block_cooldown_research_no_rebuild() -> None:
    """#1551 sticky-IP: on-block (use_curl) даёт cooldown + in-session research, НЕ
    пересоздаёт прогретую сессию и НЕ дёргает changeip-ротацию.

    МГТС sticky — один фикс. exit-IP (rebuild != новый IP), уйти на свежий IP софтом
    нельзя. Блок = rate-limit текущего IP → cooldown + re-search той же сессией.
    fetch: 1-й вызов BLOCKED, 2-й — успех.
    """
    from scraper_kit.avito_exceptions import AvitoBlockedError

    snapshot = _make_snapshot(2)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    blocked_exc = AvitoBlockedError("rate-limited")
    mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment])
    warmed_session = AsyncMock()
    fake_settings = MagicMock(
        scraper_fetch_mode="cffi",
        avito_detail_backfill_use_curl=True,
    )
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SCRAPER) as mock_scraper,
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
        patch(_BUILD_WARM, AsyncMock(return_value=warmed_session)) as mock_build,
        patch(_RESEARCH, new_callable=AsyncMock) as mock_research,
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=14,
            params={
                "batch_size": 10,
                "budget_sec": 3600,
                "max_consecutive_blocks": 5,
                "block_cooldown_sec": 0.0,
            },
        )

    assert result.blocked == 1
    assert result.enriched == 1
    assert result.attempted == 2
    # on-block: research_in_session вызван (освежить куки in-session)...
    mock_research.assert_awaited_once()
    # ...а build_warmed_session НЕ дёргался повторно (вызван только 1× в setup —
    # прогретая сессия НЕ пересоздаётся на блоке, sticky-IP rebuild бесполезен).
    assert mock_build.await_count == 1
    # changeip-ротация (legacy путь) под use_curl НЕ дёргается.
    mock_scraper.return_value._rotate_ip.assert_not_called()
    runs.mark_backfill_finished.assert_called_once()
    runs.mark_failed.assert_not_called()


# ---------------------------------------------------------------------------
# Отказы-не-блоки: брейкер + перепись причин (2026-08-06)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_backfill_aborts_on_consecutive_failures_and_names_the_reason() -> None:
    """Серия отказов без единого успеха обрывается, а причина попадает в статус прогона.

    Прод 3-5 августа: attempted≈1600, failed≈1600, blocked=0, enriched=0, весь
    бюджет 9000 с и 1600 запросов через единственный прокси — и ни слова о том,
    ЧТО именно отказало (поштучные отказы логируются WARNING, а логи контейнера
    пропадают на первом деплое). Брейкера на отказы-не-блоки не было вовсе.
    """
    snapshot = _make_snapshot(200)
    db = _mock_db(snapshot)
    runs = MagicMock()
    # Тот же класс отказа, что видели у соседнего свипа в тот же день.
    mock_fetch = AsyncMock(
        side_effect=OSError(
            "Failed to perform, curl: (56) CONNECT tunnel failed, response 502. "
            "See https://curl.se/libcurl/c/libcurl-errors.html"
        )
    )
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=True)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=77,
            params={
                "batch_size": 200,
                "budget_sec": 3600,
                "max_consecutive_failures": 25,
            },
        )

    assert result.attempted == 25, "серия отказов обязана обрываться, а не выедать бюджет"
    assert result.failed == 25
    assert result.blocked == 0

    hint = runs.mark_backfill_finished.call_args.kwargs["fail_hint"]
    assert hint is not None
    assert "OSError" in hint  # тип исключения = кому принадлежит отказ
    assert "CONNECT tunnel failed" in hint
    assert "25 из 25" in hint  # доля, а не единичный пример
    assert "https://curl.se" not in hint  # URL вырезан, иначе 1600 «разных» причин


@pytest.mark.asyncio
async def test_backfill_success_resets_failure_streak() -> None:
    """Успех между отказами обнуляет серию — здоровый прогон брейкер не трогает."""
    snapshot = _make_snapshot(5)
    db = _mock_db(snapshot)
    runs = MagicMock()
    boom = ValueError("avito detail HTTP 500 for https://www.avito.ru/x")
    # 2 отказа, успех, 2 отказа — при пороге 3 ни одна серия его не достигает.
    mock_fetch = AsyncMock(side_effect=[boom, boom, MagicMock(), boom, boom])
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=True)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=78,
            params={"batch_size": 5, "budget_sec": 3600, "max_consecutive_failures": 3},
        )

    assert result.attempted == 5
    assert result.enriched == 1
    assert result.failed == 4