"""Тёплый переиспользуемый browser-context для Avito detail-бэкфилла (#3180/#3251).

Переносит на Avito рецепт, измеренно поднявший добор карточек у DomClick (#3118):
sidecar's browser.new_page() создаёт НОВЫЙ изолированный context на КАЖДЫЙ /fetch —
пройденный QRATOR-подобный анти-бот проход площадки выбрасывается вместе с ним, и
каждая следующая карточка снова холодная. reuse_context=True держит ОДИН sidecar-
context живым на весь прогон; сброс сожжённого context'а оркестратор просит через
bf.request_context_reset() — как и DomClick, ровно один раз за прогон (#3212), а не
на каждый обнаруженный блок (иначе сброс сам себя поддерживает — см. domclick
docstring). Reset относится ТОЛЬКО к реальному бану площадкой (AvitoBlockedError и
подтипы), не к отказу нашего тракта (AvitoSidecarUnavailableError) и не к мёртвому
листингу (AvitoListingGoneError) — площадка тут ни при чём в обоих случаях, context
цел.

Контракт тестов сформулирован от требования площадки/архитектуры (что ДОЛЖНО
происходить при блоке/не-блоке), а не от текущей реализации — см. #3262, где
обратный подход закрепил баг тестами на месяцы.

Зеркалит tests/test_3118_domclick_warm_context.py (module-level patch-target
строки, helpers).
"""

from __future__ import annotations

import os
import sys
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,
    AvitoContentBlockedError,
    AvitoListingGoneError,
    AvitoSidecarUnavailableError,
)

from app.core import shutdown as _sd  # noqa: E402
from app.tasks.avito_detail_backfill import run_avito_detail_backfill  # noqa: E402

_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"
_SETTINGS = "app.tasks.avito_detail_backfill.settings"
_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"


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


def _fake_settings(**overrides: object) -> MagicMock:
    defaults: dict[str, object] = {
        "scraper_fetch_mode": "browser",
        "avito_detail_backfill_use_curl": False,
        "detail_backfill_block_ratio_window": 20,
        "detail_backfill_block_ratio_threshold": 0.7,
        "browser_http_endpoint": "http://browser:9000",
        # Реальный дефолт (config.py). Без него bare MagicMock отдаёт
        # child-MagicMock на сравнение `attempts_since_rotation >= settings.avito_
        # detail_backfill_rotate_after_attempts` в browser-режиме, и тест падает
        # TypeError: '>=' not supported between 'int' and 'MagicMock'. Та же
        # ловушка, что уже описана здесь для detail_backfill_block_ratio_window.
        "avito_detail_backfill_rotate_after_attempts": 15,
        # #3283g: та же ловушка -- bare MagicMock отдаёт child-MagicMock на
        # `ban_budget > 0` / `attempts_since_rotation < settings.avito_detail_backfill_
        # rotate_on_ban_min_gap` -> TypeError. Значения = реальные дефолты config.py,
        # чтобы существующие 3251-сценарии (<10 попыток в прогоне) не задевали
        # rotate-on-ban веткой -- она молчаливо остаётся в "gap не выдержан".
        "avito_detail_backfill_rotate_on_ban_max": 2,
        "avito_detail_backfill_rotate_on_ban_min_gap": 10,
    }
    defaults.update(overrides)
    return MagicMock(**defaults)


def _make_snapshot(n: int, *, absolute: bool = True) -> list[dict]:
    if absolute:
        return [
            {
                "id": i + 1,
                "source_url": (
                    f"https://www.avito.ru/ekaterinburg/kvartiry/"
                    f"1-k._kvartira_{i + 1}"
                ),
            }
            for i in range(n)
        ]
    return [{"id": i + 1, "source_url": f"/items/{i + 1}"} for i in range(n)]


def _mock_db(snapshot: list[dict]) -> MagicMock:
    db = MagicMock()
    sel = MagicMock()
    sel.mappings.return_value.all.return_value = snapshot
    db.execute.return_value = sel
    return db


def _mock_browser_fetcher_cls() -> tuple[MagicMock, AsyncMock]:
    """Возвращает (класс-мок, instance-мок) -- instance с реальным request_context_reset."""
    instance = AsyncMock()
    instance.__aenter__ = AsyncMock(return_value=instance)
    instance.__aexit__ = AsyncMock(return_value=False)
    # request_context_reset -- синхронный метод (не awaited caller'ом).
    instance.request_context_reset = MagicMock()
    return MagicMock(return_value=instance), instance


# ── BrowserFetcher(reuse_context=True) ──────────────────────────────────────────


@pytest.mark.asyncio
async def test_backfill_constructs_browser_fetcher_with_reuse_context() -> None:
    """BrowserFetcher поднимается с reuse_context=True (#3180) -- тёплый sidecar-context."""
    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(return_value=MagicMock())
    fake_settings = _fake_settings()
    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        await run_avito_detail_backfill(
            db, run_id=200, params={"batch_size": 10, "budget_sec": 3600}
        )

    mock_bf_cls.assert_called_once()
    _, call_kwargs = mock_bf_cls.call_args
    assert call_kwargs["source"] == "avito"
    assert call_kwargs["reuse_context"] is True


# ── reset: ровно один раз за прогон, только на реальный бан площадкой ──────────


@pytest.mark.asyncio
async def test_context_reset_requested_at_most_once_per_run() -> None:
    """3 подряд AvitoBlockedError -> request_context_reset() запрошен ровно 1 раз.

    Зеркалит domclick #3212: сброс на КАЖДЫЙ блок сам себя поддерживает (пройденный
    анти-бот проход живёт в context'е, сброс его выбрасывает -> следующая проверка с
    того же IP снова блокируется) -- одна осечка становится необратимым каскадом.
    """
    snapshot = _make_snapshot(10)
    db = _mock_db(snapshot)
    runs = MagicMock()
    blocked_exc = AvitoBlockedError("firewall/soft-block")
    mock_fetch = AsyncMock(side_effect=blocked_exc)
    fake_settings = _fake_settings()
    mock_bf_cls, instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=201,
            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 3},
        )

    assert result.blocked == 3
    assert instance.request_context_reset.call_count == 1, (
        "ожидалась ровно одна попытка начать с чистого листа за весь прогон"
    )


@pytest.mark.asyncio
async def test_content_blocked_subtype_also_requests_reset() -> None:
    """AvitoContentBlockedError (подтип AvitoBlockedError) тоже считается блоком площадки."""
    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(side_effect=AvitoContentBlockedError("0 карточек, captcha"))
    fake_settings = _fake_settings()
    mock_bf_cls, instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=202,
            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
        )

    assert result.blocked == 3
    instance.request_context_reset.assert_called_once()


@pytest.mark.asyncio
async def test_sidecar_unavailable_does_not_request_context_reset() -> None:
    """AvitoSidecarUnavailableError -- НАШ тракт, площадка ни при чём: reset НЕ звать.

    Context цел, сбрасывать его на отказе сайдкара (503/timeout ДО площадки) незачем
    и вредно -- анти-бот проход площадки внутри context'а тут ни при чём.
    """
    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(side_effect=AvitoSidecarUnavailableError("browser unavailable"))
    fake_settings = _fake_settings()
    mock_bf_cls, instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=203,
            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
        )

    assert result.blocked == 3
    instance.request_context_reset.assert_not_called()


@pytest.mark.asyncio
async def test_listing_gone_does_not_request_context_reset() -> None:
    """AvitoListingGoneError (мёртвый 404-листинг) -- не блок, reset НЕ звать."""
    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(side_effect=AvitoListingGoneError("404 gone"))
    fake_settings = _fake_settings()
    mock_bf_cls, instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=204, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.gone == 3
    assert result.blocked == 0
    instance.request_context_reset.assert_not_called()


@pytest.mark.asyncio
async def test_successful_fetch_does_not_request_context_reset() -> None:
    """Успешный прогон без блоков -> request_context_reset() ни разу не вызывается."""
    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(return_value=MagicMock())
    fake_settings = _fake_settings()
    mock_bf_cls, instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=205, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.blocked == 0
    instance.request_context_reset.assert_not_called()


# ── origin/browser_referer wiring в fetch_detail ────────────────────────────────


@pytest.mark.asyncio
async def test_backfill_passes_serp_origin_to_fetch_detail_in_browser_mode() -> None:
    """browser-режим: fetch_detail получает origin=browser_referer=SERP-URL карточки."""
    snapshot = _make_snapshot(1, absolute=True)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(return_value=MagicMock())
    fake_settings = _fake_settings()
    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        await run_avito_detail_backfill(
            db, run_id=206, params={"batch_size": 10, "budget_sec": 3600}
        )

    mock_fetch.assert_called_once()
    _, kwargs = mock_fetch.call_args
    expected = "https://www.avito.ru/ekaterinburg/kvartiry/prodam"
    assert kwargs.get("origin") == expected
    assert kwargs.get("browser_referer") == expected


@pytest.mark.asyncio
async def test_backfill_omits_origin_when_url_unparseable() -> None:
    """source_url без города/категории (relative path) -> origin/browser_referer=None.

    Хелпер не смог вывести SERP-URL -> caller НЕ выдумывает его, просто не передаёт
    (fetch_detail тогда идёт как до #3251).
    """
    snapshot = _make_snapshot(1, absolute=False)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(return_value=MagicMock())
    fake_settings = _fake_settings()
    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        await run_avito_detail_backfill(
            db, run_id=207, params={"batch_size": 10, "budget_sec": 3600}
        )

    mock_fetch.assert_called_once()
    _, kwargs = mock_fetch.call_args
    assert kwargs.get("origin") is None
    assert kwargs.get("browser_referer") is None


@pytest.mark.asyncio
async def test_curl_mode_does_not_pass_origin() -> None:
    """use_curl=True (прод-дефолт warm-batch путь) -- origin/browser_referer НЕ передаются.

    Тот путь уже прогрет warm-batch'ем (referer=), origin -- ТОЛЬКО browser-режим (см.
    fetch_detail docstring): в curl-режиме передача origin стоила бы лишней полной
    навигации без реального смысла (browser_fetcher там вообще не участвует).
    """
    snapshot = _make_snapshot(1, absolute=True)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(return_value=MagicMock())
    fake_settings = _fake_settings(avito_detail_backfill_use_curl=True)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER) as mock_bf_cls,
        patch(_FETCH, mock_fetch),
        patch(
            "app.tasks.avito_detail_backfill.build_warmed_session",
            AsyncMock(return_value=AsyncMock()),
        ),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        await run_avito_detail_backfill(
            db, run_id=208, params={"batch_size": 10, "budget_sec": 3600}
        )

    mock_bf_cls.assert_not_called()
    mock_fetch.assert_called_once()
    _, kwargs = mock_fetch.call_args
    assert kwargs.get("origin") is None
    assert kwargs.get("browser_referer") is None