From 3320805723e83460302dc5696d0e244998158270 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 28 Jun 2026 20:52:30 +0300 Subject: [PATCH] feat(tradein/scraper): cooperative SIGTERM-drain coverage for sweep/full-load loops (#1182 Phase 3a) Extend the cooperative shutdown-drain (Phase 2) to the long sweep and full-load loops in scrape_pipeline.py so they stop CLEANLY on SIGTERM instead of being hard-cancelled at the 80s _drain_inflight timeout. At every between-unit checkpoint that already polls scrape_runs.is_cancelled (user-cancel) we add an ALONGSIDE shutdown_requested() branch: - anchor-loop sweeps (avito/yandex/cian city) + avito newbuilding + domclick pre-SERP: update_heartbeat + mark_done(partial) + return; - full-load on_bucket callbacks (cian/yandex/avito): raise a distinct RuntimeError("shutdown") sentinel, handled next to the existing "cancelled" sentinel -> mark_done(partial), skipping the detail phase; - cian full-load detail loop: break -> existing post-loop mark_done. Drain finalizes as mark_done(partial), NOT mark_cancelled (that stays user-cancel only). is_cancelled -> user-cancel semantics are unchanged, and behaviour is identical when shutdown_requested() is False. No resume_cursor or paused status yet (Phase 3b, deferred): this is drain-coverage only. Tests: tests/test_sweep_drain.py covers both structural variants - run_avito_city_sweep / run_cian_city_sweep (anchor-loop) and run_avito_full_load (on_bucket sentinel): shutdown flips True after the first unit, asserting the sweep stops at the checkpoint, calls mark_done with partial counters, never calls mark_cancelled, and does not process later anchors/buckets/phases. --- .../backend/app/services/scrape_pipeline.py | 136 +++++++- tradein-mvp/backend/tests/test_sweep_drain.py | 310 ++++++++++++++++++ 2 files changed, 443 insertions(+), 3 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_sweep_drain.py diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 90aa11d7..b33f6142 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -41,6 +41,7 @@ from sqlalchemy import text from sqlalchemy.orm import Session from app.core.config import settings +from app.core.shutdown import shutdown_requested from app.services import scrape_runs from app.services.scrapers.avito import AvitoScraper from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment @@ -763,6 +764,21 @@ async def run_avito_city_sweep( ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) return counters + elif shutdown_requested(): + # Кооперативный SIGTERM-drain (#1182 Phase 3a): останавливаемся + # на границе anchor'а и финализируем run как mark_done(partial) — + # это НЕ user-cancel, поэтому НЕ mark_cancelled. Дальнейшие anchor'ы + # и IMV-фаза не выполняются. + logger.info( + "city-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)", + run_id, + idx, + len(_anchors), + name, + ) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_done(db, run_id, counters.to_dict()) + return counters logger.info( "city-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", @@ -1250,6 +1266,16 @@ async def run_avito_city_sweep( ) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters + elif shutdown_requested(): + # SIGTERM-drain до IMV-фазы: финализируем без дорогой IMV-оценки. + logger.info( + "city-sweep run_id=%d: SIGTERM-drain — skipping IMV phase " + "(%d houses skipped)", + run_id, + len(all_touched_house_ids), + ) + scrape_runs.mark_done(db, run_id, counters.to_dict()) + return counters logger.info( "city-sweep run_id=%d: IMV phase — %d touched houses", @@ -1384,6 +1410,13 @@ async def run_avito_newbuilding_sweep( logger.info("nb-sweep run_id=%d: cancelled before SERP phase", run_id) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters + elif shutdown_requested(): + logger.info( + "nb-sweep run_id=%d: SIGTERM-drain — stopping before SERP phase", run_id + ) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_done(db, run_id, counters.to_dict()) + return counters scraper = AvitoScraper() if browser_mode: @@ -1582,6 +1615,18 @@ async def run_yandex_city_sweep( ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) return counters + elif shutdown_requested(): + # SIGTERM-drain: чистый стоп на границе anchor'а → mark_done(partial). + logger.info( + "yandex-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)", + run_id, + idx, + len(_anchors), + name, + ) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_done(db, run_id, counters.to_dict()) + return counters logger.info( "yandex-sweep run_id=%d center-combos #%d/%d (%s, %.4f, %.4f) " @@ -2076,6 +2121,18 @@ async def run_cian_city_sweep( ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) return counters + elif shutdown_requested(): + # SIGTERM-drain: чистый стоп на границе anchor'а → mark_done(partial). + logger.info( + "cian-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)", + run_id, + idx, + len(_anchors), + name, + ) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_done(db, run_id, counters.to_dict()) + return counters logger.info( "cian-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", @@ -2506,6 +2563,15 @@ async def run_cian_full_load( if scrape_runs.is_cancelled(db, run_id): logger.info("cian-full-load run_id=%d: cancel detected in on_bucket", run_id) raise RuntimeError("cancelled") + elif shutdown_requested(): + # SIGTERM-drain: тот же sentinel-механизм, что и user-cancel, но отдельный + # маркер "shutdown" → handler финализирует mark_done(partial), не mark_failed. + logger.info( + "cian-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", + run_id, + bucket_key, + ) + raise RuntimeError("shutdown") if not lots: done.add(bucket_key) scrape_runs.update_heartbeat( @@ -2639,6 +2705,14 @@ async def run_cian_full_load( if scrape_runs.is_cancelled(db, run_id): logger.info("cian-full-load run_id=%d: cancelled during detail enrich", run_id) break + elif shutdown_requested(): + logger.info( + "cian-full-load run_id=%d: SIGTERM-drain — stopping detail at #%d/%d", + run_id, + didx + 1, + len(priority_rows), + ) + break listing_id: int = row["id"] source_url: str = row["source_url"] counters.detail_attempted += 1 @@ -2684,7 +2758,8 @@ async def run_cian_full_load( return counters except RuntimeError as exc: - # on_bucket кидает RuntimeError("cancelled") при cooperative cancel + # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, + # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). if str(exc) == "cancelled": logger.info( "cian-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", @@ -2695,6 +2770,16 @@ async def run_cian_full_load( ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters + if str(exc) == "shutdown": + logger.info( + "cian-full-load run_id=%d: SIGTERM-drain — partial results unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters logger.exception("cian-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise @@ -2790,6 +2875,13 @@ async def run_yandex_full_load( if scrape_runs.is_cancelled(db, run_id): logger.info("yandex-full-load run_id=%d: cancel detected in on_bucket", run_id) raise RuntimeError("cancelled") + elif shutdown_requested(): + logger.info( + "yandex-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", + run_id, + bucket_key, + ) + raise RuntimeError("shutdown") if not lots: done.add(bucket_key) scrape_runs.update_heartbeat( @@ -2867,7 +2959,8 @@ async def run_yandex_full_load( return counters except RuntimeError as exc: - # on_bucket кидает RuntimeError("cancelled") при cooperative cancel + # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, + # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). if str(exc) == "cancelled": logger.info( "yandex-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", @@ -2878,6 +2971,17 @@ async def run_yandex_full_load( ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters + if str(exc) == "shutdown": + logger.info( + "yandex-full-load run_id=%d: SIGTERM-drain — partial results " + "unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters logger.exception("yandex-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise @@ -2983,6 +3087,13 @@ async def run_avito_full_load( if scrape_runs.is_cancelled(db, run_id): logger.info("avito-full-load run_id=%d: cancel detected in on_bucket", run_id) raise RuntimeError("cancelled") + elif shutdown_requested(): + logger.info( + "avito-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", + run_id, + bucket_key, + ) + raise RuntimeError("shutdown") if not lots: done.add(bucket_key) scrape_runs.update_heartbeat( @@ -3059,7 +3170,8 @@ async def run_avito_full_load( return counters except RuntimeError as exc: - # on_bucket кидает RuntimeError("cancelled") при cooperative cancel + # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, + # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). if str(exc) == "cancelled": logger.info( "avito-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", @@ -3070,6 +3182,17 @@ async def run_avito_full_load( ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters + if str(exc) == "shutdown": + logger.info( + "avito-full-load run_id=%d: SIGTERM-drain — partial results " + "unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters logger.exception("avito-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise @@ -3173,6 +3296,13 @@ async def run_domclick_city_sweep( logger.info("domclick-sweep run_id=%d: cancelled before SERP", run_id) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) return counters + elif shutdown_requested(): + # SIGTERM-drain до SERP-фазы: финализируем run как mark_done(partial), + # минуя honest-status (это чистый drain, не QRATOR-блок). + logger.info("domclick-sweep run_id=%d: SIGTERM-drain — stopping before SERP", run_id) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_done(db, run_id, counters.to_dict()) + return counters logger.info( "domclick-sweep run_id=%d: BFF citywide sweep city_id=%d " diff --git a/tradein-mvp/backend/tests/test_sweep_drain.py b/tradein-mvp/backend/tests/test_sweep_drain.py new file mode 100644 index 00000000..7f7d7722 --- /dev/null +++ b/tradein-mvp/backend/tests/test_sweep_drain.py @@ -0,0 +1,310 @@ +"""Тесты кооперативного SIGTERM-drain в длинных sweep/full-load циклах (#1182 Phase 3a). + +Проверяют, что при выставленном shutdown_requested() sweep'ы scrape_pipeline.py +останавливаются ЧИСТО на своих between-unit checkpoint'ах: + - финализируют run как mark_done(partial) — НЕ mark_cancelled (это не user-cancel); + - НЕ обрабатывают последующие anchor'ы / бакеты / дорогие фазы (IMV / detail); + - user-cancel ветка (is_cancelled) не задействуется. + +Тесты network-free: внутренние scrape-вызовы и lifecycle скрейперов monkeypatch'атся, +shutdown_requested() патчится контролируемым флагом, переключающимся в True ПОСЛЕ +первого unit'а (а не с самого старта) — чтобы первый unit прошёл, а на втором +сработал drain. +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.services import scrape_pipeline +from app.services.scrapers.avito import AvitoScraper +from app.services.scrapers.base import ScrapedLot +from app.services.scrapers.cian import CianScraper + +# Два anchor'а: первый успешен, второй должен быть отрезан drain'ом. +TWO_ANCHORS: list[tuple[float, float, str]] = [ + (56.8400, 60.6050, "First"), + (56.7950, 60.5300, "Second"), +] + + +# ── Fake scrape_runs (записывает финализаторы) ────────────────────────────── + + +class _FakeRuns: + def __init__(self) -> None: + self.heartbeats: list[dict[str, int]] = [] + self.done: dict[str, int] | None = None + self.failed: tuple[str, dict[str, int]] | None = None + self.banned: tuple[str, dict[str, int]] | None = None + self.cancelled_calls: int = 0 + self.is_cancelled_calls: int = 0 + + def is_cancelled(self, _db: Any, _run_id: int) -> bool: + self.is_cancelled_calls += 1 + return False + + def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None: + self.heartbeats.append(dict(counters)) + + def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None: + self.done = dict(counters) + + def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None: + self.failed = (error, dict(counters)) + + def mark_banned(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None: + self.banned = (error, dict(counters)) + + def mark_cancelled(self, _db: Any, _run_id: int) -> bool: + self.cancelled_calls += 1 + return True + + +def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None: + monkeypatch.setattr(scrape_pipeline.scrape_runs, "is_cancelled", fake.is_cancelled) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "update_heartbeat", fake.update_heartbeat) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_done", fake.mark_done) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_failed", fake.mark_failed) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_banned", fake.mark_banned) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_cancelled", fake.mark_cancelled) + + +def _fake_lot(source: str, offer_id: str) -> ScrapedLot: + return ScrapedLot( + source=source, + source_url=f"https://example.ru/{offer_id}/", + source_id=offer_id, + address="Екатеринбург, улица Тестовая, 1", + price_rub=4_000_000, + area_m2=42.0, + rooms=1, + ) + + +# ── Avito city sweep: drain на границе anchor'а ───────────────────────────── + + +async def test_avito_city_sweep_drains_on_shutdown(monkeypatch: pytest.MonkeyPatch) -> None: + """SIGTERM после первого anchor'а → sweep останавливается на втором anchor'е: + + - второй anchor НЕ фетчится (drain до начала его фаз); + - run финализируется mark_done(partial) с anchors_done==1; + - mark_cancelled НЕ вызывается (это drain, не user-cancel); + - mark_failed / mark_banned не вызываются. + """ + import curl_cffi.requests as _cffi_mod + + state = {"flag": False} + monkeypatch.setattr(scrape_pipeline, "shutdown_requested", lambda: state["flag"]) + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock()) + + fetch_calls: list[str] = [] + + async def fake_fetch_around( + self: AvitoScraper, + lat: float, + lon: float, + radius_m: int, + pages: int = 1, + delay_override_sec: float | None = None, + ) -> list[ScrapedLot]: + fetch_calls.append(f"{lat:.4f}") + lots = [_fake_lot("avito", f"{lat:.4f}-{i}") for i in range(3)] + # SIGTERM приходит ПОСЛЕ первого anchor'а — следующий top-of-loop checkpoint + # увидит флаг и сделает drain. + state["flag"] = True + return lots + + monkeypatch.setattr(AvitoScraper, "fetch_around", fake_fetch_around) + monkeypatch.setattr(scrape_pipeline, "save_listings", lambda _db, lots, **_kw: (len(lots), 0)) + + # Stub shared curl_cffi AsyncSession (non-browser fetch mode). + fake_session = AsyncMock() + fake_session.__aenter__ = AsyncMock(return_value=fake_session) + fake_session.__aexit__ = AsyncMock(return_value=None) + monkeypatch.setattr(_cffi_mod, "AsyncSession", lambda **_kw: fake_session) + + # IMV-фаза не должна вызываться при drain (enrich_imv=False всё равно). + monkeypatch.setattr( + "app.services.house_imv_backfill.process_houses_imv_batch", + AsyncMock(side_effect=AssertionError("IMV phase must not run after drain")), + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_avito_city_sweep( + db=MagicMock(), + run_id=1, + anchors=TWO_ANCHORS, + enrich_houses=False, + detail_top_n=0, + enrich_imv=False, + request_delay_sec=0.0, + ) + + # Второй anchor не дошёл до SERP — fetch вызван ровно один раз. + assert fetch_calls == ["56.8400"], f"только первый anchor должен фетчиться; got {fetch_calls}" + # Partial-финализация: обработан только первый anchor. + assert counters.anchors_done == 1 + assert counters.lots_fetched == 3 + # mark_done(partial) вызван (drain-ветка), и счётчики partial. + assert fake.done is not None, "drain должен вызвать mark_done(partial)" + assert fake.done["anchors_done"] == 1 + # Это НЕ user-cancel. + assert fake.cancelled_calls == 0, "mark_cancelled не должен вызываться при drain" + assert fake.failed is None + assert fake.banned is None + + +# ── Cian city sweep: drain на границе anchor'а ────────────────────────────── + + +@pytest.fixture +def _stub_cian_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None: + def _init(self: CianScraper) -> None: + self.request_delay_sec = 0.0 + self._cffi = None + self._browser = None + + async def _aenter(self: CianScraper) -> CianScraper: + return self + + async def _aexit(self: CianScraper, *_args: Any) -> None: + return None + + monkeypatch.setattr(CianScraper, "__init__", _init) + monkeypatch.setattr(CianScraper, "__aenter__", _aenter) + monkeypatch.setattr(CianScraper, "__aexit__", _aexit) + + +async def test_cian_city_sweep_drains_on_shutdown( + monkeypatch: pytest.MonkeyPatch, _stub_cian_lifecycle: None +) -> None: + """SIGTERM после первого anchor'а → cian sweep drain на втором anchor'е → mark_done(partial).""" + state = {"flag": False} + monkeypatch.setattr(scrape_pipeline, "shutdown_requested", lambda: state["flag"]) + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock()) + + serp_calls: list[str] = [] + + async def fake_fetch_multi( + self: CianScraper, lat: float, lon: float, r: int, pages: int = 3 + ) -> list[ScrapedLot]: + serp_calls.append(f"{lat:.4f}") + state["flag"] = True + return [_fake_lot("cian", f"{lat:.4f}-1")] + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", AsyncMock(return_value=None) + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=2, + anchors=TWO_ANCHORS, + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + ) + + assert serp_calls == ["56.8400"], f"только первый anchor должен фетчиться; got {serp_calls}" + assert counters.anchors_done == 1 + assert fake.done is not None, "drain должен вызвать mark_done(partial)" + assert fake.done["anchors_done"] == 1 + assert fake.cancelled_calls == 0 + assert fake.failed is None + assert fake.banned is None + + +# ── Avito full load: drain через sentinel в on_bucket ─────────────────────── + + +@pytest.fixture +def _stub_avito_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None: + def _init(self: AvitoScraper) -> None: + self.request_delay_sec = 0.0 + self._cffi = None + self._browser = None + + async def _aenter(self: AvitoScraper) -> AvitoScraper: + return self + + async def _aexit(self: AvitoScraper, *_args: Any) -> None: + return None + + monkeypatch.setattr(AvitoScraper, "__init__", _init) + monkeypatch.setattr(AvitoScraper, "__aenter__", _aenter) + monkeypatch.setattr(AvitoScraper, "__aexit__", _aexit) + + +async def test_avito_full_load_drains_on_shutdown( + monkeypatch: pytest.MonkeyPatch, _stub_avito_lifecycle: None +) -> None: + """SIGTERM после первого бакета → on_bucket поднимает sentinel → mark_done(partial): + + - бакет №2 поднимает RuntimeError("shutdown") (НЕ "cancelled"), бакет №3 не достигается; + - done_buckets == ["b1"] (бакет, упавший на drain-checkpoint, в done не попадает); + - run финализируется mark_done(partial) с unique_fetched==1; + - mark_cancelled / mark_failed / mark_banned не вызываются. + """ + state = {"flag": False} + monkeypatch.setattr(scrape_pipeline, "shutdown_requested", lambda: state["flag"]) + + saved_buckets: list[str] = [] + + def fake_save(_db: Any, lots: Any, **_kw: Any) -> tuple[int, int]: + saved_buckets.append("x") + return (len(lots), 0) + + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save) + + async def fake_fetch_all(self: AvitoScraper, **kwargs: Any) -> None: + on_bucket = kwargs["on_bucket"] + # Бакет №1 — сохраняется штатно. + on_bucket("b1", [_fake_lot("avito", "b1-1")]) + # SIGTERM приходит после первого бакета. + state["flag"] = True + # Бакет №2 — on_bucket должен поднять RuntimeError("shutdown"). + on_bucket("b2", [_fake_lot("avito", "b2-1")]) + # Бакет №3 — НЕ должен быть достигнут. + on_bucket("b3", [_fake_lot("avito", "b3-1")]) + + monkeypatch.setattr(AvitoScraper, "fetch_all_secondary", fake_fetch_all) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_avito_full_load( + db=MagicMock(), + run_id=3, + request_delay_sec=0.0, + ) + + # Только первый бакет сохранён (b2 упал на checkpoint до save, b3 не достигнут). + assert saved_buckets == ["x"], f"только b1 должен сохраниться; got {saved_buckets}" + assert counters.unique_fetched == 1 + assert counters.saved_inserted == 1 + # mark_done(partial) с done_buckets только b1. + assert fake.done is not None, "drain должен вызвать mark_done(partial)" + assert fake.done.get("done_buckets") == ["b1"] + assert fake.done["unique_fetched"] == 1 + # НЕ user-cancel, не failed/banned. + assert fake.cancelled_calls == 0 + assert fake.failed is None + assert fake.banned is None -- 2.45.3