Топология подтверждена перед удалением (docker-compose.prod.yml): tradein-backend (uvicorn app.main:app) — SCHEDULER_ENABLE=false; tradein-scraper (python -m app.scheduler_main) — SCHEDULER_ENABLE=true + USE_KIT_SCHEDULER=true. Kit-путь (_run_kit_scheduler → scraper_kit.orchestration.scheduler + product_handlers) самодостаточен: не импортирует ничего из app.services.scheduler.scheduler_loop или app.services.scrape_pipeline. Все НЕ-sweep джобы, которые kit-scheduler диспетчерит через build_product_handlers, идут напрямую в app.tasks.*/ app.services.* (либо lazy-импортят import_rosreestr_dkp/_execute_cian_backfill из scheduler.py) — мимо удаляемой legacy-машинерии. app/services/scheduler.py: 2098 → 418 строк. Удалено: scheduler_loop, get_due_schedules, reap_zombies, _claim_run, _defer_next_run_at, _spawn_tracked/ _drain_inflight/_inflight_tasks, все 27 trigger_*_run-функций, импорт app.services.scrape_pipeline, константы SCHEDULER_TICK_SEC/ZOMBIE_THRESHOLD_HOURS (достижимы были только через удалённый scheduler_loop-путь). Оставлено (живые импортёры вне удалённого): compute_next_run_at + has_running_run (admin.py), import_rosreestr_dkp + _execute_cian_backfill (lazy-импорты в product_handlers.py — job-тела kit-handler'ов). main.py: убран `from app.services.scheduler import scheduler_loop` + lifespan-блок запуска (`if settings.scheduler_enable: asyncio.create_task(scheduler_loop())`); прод-backend всегда шёл с SCHEDULER_ENABLE=false, так что это был мёртвый код. scheduler_main.py: убрана ship-dark развилка #2192 (USE_KIT_SCHEDULER=false → legacy scheduler_loop fallback) — _run_kit_scheduler() теперь безусловный путь. Поле settings.use_kit_scheduler оставлено в конфиге (Settings extra="ignore" защищает от startup-краха на leftover env var), но на ветвление не влияет. app.services.scrape_pipeline: 0 runtime-импортёров в app/+scripts/+packages/ после этого PR (только тесты, которые Part E удалит вместе с самим файлом) — подтверждено grep. scrape_pipeline.py не тронут (Part E). Тесты: удалены test_house_imv_backfill_scheduler.py (100% legacy-триггер, backfill_house_imv сервис покрыт в test_house_imv_backfill_browser_flag.py / test_backfill_wave2.py) и test_kit_registry_completeness.py (parity-инвариант против удалённого dispatch, дублирует test_scraper_kit_scheduler_parity.py). Точечно вырезаны "Scheduler wiring" секции (trigger_fn_exists/dispatch_branch_ wired/runs_in_executor) из ~10 файлов, тестирующих сами task-функции — сами task-тесты (SQL-shape, миграции, fake-db поведение) оставлены нетронутыми. test_scheduler.py: 825 → ~90 строк (остались только compute_next_run_at-тесты). test_scraper_kit_scheduler_parity.py: убрана golden-parity секция против удалённого scheduler_loop (SOURCE_TO_OLD_TRIGGER/_drive_old_one_tick/ test_routing_parity_per_source), остальное (claim/reap_zombies/dispatch/ registry-shape тесты kit-модуля) сохранено — источник этих инвариантов не app.services.scheduler, а сам scraper_kit.orchestration.scheduler. test_scheduler_main.py: 2 теста, патчившие app.services.scheduler.scheduler_loop, переведены на монкипатч sm._run_kit_scheduler (единственный путь после этого PR). test_sweep_imv_phase.py:171-371 (6 прямых импортов run_avito_city_sweep из scrape_pipeline) намеренно НЕ тронуты — Part E. Verify: полный pytest 3179 passed / 6 skipped / 1 known-unrelated fail (test_search_cache_hit, #2208, не связан с этим PR); ruff 0.7.4 чист на всех изменённых файлах; `python -c "import app.main; import app.scheduler_main"` OK.
230 lines
8.5 KiB
Python
230 lines
8.5 KiB
Python
"""Freshness-монитор данных deals по max(deal_date) — #2212.
|
||
|
||
Покрывает:
|
||
1. Чистую логику evaluate_deals_freshness (frozen now, без БД):
|
||
- нет алерта: 2026Q1 @ 2026-07-02 (порог 2026-08-15 не наступил);
|
||
- алерт: 2026Q1 @ 2026-08-20 (days_overdue=5);
|
||
- граница порога (== threshold → нет алерта; +1 день → алерт);
|
||
- кастомный lag_allowance_days.
|
||
2. check_deals_freshness с FakeDB (no-alert / alert / empty→mark_failed).
|
||
3. Свойства миграции 162 (по образцу test_deactivate_stale_listings).
|
||
4. Регистрацию в kit product_handlers (registry остаётся зелёным).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
from datetime import UTC, date, datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from app.services.product_handlers import build_product_handlers
|
||
from app.tasks import deals_freshness_monitor as mon
|
||
|
||
_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
|
||
_MIGRATION_162 = _SQL_DIR / "162_seed_deals_freshness_monitor.sql"
|
||
|
||
# max(deal_date) = 2026Q1 label (проверено на проде 2026-07-02).
|
||
_Q1_2026 = date(2026, 1, 1)
|
||
|
||
|
||
# ── evaluate_deals_freshness (чистая логика, frozen now) ──────────────────────
|
||
|
||
|
||
def _now(y: int, m: int, d: int) -> datetime:
|
||
return datetime(y, m, d, tzinfo=UTC)
|
||
|
||
|
||
def test_no_alert_now_2026_07_02() -> None:
|
||
"""Сейчас: 2026Q1 @ 2026-07-02 — порог просрочки 2026-08-15 не наступил."""
|
||
v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 7, 2))
|
||
assert v.overdue is False
|
||
assert v.days_overdue == 0
|
||
assert v.latest_q_end == date(2026, 4, 1)
|
||
assert v.overdue_threshold == date(2026, 8, 15)
|
||
|
||
|
||
def test_alert_2026_08_20() -> None:
|
||
"""К 2026-08-20 Q2 не импортирован — алерт, просрочка 5 дней (порог 2026-08-15)."""
|
||
v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 8, 20))
|
||
assert v.overdue is True
|
||
assert v.days_overdue == 5
|
||
|
||
|
||
def test_threshold_boundary_exact_no_alert() -> None:
|
||
"""Ровно в день порога (2026-08-15) алерта ещё нет (строгое >)."""
|
||
v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 8, 15))
|
||
assert v.overdue is False
|
||
assert v.days_overdue == 0
|
||
|
||
|
||
def test_threshold_boundary_next_day_alert() -> None:
|
||
"""Порог + 1 день (2026-08-16) — первый алерт, просрочка 1 день."""
|
||
v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 8, 16))
|
||
assert v.overdue is True
|
||
assert v.days_overdue == 1
|
||
|
||
|
||
def test_custom_lag_allowance_shifts_threshold() -> None:
|
||
"""lag_allowance_days=0 сдвигает порог на конец следующего квартала (2026-07-01)."""
|
||
v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 7, 2), lag_allowance_days=0)
|
||
assert v.overdue_threshold == date(2026, 7, 1)
|
||
assert v.overdue is True
|
||
assert v.days_overdue == 1
|
||
|
||
|
||
def test_add_months_quarter_starts() -> None:
|
||
assert mon._add_months(date(2026, 1, 1), 3) == date(2026, 4, 1)
|
||
assert mon._add_months(date(2026, 1, 1), 6) == date(2026, 7, 1)
|
||
assert mon._add_months(date(2026, 10, 1), 3) == date(2027, 1, 1)
|
||
|
||
|
||
def test_add_months_end_of_month_clamp() -> None:
|
||
# 31 янв + 1 мес → 28 фев (зажим на конец месяца).
|
||
assert mon._add_months(date(2026, 1, 31), 1) == date(2026, 2, 28)
|
||
|
||
|
||
# ── check_deals_freshness (FakeDB) ────────────────────────────────────────────
|
||
|
||
|
||
class _Row:
|
||
def __init__(self, latest: date | None) -> None:
|
||
self.latest = latest
|
||
|
||
|
||
class _FakeResult:
|
||
def __init__(self, latest: date | None) -> None:
|
||
self._latest = latest
|
||
|
||
def first(self) -> _Row:
|
||
return _Row(self._latest)
|
||
|
||
|
||
class _FakeDB:
|
||
def __init__(self, latest: date | None) -> None:
|
||
self._latest = latest
|
||
self.rolled_back = False
|
||
|
||
def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
|
||
return _FakeResult(self._latest)
|
||
|
||
def rollback(self) -> None:
|
||
self.rolled_back = True
|
||
|
||
|
||
def _patch_runs(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
||
calls: dict[str, Any] = {"done": None, "failed": None, "heartbeat": 0}
|
||
monkeypatch.setattr(
|
||
mon.runs_mod,
|
||
"update_heartbeat",
|
||
lambda *a, **k: calls.__setitem__("heartbeat", calls["heartbeat"] + 1),
|
||
)
|
||
monkeypatch.setattr(
|
||
mon.runs_mod,
|
||
"mark_done",
|
||
lambda _db, run_id, counters: calls.__setitem__("done", dict(counters)),
|
||
)
|
||
monkeypatch.setattr(
|
||
mon.runs_mod,
|
||
"mark_failed",
|
||
lambda _db, run_id, err, counters: calls.__setitem__("failed", err),
|
||
)
|
||
return calls
|
||
|
||
|
||
def test_check_no_alert_marks_done(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
calls = _patch_runs(monkeypatch)
|
||
db = _FakeDB(_Q1_2026)
|
||
out = mon.check_deals_freshness(db, run_id=1, params={}, now=_now(2026, 7, 2)) # type: ignore[arg-type]
|
||
assert out == {"latest_year": 2026, "latest_quarter": 1, "days_overdue": 0, "alert": 0}
|
||
assert calls["done"] == out
|
||
assert calls["failed"] is None
|
||
|
||
|
||
def test_check_overdue_marks_done_with_alert(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
calls = _patch_runs(monkeypatch)
|
||
db = _FakeDB(_Q1_2026)
|
||
out = mon.check_deals_freshness(db, run_id=2, params={}, now=_now(2026, 8, 20)) # type: ignore[arg-type]
|
||
assert out["alert"] == 1
|
||
assert out["days_overdue"] == 5
|
||
# Монитор НЕ падает при алерте — прогон done, а не failed.
|
||
assert calls["done"] == out
|
||
assert calls["failed"] is None
|
||
|
||
|
||
def test_check_empty_deals_marks_failed(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
calls = _patch_runs(monkeypatch)
|
||
db = _FakeDB(None)
|
||
out = mon.check_deals_freshness(db, run_id=3, params={}, now=_now(2026, 7, 2)) # type: ignore[arg-type]
|
||
assert out["alert"] == 0
|
||
assert calls["done"] is None
|
||
assert calls["failed"] is not None
|
||
|
||
|
||
def test_check_reads_lag_from_params(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
_patch_runs(monkeypatch)
|
||
db = _FakeDB(_Q1_2026)
|
||
# lag=0 → порог 2026-07-01 → @2026-07-02 уже просрочено.
|
||
out = mon.check_deals_freshness(
|
||
db, run_id=4, params={"lag_allowance_days": 0}, now=_now(2026, 7, 2)
|
||
) # type: ignore[arg-type]
|
||
assert out["alert"] == 1
|
||
|
||
|
||
# ── Миграция 162 ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_migration_162_exists() -> None:
|
||
assert _MIGRATION_162.is_file(), f"missing migration: {_MIGRATION_162}"
|
||
|
||
|
||
def test_migration_162_seeds_source() -> None:
|
||
sql = _MIGRATION_162.read_text("utf-8")
|
||
assert "'deals_freshness_monitor'" in sql
|
||
|
||
|
||
def test_migration_162_enabled_true() -> None:
|
||
sql = _MIGRATION_162.read_text("utf-8")
|
||
assert "true" in sql
|
||
|
||
|
||
def test_migration_162_is_idempotent() -> None:
|
||
sql = _MIGRATION_162.read_text("utf-8")
|
||
assert "ON CONFLICT (source) DO NOTHING" in sql
|
||
|
||
|
||
def test_migration_162_is_transactional() -> None:
|
||
sql = _MIGRATION_162.read_text("utf-8")
|
||
assert "BEGIN;" in sql
|
||
assert "COMMIT;" in sql
|
||
|
||
|
||
def test_migration_162_window_8_to_9_utc() -> None:
|
||
sql = _MIGRATION_162.read_text("utf-8")
|
||
assert re.search(r"\b8\b", sql), "window_start_hour 8 missing"
|
||
assert re.search(r"\b9\b", sql), "window_end_hour 9 missing"
|
||
|
||
|
||
def test_migration_162_lag_allowance_45() -> None:
|
||
sql = _MIGRATION_162.read_text("utf-8")
|
||
assert "lag_allowance_days" in sql
|
||
assert "45" in sql
|
||
|
||
|
||
def test_migration_162_no_psycopg_trap() -> None:
|
||
sql = _MIGRATION_162.read_text("utf-8")
|
||
assert not re.search(r":\w+::", sql)
|
||
|
||
|
||
# ── Регистрация в kit registry ─────────────────────────────────────────────────
|
||
|
||
|
||
def test_kit_product_handler_registered() -> None:
|
||
# ctx не нужен для сборки dict ключей — build_product_handlers его не замыкает.
|
||
handlers = build_product_handlers(ctx=None) # type: ignore[arg-type]
|
||
assert "deals_freshness_monitor" in handlers
|