"""#3118/#3074: чекпоинты для domclick_city_sweep — корзина ROOM_BUCKETS как единица.

Прод-факты (#3118): QRATOR обрывает свип внутри 1-2-й корзины при ЛЮБОМ старте
(`buckets_completed ≤ 1` из 6 во всех прогонах), banned-прогоны при этом собирают
59–1815 лотов. Сдвиг #2854 лишь распределяет потери; чекпоинт превращает
случайную ротацию в систематический обход — шесть прогонов закрывают шесть
корзин. Моё раннее «чекпоинтить нечего» (комментарий в #3074) опровергнуто
данными #3118 — этот файл и есть исправление того вывода кодом.

Три слоя (зеркально yandex #3074): провайдер skip_buckets + имена завершённых;
пайплайн — resume-чтение и done_buckets в counters (мерж jsonb, финализаторы
не затирают); планировщик — generic _pick_resume (#2845) бесплатно.
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock, patch

from scraper_kit.orchestration import scheduler as sched
from scraper_kit.providers.domclick.serp import ROOM_BUCKETS, DomClickBlockedError, DomClickScraper

# ── 1. Провайдер: скип, имена, порядок с ротацией #2854 ──────────────────────


async def _no_sleep() -> None:
    pass


def _scraper() -> DomClickScraper:
    s = DomClickScraper(SimpleNamespace(scraper_proxy_url=None))
    s.sleep_between_requests = _no_sleep  # type: ignore[method-assign]
    return s


async def _run_fetch_city(
    scraper: DomClickScraper,
    *,
    start: int = 0,
    skip: set[str] | None = None,
    block_on: str | None = None,
) -> list[str]:
    """fetch_city с замоканными сеткой и обходом корзины; возвращает порядок
    корзин, по которым РЕАЛЬНО пошёл обход."""
    visited: list[str] = []

    async def _fake_sweep_bucket(*, rooms: str, **_kw: Any) -> None:
        visited.append(rooms)
        if block_on is not None and rooms == block_on:
            raise DomClickBlockedError("QRATOR")

    class _FakeFetcherCtx:
        async def __aenter__(self) -> SimpleNamespace:
            return SimpleNamespace(report_ban=lambda *_a, **_k: None)

        async def __aexit__(self, *_exc: Any) -> None:
            return None

    with (
        patch.object(scraper, "_sweep_bucket", _fake_sweep_bucket),
        patch(
            "scraper_kit.providers._base.build_browser_fetcher",
            lambda *_a, **_k: _FakeFetcherCtx(),
        ),
    ):
        await scraper.fetch_city(city_id=1, pages=1, start_bucket_index=start, skip_buckets=skip)
    return visited


async def test_skip_buckets_are_never_fetched() -> None:
    """Корзины из чекпоинта не получают ни одного обхода; buckets_total честно
    сжимается до объёма ЭТОГО прогона (иначе honest-status читал бы
    возобновлённый прогон как вечно-частичный)."""
    s = _scraper()
    visited = await _run_fetch_city(s, start=0, skip={"st", "1", "2"})
    assert visited == ["3", "4", "5+"], visited
    assert s.buckets_total == 3
    assert s.completed_buckets == ["3", "4", "5+"]


async def test_block_midway_records_completed_names() -> None:
    """Блок на 2-й корзине: имена завершённых до блока — источник чекпоинта."""
    s = _scraper()
    visited = await _run_fetch_city(s, start=1, block_on="2")
    # старт со сдвигом 1: порядок 1,2,... — блок на '2' после завершения '1'
    assert visited[:2] == ["1", "2"]
    assert s.completed_buckets == ["1"], s.completed_buckets
    assert s.blocked is True


async def test_all_skipped_is_honest_noop() -> None:
    """Цепочка накопила все 6 корзин → пустой обход без падения (гард)."""
    s = _scraper()
    visited = await _run_fetch_city(s, skip=set(ROOM_BUCKETS))
    assert visited == []
    assert s.completed_buckets == []


# ── 2. Планировщик: диспатч отдаёт точку (зеркало test_930/test_3074) ────────


def _candidate() -> SimpleNamespace:
    return SimpleNamespace(
        prev_id=6001,
        prev_status="banned",
        prev_counters={"done_buckets": ["st", "1"]},
        same_params=True,
        age_h=20.0,
        interval_days="1",
    )


class _FakeDb:
    def __init__(self, row: Any) -> None:
        self.row = row

    def execute(self, _stmt: Any, params: dict[str, Any] | None = None) -> Any:
        if params and "counters" in params:
            return MagicMock()
        return MagicMock(fetchone=lambda: self.row)

    def commit(self) -> None:
        pass


async def test_scheduler_hands_checkpoint_to_domclick_sweep() -> None:
    """Красный на main по значению: kwargs без resume_run_id → None != 6001."""
    db = _FakeDb(_candidate())
    captured: dict[str, Any] = {}

    async def _spy(*_a: Any, **kw: Any) -> None:
        captured.update(kw)

    with patch.object(sched, "run_domclick_city_sweep", _spy):
        await sched._job_domclick_city_sweep(db, 7000, {}, MagicMock())

    assert captured.get("resume_run_id") == 6001, (
        "планировщик не отдал чекпоинт домклик-свипу — прогон пойдёт с нуля"
    )