"""#2670 (остаток): лестница напоминаний не отвечает на вопрос «что сломано сейчас». #2720 вылечил «алерт ровно один раз за серию»: теперь вехи 3, 6, 12, 24, 48… Но лестница шагает по ПОДРЯД ИДУЩИМ завершённым failed/banned прогонам, а на проде 2026-08-10 три самых залежавшихся источника из шести просроченных ей недоступны — и лишь один из трёх из-за редких вех: ┌────────────────────────────┬────────┬───────┬──────────────────────────────────────┐ │ источник │ стрик │ сут. │ когда напомнит лестница │ ├────────────────────────────┼────────┼───────┼──────────────────────────────────────┤ │ avito_full_load_exhaustive │ 0 │ 49.5 │ никогда: 5 банов обнулил 'cancelled' │ │ cian_history_backfill │ 0 │ 42.1 │ никогда: прогонов нет с 30.06 │ │ avito_full_load │ 31 │ 37.7 │ веха 48 → +17 прогонов × 7 сут = 119 │ │ avito_detail_backfill │ 5 │ 5.2 │ веха 6 → завтра │ │ domclick_city_sweep │ 5 │ 5.1 │ веха 6 → завтра │ │ domclick_detail_backfill │ 4 │ 5.0 │ веха 6 → послезавтра │ └────────────────────────────┴────────┴───────┴──────────────────────────────────────┘ Уплотнение вех (3,4,5,6…) чинит ТОЛЬКО третью строку: у первых двух стрик равен нулю, уплотнять нечего — «замолчал» там означает «перестал производить прогоны», а не «серия длиннее последней вехи». Поэтому остаток задачи закрывает сводка, считающая КАЛЕНДАРНЫЙ возраст последнего успеха, а не длину серии. Фальсификация: на коде до этой правки `emit_stale_digest`/`stale_sources` не существует (ImportError на сборе тестов) — сводки нет ни в каком виде. Тест `test_ladder_is_silent_for_the_worst_two` — КОНТРОЛЬ: он зелёный и до, и после правки и показывает ровно то, чего сводка не заменяет, а добавляет: лестница на этих двух молчит. """ from __future__ import annotations import os from datetime import UTC, datetime, timedelta from types import SimpleNamespace from typing import Any from unittest.mock import MagicMock, patch import pytest os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") from scraper_kit.orchestration import runs as kit_runs from scraper_kit.orchestration import scheduler as sched NOW = datetime(2026, 8, 10, 8, 0, tzinfo=UTC) def _row(source: str, interval_days: Any, age_days: float, never_ok: bool = False) -> Any: """Строка `_STALE_SOURCES_SQL`: last_ok уже схлопнут в `since` через COALESCE.""" return SimpleNamespace( source=source, interval_days=interval_days, since=NOW - timedelta(days=age_days), never_ok=never_ok, ) # Снимок прода 2026-08-10 08:00 UTC: все 52 включённых расписания не влезают, взяты все # просроченные + четыре контрольных, каждое из которых мимо порога по своей причине. PROD_ROWS = [ _row("cian_history_backfill", None, 42.1), # такт по умолчанию (daily) _row("avito_full_load_exhaustive", 7, 49.5), _row("avito_full_load", 7, 37.7), _row("avito_detail_backfill", None, 5.2), _row("domclick_city_sweep", None, 5.1), _row("domclick_detail_backfill", None, 5.0), # ── контроль: НЕ просрочены ── _row("rosreestr_quarter_poll", 28, 24.0), # 24 сут при такте 28 — норма _row("sber_index_pull", 7, 4.1), _row("avito_city_sweep", None, 1.1), _row("proxy_healthcheck", None, 0.02), ] # Порядок — по числу ПРОПУЩЕННЫХ ТАКТОВ (age/interval), а не по календарю: 42 суток # у суточного backfill'а = 42 пропущенных такта, 49.5 у недельного = 7. PROD_STALE = [ "cian_history_backfill", # 42.1 / 1 "avito_full_load_exhaustive", # 49.5 / 7 = 7.07 "avito_full_load", # 37.7 / 7 = 5.39 "avito_detail_backfill", # 5.2 / 1 "domclick_city_sweep", # 5.1 / 1 "domclick_detail_backfill", # 5.0 / 1 ] @pytest.fixture(autouse=True) def _reset_digest_clock() -> Any: """Выпуск сводки помнится в памяти модуля — сбрасываем между тестами.""" sched._last_stale_digest_at = None yield sched._last_stale_digest_at = None def _db(rows: list[Any]) -> MagicMock: db = MagicMock() db.execute.return_value.fetchall.return_value = rows return db # ── 1. Чистая логика порога ────────────────────────────────────────────────── def test_stale_sources_names_exactly_the_prod_six() -> None: """Шесть просроченных из десяти, порядок — по числу пропущенных ТАКТОВ, не суток.""" stale = sched.stale_sources(PROD_ROWS, NOW) assert [s.source for s in stale] == PROD_STALE def test_quarterly_source_is_not_stale_at_24_days() -> None: """Порог считается в тактах: 24 сут для 28-суточного poll'а — не просрочка.""" assert sched.stale_sources([_row("rosreestr_quarter_poll", 28, 24.0)], NOW) == [] # …а 85 суток (>3×28) — уже просрочка. assert [s.source for s in sched.stale_sources([_row("q", 28, 85.0)], NOW)] == ["q"] @pytest.mark.parametrize("raw", [None, "null", "", "abc", 0, -5]) def test_broken_interval_falls_back_to_daily(raw: Any) -> None: """`interval_days: null` и мусор → такт 1 сут, как у compute_next_run_at.""" assert sched._schedule_interval_days(raw) == 1 def test_never_successful_source_is_reported_with_a_flag() -> None: """Расписание без единого 'done' считается от created_at и помечается явно.""" (only,) = sched.stale_sources([_row("brand_new", 1, 9.0, never_ok=True)], NOW) assert only.never_ok is True # ── 2. Выпуск сводки ───────────────────────────────────────────────────────── def test_digest_emits_one_event_listing_all_stale_sources() -> None: sentry = MagicMock() with patch.object(sched, "sentry_sdk", sentry): stale = sched.emit_stale_digest(_db(PROD_ROWS), now=NOW) assert [s.source for s in stale] == PROD_STALE sentry.capture_message.assert_called_once() msg = sentry.capture_message.call_args[0][0] assert msg.startswith("6 scraper sources are stale") for name in PROD_STALE: assert name in msg assert "rosreestr_quarter_poll" not in msg def test_digest_covers_the_two_sources_the_ladder_cannot_reach() -> None: """Главное свойство: стрик 0 не мешает сводке — она меряет календарь, а не серию.""" sentry = MagicMock() with patch.object(sched, "sentry_sdk", sentry): stale = sched.emit_stale_digest(_db(PROD_ROWS), now=NOW) zero_streak = {"avito_full_load_exhaustive", "cian_history_backfill"} assert zero_streak <= {s.source for s in stale} def test_digest_is_quiet_when_everything_is_fresh() -> None: sentry = MagicMock() fresh = [_row("avito_city_sweep", None, 1.1), _row("sber_index_pull", 7, 4.1)] with patch.object(sched, "sentry_sdk", sentry): assert sched.emit_stale_digest(_db(fresh), now=NOW) == [] sentry.capture_message.assert_not_called() def test_digest_is_daily_not_per_tick() -> None: """Планировщик тикает раз в минуту; сводка обязана выходить раз в сутки.""" sentry = MagicMock() db = _db(PROD_ROWS) with patch.object(sched, "sentry_sdk", sentry): sched.emit_stale_digest(db, now=NOW) sched.emit_stale_digest(db, now=NOW + timedelta(minutes=1)) sched.emit_stale_digest(db, now=NOW + timedelta(hours=23)) assert sentry.capture_message.call_count == 1 sched.emit_stale_digest(db, now=NOW + timedelta(hours=24, minutes=1)) assert sentry.capture_message.call_count == 2 def test_digest_failure_never_breaks_the_tick() -> None: """Сводка — best-effort: упавший запрос не имеет права уронить тик планировщика.""" db = MagicMock() db.execute.side_effect = RuntimeError("db down") with patch.object(sched, "sentry_sdk", MagicMock()): assert sched.emit_stale_digest(db, now=NOW) == [] # ── 3. Контроль: что именно сводка ДОБАВЛЯЕТ к лестнице ────────────────────── @pytest.mark.parametrize( ("name", "streak"), [("avito_full_load_exhaustive", 0), ("cian_history_backfill", 0), ("avito_full_load", 31)], ) def test_ladder_is_silent_for_the_worst_two(name: str, streak: int) -> None: """КОНТРОЛЬ (зелёный и до правки): у трёх худших источников лестница молчит. Стрик 0 — прогонов нет / серию обнулил 'cancelled'; стрик 31 — между вехами 24 и 48. """ rows = [SimpleNamespace(status="banned") for _ in range(streak)] rows += [SimpleNamespace(status="done") for _ in range(3)] sentry = MagicMock() with patch.object(kit_runs, "sentry_sdk", sentry): kit_runs._alert_if_consecutive_failures(_db(rows), name) sentry.capture_message.assert_not_called()