"""#2674, Avito full load: окно ретроспективы под недельный такт + 503 сайдкара.

Две находки эпика, обе — продолжающаяся потеря данных, а не история.

1. Окно разошлось с тактом. `interval_days` (каденс, миграция 206) и
   `incremental_days` (глубина обхода, миграция 129) — два независимых литерала,
   которые обязаны совпадать. На проде было {interval_days: 7, incremental_days: 2}:
   прогон видит listing_date в [D-2, D] = 3 суток из 7, дни D+1..D+4 не попадают
   ни в один прогон. Чиним в двух местах — данными (миграция 215) и кодом
   (scheduler расширяет окно до такта, чтобы расхождение не вернулось).

2. 503 сайдкара не ретраился ни разу. `_fetch_serp_html_browser` классифицировал
   "browser unavailable (proxy may be down)" как soft-ban и отправлял в бюджет
   IP-ротации, а ротация снята в #2616 (max_rot=0, `_rotate_ip()` всегда False) →
   условие `rot_done < max_rot` ложно всегда. Бюджет коротких backoff-retry стоял
   в `else:` и для soft-ban был структурно недостижим. Итог на проде: один блип
   сайдкара съедал бакет, четыре подряд — весь прогон
   (`_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED = 4`).
"""

from __future__ import annotations

import re
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import httpx
import pytest
from scraper_kit.avito_exceptions import AvitoRateLimitedError
from scraper_kit.orchestration import scheduler as kit_sched
from scraper_kit.orchestration.scheduler import (
    SchedulerContext,
    _job_avito_full_load,
    _job_avito_full_load_exhaustive,
)
from scraper_kit.providers.avito import serp as serp_module
from scraper_kit.providers.avito.serp import _AVITO_SIDECAR_TRANSIENT_RETRIES, AvitoScraper

from app.services.scraper_adapters import RealScraperConfig

_MIGRATION_215 = (
    Path(__file__).resolve().parents[1]
    / "data"
    / "sql"
    / "215_avito_full_load_window_matches_cadence.sql"
)


def _ctx() -> SchedulerContext:
    return SchedulerContext(
        config=MagicMock(),
        matcher=MagicMock(),
        enrichment=MagicMock(),
        session_factory=MagicMock(),
        runs=MagicMock(),
    )


async def _incremental_days_passed(params: dict) -> object:
    """Прогнать _job_avito_full_load с params и вернуть переданный incremental_days."""
    with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run:
        await _job_avito_full_load(MagicMock(), 1, params, _ctx())
    mock_run.assert_awaited_once()
    _args, kwargs = mock_run.call_args
    return kwargs.get("incremental_days")


# ── 1. Окно ретроспективы не уже такта ────────────────────────────────────────


async def test_window_widened_to_cadence() -> None:
    """Прод-конфиг на момент находки: такт 7 суток, окно 2 → окно расширяется до 7.

    Фальсификация: без правки в scheduler._job_avito_full_load сюда приезжает 2 —
    ровно те «двое суток из семи», о которых говорит эпик.
    """
    assert await _incremental_days_passed({"interval_days": 7, "incremental_days": 2}) == 7


async def test_window_wider_than_cadence_left_alone() -> None:
    """Окно ШИРЕ такта — осознанный запас оператора, не сужаем."""
    assert await _incremental_days_passed({"interval_days": 3, "incremental_days": 10}) == 10


async def test_daily_cadence_keeps_window() -> None:
    """Back-compat: без interval_days такт = 1 сутки, окно 2 уже перекрывает его."""
    assert await _incremental_days_passed({"incremental_days": 2}) == 2


async def test_null_interval_days_does_not_crash() -> None:
    """`"interval_days": null` в jsonb приезжает сюда как None, а int(None) — TypeError."""
    assert await _incremental_days_passed({"interval_days": None, "incremental_days": 2}) == 2


async def test_no_window_stays_exhaustive() -> None:
    """Строка без incremental_days = полный обход; такт не должен её «инкрементализировать»."""
    assert await _incremental_days_passed({"interval_days": 7}) is None


async def test_exhaustive_job_ignores_window_params() -> None:
    """Соседняя джоба всегда идёт полным обходом, что бы ни лежало в её params."""
    with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run:
        await _job_avito_full_load_exhaustive(
            MagicMock(), 1, {"interval_days": 7, "incremental_days": 2}, _ctx()
        )
    _args, kwargs = mock_run.call_args
    assert kwargs.get("incremental_days") is None


# ── 2. 503 сайдкара получает backoff-retry, а не мгновенный отказ ─────────────


def _sidecar_503() -> httpx.HTTPStatusError:
    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
async def test_sidecar_503_retried_before_giving_up() -> None:
    """503 «browser unavailable» — это упавший launch камуфокса, а не бан площадки.

    Сайдкар на таком отказе сам поднимает фоновый retry launch'а, поэтому повтор
    через пару секунд обычно проходит. Ждём 1 попытку + весь бюджет backoff-retry.

    Фальсификация: без правки soft-ban уходит в ветку ротации (max_rot=0 → условие
    ложно всегда), бюджет backoff недостижим → ровно 1 вызов fetch и немедленный
    AvitoRateLimitedError.
    """
    scraper = AvitoScraper(RealScraperConfig())
    assert scraper._cffi is None
    scraper._browser = AsyncMock()
    scraper._browser.fetch = AsyncMock(side_effect=_sidecar_503())

    with patch.object(serp_module.asyncio, "sleep", AsyncMock()):
        with pytest.raises(AvitoRateLimitedError):
            await scraper._fetch_serp_html(
                "https://www.avito.ru/ekaterinburg/kvartiry/prodam", page=1
            )

    assert scraper._browser.fetch.await_count == 1 + _AVITO_SIDECAR_TRANSIENT_RETRIES


@pytest.mark.asyncio
async def test_sidecar_503_recovers_without_aborting_bucket() -> None:
    """Один блип сайдкара больше не стоит бакета: вторая попытка отдаёт HTML."""
    scraper = AvitoScraper(RealScraperConfig())
    scraper._browser = AsyncMock()
    scraper._browser.fetch = AsyncMock(
        side_effect=[_sidecar_503(), "<html><body>serp</body></html>"]
    )

    with patch.object(serp_module.asyncio, "sleep", AsyncMock()):
        html = await scraper._fetch_serp_html(
            "https://www.avito.ru/ekaterinburg/kvartiry/prodam", page=1
        )

    assert html == "<html><body>serp</body></html>"
    assert scraper._browser.fetch.await_count == 2


# ── 3. Миграция 215: статические инварианты ──────────────────────────────────
# Живой БД в тестах нет (см. conftest.py) — эффект UPDATE'а проверяется на проде,
# здесь фиксируем форму файла: транзакционность, отсутствие DDL, точное совпадение
# source (соседняя джоба с похожим именем ловится подстрокой), вывод значения из
# фактического interval_days вместо литерала.


def _sql() -> str:
    return _MIGRATION_215.read_text(encoding="utf-8")


def _executable_sql() -> str:
    """SQL без построчных `--`-комментариев — только исполняемый код."""
    lines = []
    for raw in _sql().splitlines():
        code = raw.split("--", 1)[0]
        if code.strip():
            lines.append(code)
    return "\n".join(lines)


def _flat() -> str:
    return re.sub(r"\s+", " ", _executable_sql()).strip().lower()


def test_migration_215_is_transactional_and_data_only() -> None:
    flat = _flat()
    assert flat.startswith("begin;")
    assert flat.endswith("commit;")
    for ddl in ("create ", "alter ", "drop ", "truncate "):
        assert ddl not in flat, f"миграция только про данные, найден DDL: {ddl!r}"


def test_migration_215_targets_only_avito_full_load() -> None:
    """Точное равенство source, НЕ подстрока — 'avito_full_load_exhaustive' рядом."""
    flat = _flat()
    assert "where source = 'avito_full_load'" in flat
    assert "avito_full_load_exhaustive" not in _executable_sql()
    assert " like " not in flat


def test_migration_215_derives_window_from_cadence() -> None:
    """Окно выводится из interval_days строки, а не хардкодится числом."""
    flat = _flat()
    assert "interval_days" in flat
    assert "greatest(" in flat
    assert re.search(r"'incremental_days',\s*greatest", flat) is not None


def test_migration_215_merges_params_not_overwrites() -> None:
    """`||` мерджит ключ в default_params — соседние ключи обязаны выжить."""
    flat = _flat()
    assert "default_params ||" in flat or "default_params\n||" in _executable_sql().lower()
    assert "set default_params = jsonb_build_object" not in flat


def test_migration_215_has_no_psycopg_cast_trap() -> None:
    """Repo-конвенция: CAST(x AS type), не x::type."""
    assert "::" not in _executable_sql()
    assert "cast(" in _flat()