feat(tradein/monitor): алерт на staleness данных deals по max(deal_date) (#2212) #2226
5 changed files with 573 additions and 0 deletions
|
|
@ -183,6 +183,16 @@ async def _job_rosreestr_quarter_poll(
|
||||||
await run_rosreestr_quarter_poll(db, run_id=run_id, params=params)
|
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 ───────────────────────────────
|
# ── newbuilding_enrich — async, owns lifecycle ───────────────────────────────
|
||||||
async def _job_newbuilding_enrich(
|
async def _job_newbuilding_enrich(
|
||||||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
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"),
|
"yandex_address_backfill": Handler(_job_yandex_address_backfill, "yandex_address_backfill"),
|
||||||
"sber_index_pull": Handler(_job_sber_index_pull, "sber_index_pull"),
|
"sber_index_pull": Handler(_job_sber_index_pull, "sber_index_pull"),
|
||||||
"rosreestr_quarter_poll": Handler(_job_rosreestr_quarter_poll, "rosreestr_quarter_poll"),
|
"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"),
|
"newbuilding_enrich": Handler(_job_newbuilding_enrich, "newbuilding_enrich"),
|
||||||
"yandex_newbuilding_sweep": Handler(
|
"yandex_newbuilding_sweep": Handler(
|
||||||
_job_yandex_newbuilding_sweep, "yandex_newbuilding_sweep"
|
_job_yandex_newbuilding_sweep, "yandex_newbuilding_sweep"
|
||||||
|
|
|
||||||
|
|
@ -1165,6 +1165,44 @@ async def trigger_rosreestr_quarter_poll_run(
|
||||||
return run_id
|
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:
|
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).
|
"""Создать 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)
|
await trigger_sber_index_pull_run(db, sch)
|
||||||
elif source == "rosreestr_quarter_poll":
|
elif source == "rosreestr_quarter_poll":
|
||||||
await trigger_rosreestr_quarter_poll_run(db, sch)
|
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":
|
elif source == "newbuilding_enrich":
|
||||||
await trigger_newbuilding_enrich_run(db, sch)
|
await trigger_newbuilding_enrich_run(db, sch)
|
||||||
elif source == "yandex_newbuilding_sweep":
|
elif source == "yandex_newbuilding_sweep":
|
||||||
|
|
|
||||||
196
tradein-mvp/backend/app/tasks/deals_freshness_monitor.py
Normal file
196
tradein-mvp/backend/app/tasks/deals_freshness_monitor.py
Normal file
|
|
@ -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
|
||||||
|
|
@ -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;
|
||||||
251
tradein-mvp/backend/tests/test_deals_freshness_monitor.py
Normal file
251
tradein-mvp/backend/tests/test_deals_freshness_monitor.py
Normal file
|
|
@ -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
|
||||||
Loading…
Add table
Reference in a new issue