"""Тождество счётчиков у соседей domclick: yandex/avito detail-backfill (#3338).

Тот же дефект, что #3332 (см. tests/test_3332_domclick_counter_identity.py):
`counters.attempted` инкрементируется ДО попытки, исход дописывается при разборе
результата — а ветка `save_detail_enrichment(...) -> False` исхода не дописывала
вовсе (`if save(...): enriched += 1` без else). Страницу взяли, разобрали, а
UPDATE не задел ни одной строки (объявление удалено/деактивировано между
снимком и записью) — попытка была, исхода не было. Расхождение
`attempted - сумма исходов` читается как потерянный отказ площадки.

Разбор ВСЕХ точек выхода из цикла попыток показал, что это единственная дыра у
обоих (у domclick второй был обрыв «пул прокси пуст» — у соседей такого пути
нет: `resolve_proxy_url` бросает ProxyPoolExhaustedError ДО цикла):

  yandex  — fetch-исключение → failed; non-200 → blocked+failed; parse→None →
            failed; save→False → БЫЛА ДЫРА; общий except → failed.
  avito   — save→False → БЫЛА ДЫРА; AvitoListingGoneError → gone;
            Blocked/RateLimited → blocked; TimeoutError → failed;
            общий except → failed. Обрывы по budget/SIGTERM стоят ДО
            `attempted += 1`, они попытку не создают.

Формы тождества у файлов РАЗНЫЕ, и это не описка:
  * yandex: `blocked` документирован как ПОДМНОЖЕСТВО `failed` (dataclass,
    #3196) — non-200 инкрементирует оба, поэтому сумма исходов = enriched + failed;
  * avito: `blocked`/`gone`/`failed` — непересекающиеся корзины, сумма исходов =
    enriched + blocked + gone + failed.

Проверка ПО ЗНАЧЕНИЮ: сравниваются числа, а не «не бросило исключение». Вторым
кейсом на каждый файл идёт контроль на противоположную ошибку — что правка не
начисляет исход дважды. Харнессы зеркалят tests/test_3196_yandex_ban_kind.py и
tests/test_3283g_rotate_on_platform_ban.py.
"""

from __future__ import annotations

import os
import sys
from types import SimpleNamespace
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 AvitoBlockedError  # noqa: E402

from app.core import shutdown as _sd  # noqa: E402
from app.tasks.avito_detail_backfill import (  # noqa: E402
    AvitoDetailBackfillResult,
    run_avito_detail_backfill,
)
from app.tasks.yandex_detail_backfill import (  # noqa: E402
    YandexDetailBackfillResult,
    run_yandex_detail_backfill,
)

# ── yandex ────────────────────────────────────────────────────────────────────
_Y_ASYNC_SESSION = "app.tasks.yandex_detail_backfill.AsyncSession"
_Y_PARSE = "app.tasks.yandex_detail_backfill.YandexDetailScraper.parse"
_Y_SAVE = "app.tasks.yandex_detail_backfill.save_detail_enrichment"
_Y_RUNS = "app.tasks.yandex_detail_backfill.runs_mod"
_Y_SLEEP = "app.tasks.yandex_detail_backfill.asyncio.sleep"
_Y_RESOLVE_PROXY_URL = "app.tasks.yandex_detail_backfill.resolve_proxy_url"

# ── avito ─────────────────────────────────────────────────────────────────────
_A_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
_A_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
_A_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
_A_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
_A_SETTINGS = "app.tasks.avito_detail_backfill.settings"
_A_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
_A_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
_A_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"
_A_ROTATE_PROXY = "app.tasks.avito_detail_backfill.rotate_proxy"


@pytest.fixture(autouse=True)
def _reset_shutdown() -> None:
    _sd.reset_shutdown()
    yield
    _sd.reset_shutdown()


def _assert_identity(
    attempted: int, outcomes: int, *, expected_attempted: int, detail: str
) -> None:
    assert attempted == expected_attempted, (
        f"attempted={attempted}, ожидали {expected_attempted} попыток"
    )
    assert attempted == outcomes, (
        f"тождество нарушено: attempted={attempted}, сумма исходов={outcomes} ({detail}), "
        f"потеряно {attempted - outcomes} попыток без исхода"
    )


def _assert_yandex_identity(c: YandexDetailBackfillResult, *, expected_attempted: int) -> None:
    # blocked ⊆ failed (см. докстринг модуля) — в сумму входит только failed.
    _assert_identity(
        c.attempted,
        c.enriched + c.failed,
        expected_attempted=expected_attempted,
        detail=f"enriched={c.enriched} failed={c.failed} (blocked={c.blocked} ⊆ failed)",
    )


def _assert_avito_identity(c: AvitoDetailBackfillResult, *, expected_attempted: int) -> None:
    _assert_identity(
        c.attempted,
        c.enriched + c.blocked + c.gone + c.failed,
        expected_attempted=expected_attempted,
        detail=(f"enriched={c.enriched} blocked={c.blocked} gone={c.gone} failed={c.failed}"),
    )


def _mock_yandex_db(n: int) -> MagicMock:
    snapshot = [
        {"id": i + 1, "source_url": f"https://realty.yandex.ru/offer/{i + 1}/"} for i in range(n)
    ]
    db = MagicMock()
    sel = MagicMock()
    sel.mappings.return_value.all.return_value = snapshot
    sel.one.return_value = SimpleNamespace(url_from_offer_id=0, unenrichable_pending=0)
    db.execute.return_value = sel
    return db


# HTTP 200 обязан выглядеть ПОЛНОЙ карточкой (#3191): недогруженная страница теперь
# отсеивается до parse и даёт исход incomplete ⊆ failed — иначе эти кейсы про
# тождество счётчиков молча стали бы кейсами про полноту.
_FULL_PAGE_HTML = "<html>" + "x" * 1_100_000 + '"encryptedPhones":["a"]</html>'


def _resp(status: int) -> MagicMock:
    resp = MagicMock()
    resp.status_code = status
    resp.text = _FULL_PAGE_HTML if status == 200 else "<html>ok</html>"
    return resp


def _yandex_session_cls(responses: list[MagicMock]) -> MagicMock:
    session = AsyncMock()
    session.get = AsyncMock(side_effect=responses)
    ctx = MagicMock()
    ctx.__aenter__ = AsyncMock(return_value=session)
    ctx.__aexit__ = AsyncMock(return_value=None)
    return MagicMock(return_value=ctx)


async def _run_yandex(
    responses: list[MagicMock], *, save_ok: bool, parse_result: Any = None
) -> YandexDetailBackfillResult:
    count = len(responses)
    db = _mock_yandex_db(count)
    with (
        patch(_Y_ASYNC_SESSION, _yandex_session_cls(responses)),
        patch(_Y_PARSE, return_value=parse_result or MagicMock()),
        patch(_Y_SAVE, return_value=save_ok),
        patch(_Y_RUNS, MagicMock()),
        patch(_Y_SLEEP, new_callable=AsyncMock),
        patch(_Y_RESOLVE_PROXY_URL, MagicMock(return_value="http://proxy:3128")),
    ):
        return await run_yandex_detail_backfill(
            db,
            run_id=3338,
            params={"batch_size": count, "budget_sec": 3600, "max_consecutive_blocks": 10},
        )


@pytest.mark.asyncio
async def test_yandex_missing_row_on_save_keeps_identity() -> None:
    """save_detail_enrichment вернул False (строки уже нет) → попытка не теряется."""
    counters = await _run_yandex([_resp(200), _resp(200)], save_ok=False)

    _assert_yandex_identity(counters, expected_attempted=2)
    assert (counters.enriched, counters.failed, counters.blocked) == (0, 2, 0), (
        f"enriched={counters.enriched} failed={counters.failed} blocked={counters.blocked}: "
        "непрошедший UPDATE — не успех и не блок площадки"
    )


@pytest.mark.asyncio
async def test_yandex_outcomes_counted_once() -> None:
    """Контроль на противоположную ошибку: успех и non-200 — по одному разу.

    non-200 инкрементирует и blocked, и failed НАРОЧНО (blocked ⊆ failed, #3196);
    правка #3338 не должна добавлять там третий инкремент.
    """
    counters = await _run_yandex([_resp(200), _resp(403)], save_ok=True)

    _assert_yandex_identity(counters, expected_attempted=2)
    assert (counters.enriched, counters.failed, counters.blocked) == (1, 1, 1), (
        f"enriched={counters.enriched} failed={counters.failed} blocked={counters.blocked}, "
        "ожидали 1/1/1 — исход начислен дважды"
    )


def _fake_avito_settings() -> MagicMock:
    return MagicMock(
        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",
        avito_detail_backfill_rotate_after_attempts=15,
        avito_detail_backfill_rotate_on_ban_max=0,
        avito_detail_backfill_rotate_on_ban_min_gap=10,
    )


def _mock_avito_db(n: int) -> MagicMock:
    snapshot = [
        {
            "id": i + 1,
            "source_url": f"https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_{i + 1}",
        }
        for i in range(n)
    ]
    db = MagicMock()
    sel = MagicMock()
    sel.mappings.return_value.all.return_value = snapshot
    db.execute.return_value = sel
    return db


def _mock_avito_browser_fetcher_cls() -> MagicMock:
    instance = AsyncMock()
    instance.__aenter__ = AsyncMock(return_value=instance)
    instance.__aexit__ = AsyncMock(return_value=False)
    instance.request_context_reset = MagicMock()
    instance.lease_id = 42
    return MagicMock(return_value=instance)


async def _run_avito(fetch_results: list[Any], *, save_ok: bool) -> AvitoDetailBackfillResult:
    count = len(fetch_results)
    db = _mock_avito_db(count)
    with (
        patch(_A_SETTINGS, _fake_avito_settings()),
        patch(_A_SESSION),
        patch(_A_SCRAPER),
        patch(_A_RUNS, MagicMock()),
        patch(_A_BROWSER_FETCHER, _mock_avito_browser_fetcher_cls()),
        patch(_A_FETCH, AsyncMock(side_effect=fetch_results)),
        patch(_A_SAVE, return_value=save_ok),
        patch(_A_ROTATE_PROXY, AsyncMock()),
        patch(_A_SLEEP, new_callable=AsyncMock),
    ):
        return await run_avito_detail_backfill(
            db,
            run_id=3338,
            params={"batch_size": count, "budget_sec": 3600, "max_consecutive_blocks": 10},
        )


@pytest.mark.asyncio
async def test_avito_missing_row_on_save_keeps_identity() -> None:
    """save_detail_enrichment вернул False (строки уже нет) → попытка не теряется."""
    counters = await _run_avito([MagicMock(), MagicMock()], save_ok=False)

    _assert_avito_identity(counters, expected_attempted=2)
    assert (counters.enriched, counters.failed) == (0, 2), (
        f"enriched={counters.enriched} failed={counters.failed}: непрошедший UPDATE — не успех"
    )
    assert (counters.blocked, counters.gone) == (0, 0), (
        f"blocked={counters.blocked} gone={counters.gone}: площадка ответила и ничего "
        "не снимала — исход отказа НАШЕЙ стороны, чужие корзины трогать нельзя"
    )


@pytest.mark.asyncio
async def test_avito_outcomes_counted_once() -> None:
    """Контроль на противоположную ошибку: успех/блок/таймаут — по одному разу."""
    counters = await _run_avito(
        [MagicMock(), AvitoBlockedError("firewall/soft-block"), TimeoutError("fetch stalled")],
        save_ok=True,
    )

    _assert_avito_identity(counters, expected_attempted=3)
    assert (counters.enriched, counters.blocked, counters.failed, counters.gone) == (1, 1, 1, 0), (
        f"enriched={counters.enriched} blocked={counters.blocked} "
        f"failed={counters.failed} gone={counters.gone}, "
        "ожидали 1/1/1/0 — правка #3338 не должна начислять исход дважды"
    )