gendesign/tradein-mvp/backend/tests/test_3074_avito_newbuilding_checkpoint.py

"""Чекпоинт по страницам для avito_newbuilding_sweep (#3074).

Последний длинный свип без чекпоинтов: 27.08 прогон убит деплоем на 343-й
минуте, собранное потеряно целиком — save_listings был ОДИН на весь sweep,
ждать в конце было нечего. Единица возобновления — СТРАНИЦА выдачи (у функции
уже есть `pages`; цикл по страницам живёт в `AvitoScraper._paginate_sweep`).

Ключ чекпоинта — номер страницы. В отличие от якорей/combo, страницы строго
последовательны (break-on-empty), поэтому resume — это `start_page =
max(done_buckets) + 1`, без skip-набора произвольных элементов.

ИНВАРИАНТЫ, РАДИ КОТОРЫХ ТЕСТ:
1. Резюм пропускает уже собранные страницы БЕЗ единого запроса к источнику.
2. В чекпоинт попадает только страница, пройденная до конца — оборванная
   исключением/break страница НЕ фиксируется, иначе следующий прогон
   пропустил бы её навсегда, причём молча (прогон завершится штатно).
3. Запись идёт мержем (`counters || :counters`), а не заменой — посторонний
   ключ в `counters` текущего run'а переживает запись чекпоинта.
"""

from __future__ import annotations

import os

# Settings собирается автофикстурой conftest'а и требует database_url. Выставляем
# до остальных импортов — так же, как в test_3074_avito_anchor_checkpoint.py.
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


class _FakeDb:
    """Резюм-SELECT отдаёт counters предшественника; heartbeat'ы записываются."""

    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 _MergingFakeDb(_FakeDb):
    """Как _FakeDb, но heartbeat-запись мержит в текущую строку (jsonb `||`),
    а не просто копится списком — для теста инварианта #3 (посторонний ключ)."""

    def __init__(self, prev_counters: dict[str, Any] | None = None) -> None:
        super().__init__(prev_counters)
        self.row: dict[str, Any] = {}

    def execute(self, _stmt: Any, params: dict[str, Any] | None = None) -> Any:
        if params and "counters" in params:
            incoming = json.loads(params["counters"])
            self.row = {**self.row, **incoming}
            self.heartbeats.append(dict(self.row))
            return MagicMock()
        return super().execute(_stmt, params)


class _FakeAsyncSession:
    def __init__(self, *_a: Any, **_kw: Any) -> None: ...
    async def __aenter__(self) -> _FakeAsyncSession:
        return self

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


def _lot(tag: str) -> MagicMock:
    m = MagicMock()
    m.source_id = tag
    return m


class _FakeScraper:
    """Двойник AvitoScraper: помнит реально пройденные страницы, умеет упасть."""

    visited_pages: list[int] = []  # noqa: RUF012 — тестовый сборник
    fail_on: int | None = None

    def __init__(self, *_a: Any, **_kw: Any) -> None:
        self._browser = None
        self._cffi = None

    async def fetch_newbuildings(
        self,
        *,
        pages: int,
        start_page: int = 1,
        on_page: Any = None,
        delay_override_sec: float | None = None,
    ) -> list[Any]:
        all_lots: list[Any] = []
        for page in range(start_page, pages + 1):
            _FakeScraper.visited_pages.append(page)
            if _FakeScraper.fail_on == page:
                # Страница оборвана (сеть/парсинг) — break, on_page НЕ зовём.
                break
            new_lots = [_lot(f"p{page}-{i}") for i in range(2)]
            all_lots.extend(new_lots)
            if on_page is not None:
                on_page(page, new_lots)
        return all_lots


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


def _make_save(fail_on_call: int | None):
    """save_listings, падающий на N-м вызове — имитация отказа БД на одной странице."""
    calls = {"n": 0}

    def _save(*_a, **_kw):
        calls["n"] += 1
        if fail_on_call is not None and calls["n"] == fail_on_call:
            raise RuntimeError(f"save_listings упал на странице {calls['n']}")
        return (2, 0)

    return _save


async def _run(
    db: _FakeDb,
    *,
    resume_run_id: int | None,
    pages: int = 4,
    fail_on: int | None = None,
    save_fail_on_call: int | None = None,
):
    from scraper_kit.orchestration import pipeline as pl

    _FakeScraper.visited_pages = []
    _FakeScraper.fail_on = fail_on

    with (
        patch.object(pl, "AvitoScraper", _FakeScraper),
        patch.object(pl, "AsyncSession", _FakeAsyncSession),
        patch.object(pl, "save_listings", _make_save(save_fail_on_call)),
    ):
        await pl.run_avito_newbuilding_sweep(
            db,  # type: ignore[arg-type]
            run_id=8001,
            config=_config(),
            matcher=MagicMock(),
            pages=pages,
            request_delay_sec=0.0,
            resume_run_id=resume_run_id,
        )
    return db


def _last_checkpoint(db: _FakeDb) -> list[int]:
    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_resume_skips_done_pages_and_fetches_the_rest() -> None:
    """Резюм с чекпоинтом [1,2] на pages=4 обходит только страницы 3 и 4 —
    ни одного запроса к уже собранным."""
    db = _FakeDb({"done_buckets": [1, 2]})
    await _run(db, resume_run_id=7999, pages=4)

    assert _FakeScraper.visited_pages == [3, 4], (
        "уже собранные страницы 1/2 не должны опрашиваться повторно"
    )
    assert _last_checkpoint(db) == [1, 2, 3, 4], "чекпоинт дописывается поверх унаследованного"


@pytest.mark.asyncio
async def test_without_resume_starts_from_page_one() -> None:
    """Без resume_run_id поведение прежнее — обход с первой страницы."""
    db = _FakeDb(None)
    await _run(db, resume_run_id=None, pages=3)

    assert _FakeScraper.visited_pages == [1, 2, 3]
    assert _last_checkpoint(db) == [1, 2, 3]


@pytest.mark.asyncio
async def test_failed_page_does_not_enter_checkpoint() -> None:
    """Страница, оборванная на середине (fail_on=2), НЕ считается пройденной.

    Иначе следующий прогон пропустит её навсегда молча — sweep завершится
    штатно, просто часть выдачи не соберётся никогда.
    """
    db = _FakeDb(None)
    await _run(db, resume_run_id=None, pages=3, fail_on=2)

    # Страница 2 была АТАКОВАНА (попытка была — visited), но не пройдена до конца.
    assert _FakeScraper.visited_pages == [1, 2]
    ckpt = _last_checkpoint(db)
    assert ckpt == [1], "упавшая страница 2 (и не начатая 3) не должны попасть в чекпоинт"


@pytest.mark.asyncio
async def test_checkpoint_write_merges_and_preserves_foreign_key() -> None:
    """Запись чекпоинта — мерж (`counters || :counters`), не замена.

    Симулируем текущую строку run'а с посторонним ключом (например, оставленным
    другим писателем/предыдущим heartbeat'ом) — он обязан пережить наши записи.
    """
    db = _MergingFakeDb(None)
    db.row = {"foreign_key": "survives-me"}

    await _run(db, resume_run_id=None, pages=1)

    assert db.row.get("foreign_key") == "survives-me", (
        "посторонний ключ в counters не пережил запись чекпоинта — запись была заменой, не мержем"
    )
    assert "done_buckets" in db.row


@pytest.mark.asyncio
async def test_page_whose_save_failed_does_not_enter_checkpoint() -> None:
    """Страница собрана, но её save упал — в чекпоинт она попасть не должна.

    Отказ save_listings перехватывается и прогон продолжается (это осознанно:
    одна упавшая страница не должна ронять весь sweep). Но если отметить её
    пройденной, следующий прогон её пропустит, и объявления оттуда не соберутся
    НИКОГДА — молча, потому что прогон завершится штатно.
    """
    db = _FakeDb(None)
    await _run(db, resume_run_id=None, pages=3, save_fail_on_call=2)

    assert _FakeScraper.visited_pages == [1, 2, 3]
    ckpt = _last_checkpoint(db)
    assert 2 not in ckpt, "страница с упавшим save попала в чекпоинт — покрытие потеряно молча"
    assert ckpt == [1, 3]


@pytest.mark.asyncio
async def test_resume_starts_at_first_gap_not_after_the_last_page() -> None:
    """Дыра в чекпоинте перечитывается, а не перепрыгивается.

    `max(done)+1` пропустил бы страницу 2 навсегда. Продолжаем с первой
    несобранной: страницы после дыры перечитаются, что дешевле потери и
    безопасно — повторная запись схлопывается по dedup_hash.
    """
    db = _FakeDb({"done_buckets": [1, 3]})
    await _run(db, resume_run_id=7777, pages=4)

    assert _FakeScraper.visited_pages[0] == 2, (
        "подхват начался не с дыры — пропущенная страница не соберётся никогда"
    )