"""Чекпоинт по якорям для cian_city_sweep (#3074).

Замер за 60 дней, по которому выбран источник: 65 прогонов, среднее 35 минут,
максимум 72, две отмены деплоем. Пятиминутного дренажа (#3029) на такие прогоны
не хватает — убитый на 35-й минуте сбор начинался заново с первого якоря.

Ключ чекпоинта — ИМЯ якоря, а не индекс: состав списка зависит от `city_slug`
(областные свипы идут по своим наборам), позиция между городами не устойчива.

ИНВАРИАНТ. В чекпоинт попадает только якорь, пройденный до конца. У циана эта
граница уже проведена потоком управления: все ветки отказа делают `return` или
`continue`, и до записи не доходят. Этим он отличается от avito-свипа, где успех
и неудача сходились в одной строке и потребовался отдельный флаг. Тест ниже
проверяет, что граница не нарушена: упавший якорь не должен попасть в чекпоинт,
иначе следующий прогон пропустит его навсегда — молча, потому что прогон
завершится штатно, просто часть города не соберётся.
"""

from __future__ import annotations

import os

# Settings собирается автофикстурой conftest'а и требует database_url.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

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

import pytest

ANCHOR_A = (56.83, 60.60, "ekb-center")
ANCHOR_B = (56.79, 60.63, "ekb-south")


class _FakeDb:
    def __init__(self, prev_counters: dict[str, Any] | None = None) -> None:
        self.prev_counters = prev_counters or {}
        self.heartbeats: list[dict[str, Any]] = []

    def execute(self, _stmt: Any, params: dict[str, Any] | None = None) -> Any:
        if params and "counters" in params:
            self.heartbeats.append(json.loads(params["counters"]))
            return MagicMock()
        if params and "rid" in params:
            return MagicMock(fetchone=lambda: types.SimpleNamespace(counters=self.prev_counters))
        return MagicMock()

    def commit(self) -> None: ...
    def rollback(self) -> None: ...


class _FakeScraper:
    """Двойник CianScraper: помнит, за какими якорями реально ходили."""

    visited: list[tuple[float, float]] = []  # noqa: RUF012 — тестовый сборник
    raise_on: tuple[float, float] | None = None

    def __init__(self, *_a: Any, **_kw: Any) -> None:
        # Счётчики, которые конвейер читает у скрапера после каждого якоря
        # (#2625 — диагностика «все якоря упали»). Без них падает не проверяемая
        # логика, а сам двойник.
        self.state_extraction_attempts = 1
        self.state_extraction_failures = 0
        self.request_delay_sec = 0.0

    async def __aenter__(self) -> _FakeScraper:
        return self

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

    async def fetch_around_multi_room(self, lat: float, lon: float, *_a: Any, **_kw: Any) -> list:
        _FakeScraper.visited.append((lat, lon))
        if _FakeScraper.raise_on == (lat, lon):
            raise RuntimeError("якорь упал")
        return []


def _config() -> types.SimpleNamespace:
    return types.SimpleNamespace(
        scraper_proxy_url=None,
        scraper_fetch_mode="cffi",
        use_proxy_pool_browser=False,
        browser_http_endpoint=None,
        environment="test",
    )


async def _run(prev: dict[str, Any] | None, raise_on: tuple[float, float] | None = None) -> _FakeDb:
    from scraper_kit.orchestration import pipeline as pl

    _FakeScraper.visited = []
    _FakeScraper.raise_on = raise_on
    db = _FakeDb(prev)

    with (
        patch.object(pl, "CianScraper", _FakeScraper),
        patch.object(pl, "save_listings", lambda *_a, **_kw: (0, 0)),
        patch.object(pl.runs, "is_cancelled", lambda *_a: False),
    ):
        await pl.run_cian_city_sweep(
            db,  # type: ignore[arg-type]
            run_id=8001,
            config=_config(),
            matcher=MagicMock(),
            anchors=[ANCHOR_A, ANCHOR_B],
            enrich_houses=False,
            detail_top_n=0,
            request_delay_sec=0.0,
            resume_run_id=7999 if prev is not None else None,
        )
    return db


def _last_checkpoint(db: _FakeDb) -> list[str]:
    with_ckpt = [hb for hb in db.heartbeats if "done_buckets" in hb]
    assert with_ckpt, "ни один heartbeat не унёс done_buckets — чекпоинт не персистится"
    return with_ckpt[-1]["done_buckets"]


@pytest.mark.asyncio
async def test_checkpointed_anchor_is_skipped_without_a_single_request() -> None:
    """Якорь из чекпоинта не опрашивается вовсе."""
    await _run({"done_buckets": ["ekb-center"]})
    assert (ANCHOR_A[0], ANCHOR_A[1]) not in _FakeScraper.visited, (
        "якорь из чекпоинта всё-таки опрашивали"
    )
    assert (ANCHOR_B[0], ANCHOR_B[1]) in _FakeScraper.visited, "второй якорь не обошли"


@pytest.mark.asyncio
async def test_checkpoint_accumulates_over_inherited() -> None:
    """Пройденный якорь дописывается поверх унаследованных, а не затирает их."""
    db = await _run({"done_buckets": ["ekb-center"]})
    assert _last_checkpoint(db) == ["ekb-center", "ekb-south"]


@pytest.mark.asyncio
async def test_without_resume_all_anchors_are_visited() -> None:
    """Без чекпоинта поведение прежнее — обходятся все якоря."""
    db = await _run(None)
    assert len(_FakeScraper.visited) == 2
    assert _last_checkpoint(db) == ["ekb-center", "ekb-south"]


@pytest.mark.asyncio
async def test_failed_anchor_does_not_enter_checkpoint() -> None:
    """Упавший якорь НЕ считается пройденным.

    У циана это обеспечено потоком управления, а не флагом: ветка отказа делает
    `continue` до записи. Тест сторожит именно это — рефакторинг, сливающий ветки
    в одну, сломает инвариант незаметно.
    """
    db = await _run(None, raise_on=(ANCHOR_A[0], ANCHOR_A[1]))
    ckpt = _last_checkpoint(db)
    assert "ekb-center" not in ckpt, "упавший якорь попал в чекпоинт"
    assert "ekb-south" in ckpt, "исправный якорь не зафиксирован"