diff --git a/tradein-mvp/backend/app/services/product_handlers.py b/tradein-mvp/backend/app/services/product_handlers.py index 2810bc2e..a52d0146 100644 --- a/tradein-mvp/backend/app/services/product_handlers.py +++ b/tradein-mvp/backend/app/services/product_handlers.py @@ -183,6 +183,16 @@ async def _job_rosreestr_quarter_poll( await run_rosreestr_quarter_poll(db, run_id=run_id, params=params) +# ── deals_freshness_monitor — sync DB-only freshness check в executor ───────── +async def _job_deals_freshness_monitor( + db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext +) -> None: + from app.tasks.deals_freshness_monitor import check_deals_freshness + + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, check_deals_freshness, db, run_id, params) + + # ── newbuilding_enrich — async, owns lifecycle ─────────────────────────────── async def _job_newbuilding_enrich( db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext @@ -348,6 +358,7 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]: "yandex_address_backfill": Handler(_job_yandex_address_backfill, "yandex_address_backfill"), "sber_index_pull": Handler(_job_sber_index_pull, "sber_index_pull"), "rosreestr_quarter_poll": Handler(_job_rosreestr_quarter_poll, "rosreestr_quarter_poll"), + "deals_freshness_monitor": Handler(_job_deals_freshness_monitor, "deals_freshness_monitor"), "newbuilding_enrich": Handler(_job_newbuilding_enrich, "newbuilding_enrich"), "yandex_newbuilding_sweep": Handler( _job_yandex_newbuilding_sweep, "yandex_newbuilding_sweep" diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index c233e498..03905a6a 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -1165,6 +1165,44 @@ async def trigger_rosreestr_quarter_poll_run( return run_id +async def trigger_deals_freshness_monitor_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch check_deals_freshness в executor (sync DB-only task). + + Мониторинг свежести ДАННЫХ deals (#2212): смотрит max(deal_date) и поднимает + ERROR-алерт (→ GlitchTip), когда следующий квартал просрочен публикацией. Джоба- + статус rosreestr_* остаётся зелёным при staleness данных — этот монитор ловит + именно устаревание данных, а не сбой прогона. + + Задача синхронная (один SELECT max(deal_date), без внешних HTTP) — гоняем в + run_in_executor по образцу trigger_listing_source_snapshot_run. + + Returns run_id (или None если skip — есть running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.deals_freshness_monitor import check_deals_freshness + + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, check_deals_freshness, run_db, run_id, params) + except Exception: + logger.exception("scheduler: check_deals_freshness crashed run_id=%d", run_id) + finally: + run_db.close() + + _spawn_tracked(_run()) + logger.info("scheduler: triggered deals_freshness_monitor run_id=%d", run_id) + return run_id + + async def trigger_newbuilding_enrich_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch run_newbuilding_enrich в asyncio.create_task (#973). @@ -1903,6 +1941,8 @@ async def scheduler_loop() -> None: await trigger_sber_index_pull_run(db, sch) elif source == "rosreestr_quarter_poll": await trigger_rosreestr_quarter_poll_run(db, sch) + elif source == "deals_freshness_monitor": + await trigger_deals_freshness_monitor_run(db, sch) elif source == "newbuilding_enrich": await trigger_newbuilding_enrich_run(db, sch) elif source == "yandex_newbuilding_sweep": diff --git a/tradein-mvp/backend/app/tasks/deals_freshness_monitor.py b/tradein-mvp/backend/app/tasks/deals_freshness_monitor.py new file mode 100644 index 00000000..0a687913 --- /dev/null +++ b/tradein-mvp/backend/app/tasks/deals_freshness_monitor.py @@ -0,0 +1,196 @@ +"""Мониторинг свежести ДАННЫХ deals (не статуса джобы) — #2212. + +Проблема аудита: rosreestr_quarter_poll / rosreestr_dkp_import честно рапортуют +`done` (rows_fetched>0, inserted=0 == «всё уже импортировано»), но это МАСКИРУЕТ +staleness самих данных. `deals.deal_date` — лейбл начала квартала (2026-01-01 = 2026Q1); +Росреестр публикует квартальными дампами, скачивание дампа — РУЧНОЙ операторский шаг +(poll лишь алертит о доступности). Если оператор пропустил импорт нового квартала, +джоба-статус остаётся зелёным, а данные тихо устаревают. + +Этот монитор смотрит на `max(deal_date)` и поднимает ERROR-алерт (→ GlitchTip через +sentry logging integration), когда следующий квартал ПРОСРОЧЕН публикацией. + +Арифметика просрочки (проверена на проде 2026-07-02, max=2026Q1, q_end=2026-03-31): + latest_q_start = max(deal_date) (2026-01-01) + latest_q_end = latest_q_start + 3 months (2026-04-01, старт Q2) + overdue_threshold = latest_q_end + 3 months + LAG_DAYS (2026-07-01 + 45д = 2026-08-15) + overdue = now() > overdue_threshold +Смысл: следующий квартал (Q2) заканчивается через 3 мес после конца имеющегося; даём +LAG_ALLOWANCE_DAYS (по умолчанию 45) на публикацию дампа Росреестром. Сейчас (2026-07-02) +до порога 2026-08-15 ещё далеко → алерта НЕТ. Если к 2026-08-16 Q2 не импортирован → алерт. + +Задача синхронная (DB-only, один SELECT max(deal_date)) — запускается in-app scheduler'ом +через trigger_deals_freshness_monitor_run() в run_in_executor, по образцу +listing_source_snapshot / rosreestr_dkp_import. Вердикт вычисляет ЧИСТАЯ функция +evaluate_deals_freshness() (frozen-now тестируется без БД). + +Прогон НЕ помечается failed при алерте (это МОНИТОР, а не сбой джобы) — ERROR-лог +достаточен. mark_failed только если deals недоступна/пуста (нечего оценивать). +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import UTC, date, datetime, timedelta + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.services import scrape_runs as runs_mod + +logger = logging.getLogger(__name__) + +__all__ = [ + "DEFAULT_LAG_ALLOWANCE_DAYS", + "FreshnessVerdict", + "check_deals_freshness", + "evaluate_deals_freshness", +] + +# Лаг публикации дампа Росреестром: сколько дней после конца следующего квартала +# считаем допустимой задержкой до алерта. Читается из default_params.lag_allowance_days. +DEFAULT_LAG_ALLOWANCE_DAYS = 45 + +_LATEST_DEAL_DATE_SQL = text("SELECT max(deal_date) AS latest FROM deals") + + +@dataclass(frozen=True) +class FreshnessVerdict: + """Вердикт свежести deals по max(deal_date).""" + + latest_q_start: date + latest_q_end: date + overdue_threshold: date + overdue: bool + days_overdue: int + + +def _add_months(d: date, months: int) -> date: + """Прибавить `months` к дате. deal_date всегда старт квартала (день=1), но + хелпер устойчив к произвольному дню — при переполнении зажимает на конец месяца. + """ + total = (d.month - 1) + months + year = d.year + total // 12 + month = total % 12 + 1 + # Зажим дня на последний день целевого месяца (для дней 29-31; для day=1 no-op). + if month == 12: + next_month_first = date(year + 1, 1, 1) + else: + next_month_first = date(year, month + 1, 1) + last_day = (next_month_first - timedelta(days=1)).day + return date(year, month, min(d.day, last_day)) + + +def evaluate_deals_freshness( + latest_q_start: date, + now: datetime, + lag_allowance_days: int = DEFAULT_LAG_ALLOWANCE_DAYS, +) -> FreshnessVerdict: + """Чистая логика: просрочен ли следующий квартал относительно max(deal_date). + + overdue = now() > (latest_q_start + 6 месяцев + lag_allowance_days). + days_overdue = max(0, дни сверх порога) — 0 когда не просрочено. + + Тестируется с frozen `now` без БД. + """ + latest_q_end = _add_months(latest_q_start, 3) + overdue_threshold = _add_months(latest_q_start, 6) + timedelta(days=lag_allowance_days) + now_date = now.date() + overdue = now_date > overdue_threshold + days_overdue = max(0, (now_date - overdue_threshold).days) + return FreshnessVerdict( + latest_q_start=latest_q_start, + latest_q_end=latest_q_end, + overdue_threshold=overdue_threshold, + overdue=overdue, + days_overdue=days_overdue, + ) + + +def check_deals_freshness( + db: Session, + run_id: int, + params: dict | None = None, # type: ignore[type-arg] + now: datetime | None = None, +) -> dict[str, int]: + """Проверить свежесть deals по max(deal_date) и поднять алерт при просрочке. + + Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources). + Читает один SELECT max(deal_date), считает вердикт чистой функцией, логирует + ERROR при overdue (→ GlitchTip) и финализирует run. + + Params (default_params jsonb): + lag_allowance_days: int — допустимый лаг публикации дампа (default 45). + `now` инъектируется в тестах (frozen); в проде — None → datetime.now(UTC). + + Returns counters {latest_year, latest_quarter, days_overdue, alert}. + mark_failed только если deals пуста/недоступна (нечего оценивать); при алерте + прогон помечается done (это монитор, не сбой джобы). + """ + params = params or {} + now = now or datetime.now(UTC) + counters: dict[str, int] = { + "latest_year": 0, + "latest_quarter": 0, + "days_overdue": 0, + "alert": 0, + } + try: + runs_mod.update_heartbeat(db, run_id, counters) + + row = db.execute(_LATEST_DEAL_DATE_SQL).first() + latest: date | None = row.latest if row is not None else None + if latest is None: + logger.warning( + "deals freshness: таблица deals пуста/недоступна — оценить свежесть нельзя" + ) + runs_mod.mark_failed(db, run_id, "deals empty or unavailable", counters) + return counters + + lag_days = int(params.get("lag_allowance_days", DEFAULT_LAG_ALLOWANCE_DAYS)) + verdict = evaluate_deals_freshness(latest, now, lag_days) + + quarter = (latest.month - 1) // 3 + 1 + counters = { + "latest_year": latest.year, + "latest_quarter": quarter, + "days_overdue": verdict.days_overdue, + "alert": int(verdict.overdue), + } + + if verdict.overdue: + logger.error( + "deals freshness: max(deal_date)=%s — новый квартал просрочен на %d дней; " + "проверь rosreestr_quarter_poll + операторский импорт " + "(data/sql/02_load_all_quarters.sh)", + latest, + verdict.days_overdue, + ) + else: + logger.info( + "deals freshness: max(deal_date)=%s (Q%d %d) свежий — порог просрочки %s " + "ещё не наступил, алерта нет", + latest, + quarter, + latest.year, + verdict.overdue_threshold, + ) + + runs_mod.mark_done(db, run_id, counters) + logger.info( + "check_deals_freshness run_id=%d done: latest=%s alert=%d days_overdue=%d", + run_id, + latest, + counters["alert"], + counters["days_overdue"], + ) + return counters + except Exception as exc: + logger.exception("check_deals_freshness run_id=%d failed", run_id) + try: + db.rollback() + except Exception: + pass + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/data/sql/162_seed_deals_freshness_monitor.sql b/tradein-mvp/backend/data/sql/162_seed_deals_freshness_monitor.sql new file mode 100644 index 00000000..e5a93c38 --- /dev/null +++ b/tradein-mvp/backend/data/sql/162_seed_deals_freshness_monitor.sql @@ -0,0 +1,75 @@ +-- 162_seed_deals_freshness_monitor.sql +-- Issue #2212 — seed scrape_schedules row for the daily deals-data freshness monitor. +-- +-- Context (verified on prod 2026-07-02): +-- deals.deal_date is the LABEL of a quarter's start (2026-01-01 == 2026Q1); Rosreestr +-- publishes quarterly dumps. Current max(deal_date) = 2026Q1 — the LATEST published +-- quarter, so the import is NOT broken (rows_fetched 49791 / inserted 0 == already +-- fully imported). Downloading a new dump is a MANUAL operator step; rosreestr_quarter_poll +-- only detects availability and alerts — it does not ingest. +-- +-- Audit gap: the `done` status of rosreestr_quarter_poll / rosreestr_dkp_import MASKS +-- staleness of the DATA. A green job-status can sit on top of quarter-stale deals if +-- the operator missed the manual import. This monitor watches max(deal_date) directly +-- and raises an ERROR alert (→ GlitchTip via sentry logging integration) when the next +-- quarter is overdue for publication. +-- +-- Overdue arithmetic (app/tasks/deals_freshness_monitor.py::evaluate_deals_freshness): +-- latest_q_start = max(deal_date) (2026-01-01) +-- latest_q_end = latest_q_start + 3 months (2026-04-01, start of Q2) +-- overdue_threshold = latest_q_end + 3 months + lag_days (2026-07-01 + 45d = 2026-08-15) +-- overdue = now() > overdue_threshold +-- At 2026-07-02 the threshold (2026-08-15) is not reached -> NO alert. If Q2 is still +-- not imported by 2026-08-16, the monitor starts alerting daily. +-- +-- Components deployed together: +-- 1. app/tasks/deals_freshness_monitor.py — check_deals_freshness() + evaluate_deals_freshness() +-- 2. app/services/scheduler.py — trigger_deals_freshness_monitor_run() + dispatch branch +-- `elif source == "deals_freshness_monitor"` +-- 3. app/services/product_handlers.py — kit Handler 'deals_freshness_monitor' +-- 4. This migration — seeds the scrape_schedules row (enabled=true, daily window) +-- +-- Schedule window 08:00-09:00 UTC: +-- Daily cadence, placed AFTER rosreestr_dkp_import (04:16 UTC window) so the monitor +-- observes the freshest possible deals state each morning. Cheap: one SELECT max(deal_date), +-- no external HTTP — safe to run daily. +-- +-- default_params.lag_allowance_days = 45: +-- Allowance for Rosreestr to publish the next quarter's dump after that quarter ends. +-- +-- next_run_at bootstrapped to tomorrow 08:00 UTC — scheduler will not fire immediately +-- on deploy (same pattern as 093/096/160). +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)) +-- 002_core_tables.sql (deals table with deal_date) +-- app/services/scheduler.py + product_handlers.py + tasks/deals_freshness_monitor.py deployed +-- +-- Deploy order: +-- Apply after deploying the scheduler.py + product_handlers.py + task changes so the +-- dispatch can resolve 'deals_freshness_monitor' correctly on first fire. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'deals_freshness_monitor', + true, -- SAFE: single SELECT max(deal_date), no ext calls + 8, -- window 08:00-09:00 UTC (after rosreestr_dkp_import) + 9, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 8)) AT TIME ZONE 'UTC', + '{"lag_allowance_days": 45}'::jsonb -- publication-lag allowance before alerting +) +ON CONFLICT (source) DO NOTHING; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_deals_freshness_monitor.py b/tradein-mvp/backend/tests/test_deals_freshness_monitor.py new file mode 100644 index 00000000..f725e282 --- /dev/null +++ b/tradein-mvp/backend/tests/test_deals_freshness_monitor.py @@ -0,0 +1,251 @@ +"""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. Регистрацию в scheduler-dispatch + kit product_handlers (registry остаётся зелёным). +""" + +from __future__ import annotations + +import inspect +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 import scheduler +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) + + +# ── Регистрация в scheduler + kit registry ──────────────────────────────────── + + +def test_trigger_exists_and_is_async() -> None: + fn = getattr(scheduler, "trigger_deals_freshness_monitor_run", None) + assert fn is not None, "trigger_deals_freshness_monitor_run missing from scheduler" + assert inspect.iscoroutinefunction(fn) + + +def test_dispatch_branch_present() -> None: + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "deals_freshness_monitor"' in loop_src + assert "trigger_deals_freshness_monitor_run(db, sch)" in loop_src + + +def test_trigger_runs_sync_task_in_executor() -> None: + trig_src = inspect.getsource(scheduler.trigger_deals_freshness_monitor_run) + assert "run_in_executor(" in trig_src + assert "check_deals_freshness" in trig_src + assert "_claim_run(db, schedule_row)" in trig_src + + +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