gendesign/tradein-mvp/backend/tests/test_sber_freshness_monitor.py

"""Freshness-монитор данных СберИндекса по max(period_month) — audit п.1.

Покрывает:
  1. Чистую логику evaluate_sber_freshness (frozen now, без БД):
     - fresh:  age <= max_age_days (алерта нет);
     - stale:  age > max_age_days (алерт);
     - граница порога (== max_age_days → нет алерта; +1 день → алерт).
  2. check_sber_freshness с FakeDB (fresh / stale / empty→mark_failed / кастомный lag).
  3. Свойства миграции 180 (по образцу test_deals_freshness_monitor).
  4. Регистрацию в kit product_handlers (registry остаётся зелёным).
"""

from __future__ import annotations

import os
import re
from datetime import UTC, date, datetime, timedelta
from pathlib import Path
from typing import Any

import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.core.config import settings
from app.services.product_handlers import build_product_handlers
from app.tasks import sber_freshness_monitor as mon

_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
_MIGRATION_180 = _SQL_DIR / "180_seed_sber_freshness_monitor.sql"
_MIGRATION_212 = _SQL_DIR / "212_sber_index_pull_weekly.sql"

# max(period_month) вторичного сегмента = 2026-05-01 (проверено на проде 2026-07-12).
_MAY_2026 = date(2026, 5, 1)


# ── evaluate_sber_freshness (чистая логика, frozen now) ───────────────────────


def _now(y: int, m: int, d: int) -> datetime:
    return datetime(y, m, d, tzinfo=UTC)


def test_fresh_within_max_age() -> None:
    """age=30 ≤ max_age_days=60 — свежий, алерта нет."""
    v = mon.evaluate_sber_freshness(_MAY_2026, _now(2026, 5, 31), max_age_days=60)
    assert v.stale is False
    assert v.age_days == 30
    assert v.latest_period == _MAY_2026


def test_stale_beyond_max_age() -> None:
    """Прод-состояние: 2026-05-01 @ 2026-07-12 — age=72 > 60 → алерт."""
    v = mon.evaluate_sber_freshness(_MAY_2026, _now(2026, 7, 12), max_age_days=60)
    assert v.stale is True
    assert v.age_days == 72


def test_threshold_boundary_exact_no_alert() -> None:
    """Ровно на пороге (age == max_age_days) алерта ещё нет (строгое >)."""
    v = mon.evaluate_sber_freshness(_MAY_2026, _now(2026, 6, 30), max_age_days=60)
    assert v.age_days == 60
    assert v.stale is False


def test_threshold_boundary_next_day_alert() -> None:
    """Порог + 1 день (age=61) — первый алерт."""
    v = mon.evaluate_sber_freshness(_MAY_2026, _now(2026, 7, 1), max_age_days=60)
    assert v.age_days == 61
    assert v.stale is True


# ── check_sber_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_fresh_marks_done(monkeypatch: pytest.MonkeyPatch) -> None:
    calls = _patch_runs(monkeypatch)
    db = _FakeDB(_MAY_2026)
    # @2026-05-31: age=30 ≤ 35+25=60 → нет алерта.
    out = mon.check_sber_freshness(db, run_id=1, params={}, now=_now(2026, 5, 31))  # type: ignore[arg-type]
    assert out == {"latest_year": 2026, "latest_month": 5, "age_days": 30, "alert": 0}
    assert calls["done"] == out
    assert calls["failed"] is None


def test_check_stale_marks_done_with_alert(monkeypatch: pytest.MonkeyPatch) -> None:
    calls = _patch_runs(monkeypatch)
    db = _FakeDB(_MAY_2026)
    # @2026-07-12 (прод): age=72 > 60 → алерт.
    out = mon.check_sber_freshness(db, run_id=2, params={}, now=_now(2026, 7, 12))  # type: ignore[arg-type]
    assert out["alert"] == 1
    assert out["age_days"] == 72
    assert out["latest_month"] == 5
    # Монитор НЕ падает при алерте — прогон done, а не failed.
    assert calls["done"] == out
    assert calls["failed"] is None


def test_check_empty_index_marks_failed(monkeypatch: pytest.MonkeyPatch) -> None:
    calls = _patch_runs(monkeypatch)
    db = _FakeDB(None)
    out = mon.check_sber_freshness(db, run_id=3, params={}, now=_now(2026, 7, 12))  # 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(_MAY_2026)
    # lag=0 → порог = sber_index_max_age_days (35) → @2026-07-12 (age=72) просрочено.
    out = mon.check_sber_freshness(
        db, run_id=4, params={"lag_allowance_days": 0}, now=_now(2026, 7, 12)
    )  # type: ignore[arg-type]
    assert out["alert"] == 1


def test_check_default_threshold_uses_setting_plus_lag(monkeypatch: pytest.MonkeyPatch) -> None:
    """Дефолтный порог = sber_index_max_age_days + DEFAULT_LAG_ALLOWANCE_DAYS."""
    _patch_runs(monkeypatch)
    db = _FakeDB(_MAY_2026)
    threshold = settings.sber_index_max_age_days + mon.DEFAULT_LAG_ALLOWANCE_DAYS
    # Ровно на пороге (age == threshold) — алерта нет; +1 день — алерт.
    exact = datetime(2026, 5, 1, tzinfo=UTC) + timedelta(days=threshold)
    out_exact = mon.check_sber_freshness(db, run_id=5, params={}, now=exact)  # type: ignore[arg-type]
    assert out_exact["age_days"] == threshold
    assert out_exact["alert"] == 0
    out_over = mon.check_sber_freshness(db, run_id=6, params={}, now=exact + timedelta(days=1))  # type: ignore[arg-type]
    assert out_over["alert"] == 1


# ── Миграция 180 ──────────────────────────────────────────────────────────────


def test_migration_180_exists() -> None:
    assert _MIGRATION_180.is_file(), f"missing migration: {_MIGRATION_180}"


def test_migration_180_seeds_source() -> None:
    sql = _MIGRATION_180.read_text("utf-8")
    assert "'sber_freshness_monitor'" in sql


def test_migration_180_enabled_true() -> None:
    sql = _MIGRATION_180.read_text("utf-8")
    assert "true" in sql


def test_migration_180_is_idempotent() -> None:
    sql = _MIGRATION_180.read_text("utf-8")
    assert "ON CONFLICT (source) DO NOTHING" in sql


def test_migration_180_is_transactional() -> None:
    sql = _MIGRATION_180.read_text("utf-8")
    assert "BEGIN;" in sql
    assert "COMMIT;" in sql


def test_migration_180_window_9_to_10_utc() -> None:
    sql = _MIGRATION_180.read_text("utf-8")
    assert re.search(r"\b9\b", sql), "window_start_hour 9 missing"
    assert re.search(r"\b10\b", sql), "window_end_hour 10 missing"


def test_migration_180_lag_allowance_25() -> None:
    sql = _MIGRATION_180.read_text("utf-8")
    assert "lag_allowance_days" in sql
    assert "25" in sql


def test_migration_180_no_psycopg_trap() -> None:
    sql = _MIGRATION_180.read_text("utf-8")
    assert not re.search(r":\w+::", sql)


# ── Миграция 212: такт загрузки не должен пересекать порог монитора ───────────
#
# Прод-разбор (ревью PR #2681): загрузка раз в 28 дней давала возраст-пилу 46..74
# при пороге 60 — тревога срабатывала 14 суток из 28 БЕЗ всякого застоя источника.
# Тест держит инвариант: потолок возраста (пол + такт загрузки) < порога монитора.


def test_migration_212_makes_pull_cadence_weekly() -> None:
    sql = _MIGRATION_212.read_text("utf-8")
    assert "sber_index_pull" in sql
    assert '"interval_days": 7' in sql
    assert "BEGIN;" in sql and "COMMIT;" in sql
    assert not re.search(r":\w+::", sql)  # psycopg v3: только CAST(:x AS type)


def test_pull_cadence_leaves_margin_under_monitor_threshold() -> None:
    """Инвариант: пол возраста + такт загрузки < порога монитора.

    Пол = 46 суток (прод 2026-07-17: загрузка принесла 2026-06-01). Порог =
    sber_index_max_age_days + lag_allowance. При такте 7: 46+7=53 < 60 — запас
    7 суток. При прежних 28: 46+28=74 > 60 — тревога каждый цикл, что и наблюдали.
    """
    interval_days = int(
        re.search(r'"interval_days":\s*(\d+)', _MIGRATION_212.read_text("utf-8")).group(1)
    )
    observed_floor_days = 46
    threshold = settings.sber_index_max_age_days + mon.DEFAULT_LAG_ALLOWANCE_DAYS
    assert observed_floor_days + interval_days < threshold, (
        f"такт {interval_days}д даёт потолок возраста "
        f"{observed_floor_days + interval_days}д при пороге {threshold}д — "
        "монитор снова будет мерить наш такт, а не застой источника"
    )


# ── Регистрация в 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 "sber_freshness_monitor" in handlers