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

"""#3051: поправка СберИндекса стала регион-зависимой (и монитор — многорядным).

ЧТО БЫЛО НЕ ТАК. sber_price_index ключуется ТЕКСТОВОЙ колонкой city, а оценщик
подставлял в неё константу «Свердловская область» для ЛЮБОЙ сделки — включая
212 937 московских (region_code=77, все ДКП). Замер прода 2026-09-10:
средневзвешенный по 69 138 московским сделкам за 12 месяцев time-фактор равен
1.0313 по свердловскому ряду против 1.0917 по московскому — московский коридор
занижен на ~5.9%. Коридор не advisory: он участвует в clamp headline.

Ряды в тестах синтетические, но их ОТНОШЕНИЯ взяты из этого замера (1.0313 / 1.0917),
чтобы промах ряда был виден числом, а не только именем.
"""

from __future__ import annotations

import contextlib
import logging
import os
from datetime import UTC, date, datetime
from typing import Any
from unittest.mock import MagicMock

import pytest
from sqlalchemy.exc import OperationalError, PendingRollbackError

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

from app.services import estimator as est
from app.services import regions as regions_mod
from app.tasks import sber_freshness_monitor as mon

# Харнесс соседей: настоящий sentry-клиент с прод-настройками контейнера скрапера.
# Импортируется, а не копируется, — тревога должна проверяться ТЕМ ЖЕ способом, каким
# её проверяют все соседние алерты (#2674).
from tests.test_alerts_become_events import event_texts, glitchtip_events

_BASE_MONTH = date(2025, 7, 1)
_LATEST_MONTH = date(2026, 7, 1)

# index_value подобран так, чтобы factor = latest/base дал ровно замеренные числа.
_SERIES_ROWS: dict[str, list[dict[str, Any]]] = {
    "Свердловская область": [
        {"period_month": _BASE_MONTH, "index_value_rub_m2": 100_000.0},
        {"period_month": _LATEST_MONTH, "index_value_rub_m2": 103_130.0},  # ×1.0313
    ],
    "Москва": [
        {"period_month": _BASE_MONTH, "index_value_rub_m2": 100_000.0},
        {"period_month": _LATEST_MONTH, "index_value_rub_m2": 109_170.0},  # ×1.0917
    ],
    "Россия": [
        {"period_month": _BASE_MONTH, "index_value_rub_m2": 100_000.0},
        {"period_month": _LATEST_MONTH, "index_value_rub_m2": 106_000.0},  # ×1.06
    ],
}


# ── резолвер ────────────────────────────────────────────────────────────────


def test_resolver_maps_66_to_sverdlovsk() -> None:
    assert est.sber_region_series_name(66) == "Свердловская область"
    assert est.sber_region_series_name(66) == est.SBER_TIME_ADJUST_REGION


def test_resolver_maps_77_to_moscow() -> None:
    """КРАСНЫЙ НА main: там ряда для 77 нет вовсе, Москва получала свердловский."""
    assert est.sber_region_series_name(77) == "Москва"


def test_resolver_unknown_region_falls_back_to_russia_not_sverdlovsk() -> None:
    """Неизвестный регион не падает и НЕ получает молча чужой региональный ряд."""
    assert est.sber_region_series_name(99) == "Россия"
    assert est.sber_region_series_name(None) == "Россия"
    assert est.SBER_TIME_ADJUST_FALLBACK_REGION == "Россия"


def test_resolver_unknown_region_is_logged_not_silent(
    caplog: pytest.LogCaptureFixture,
) -> None:
    with caplog.at_level(logging.WARNING, logger=est.logger.name):
        est.sber_region_series_name(1)
    assert any("_SBER_REGION_SERIES" in r.getMessage() for r in caplog.records)


def test_series_names_agree_with_region_registry() -> None:
    """Карта заведена явно, но расходиться с реестром регионов она не должна."""
    for code, name in est._SBER_REGION_SERIES.items():
        assert regions_mod.REGIONS[code].name == name


def test_resolver_maps_50_to_moscow_oblast_not_russia() -> None:
    """Область получает СВОЙ ряд, а не общероссийский фолбэк.

    Загрузчик научился тянуть REF_AREA=50 (#3498), и 12.09.2026 ряд приехал на
    прод: real_estate_deals/«Вторичный», 116 месяцев 2017-01..2026-08 — та же
    глубина, что у Москвы и обл.66. Пока кода 50 в карте не было, коридор
    области считался по «России», то есть ошибался на всё расхождение области
    со страной. Тест красный на прежней карте ПО ЗНАЧЕНИЮ: там вернулась бы
    «Россия».
    """
    assert est.sber_region_series_name(50) == "Московская область"
    assert est.sber_region_series_name(50) != est.SBER_TIME_ADJUST_FALLBACK_REGION
    assert "Московская область" in est.SBER_REQUIRED_REGIONS


def test_monitored_regions_cover_every_series_the_estimator_can_read() -> None:
    assert set(est.SBER_MONITORED_REGIONS) == set(est._SBER_REGION_SERIES.values()) | {
        est.SBER_TIME_ADJUST_FALLBACK_REGION
    }
    assert est.SBER_TIME_ADJUST_REGION in est.SBER_REQUIRED_REGIONS
    assert "Москва" in est.SBER_REQUIRED_REGIONS


# ── коридор: какой ряд реально читается ─────────────────────────────────────


def _corridor_db(deal_rows: list[dict[str, Any]]) -> tuple[Any, list[str]]:
    """Session-двойник: sber-запрос отвечает ПО ЗАПРОШЕННОМУ region, deals — сделками."""
    asked: list[str] = []
    db = MagicMock()

    def _execute(query: Any, params: dict[str, Any] | None = None) -> MagicMock:
        params = params or {}
        result = MagicMock()
        sql = str(query)
        if "sber_price_index" in sql:
            asked.append(params["region"])
            result.mappings.return_value.all.return_value = _SERIES_ROWS.get(params["region"], [])
        elif "deal_city_price_bands" in sql and "d.address ILIKE" in sql:
            result.mappings.return_value.all.return_value = deal_rows
        else:
            result.mappings.return_value.all.return_value = []
        return result

    db.execute.side_effect = _execute
    return db, asked


_DEALS = [{"price_per_m2": 200_000, "deal_date": _BASE_MONTH}]


def _fetch(region_code: int | None = None) -> tuple[dict[str, Any] | None, list[str]]:
    db, asked = _corridor_db(_DEALS)
    kwargs: dict[str, Any] = {}
    if region_code is not None:
        kwargs["region_code"] = region_code
    raw = est._fetch_dkp_corridor(
        db, address="Москва, улица Тверская, 10", rooms=2, area=60.0, **kwargs
    )
    return raw, asked


def test_moscow_corridor_reads_the_moscow_series() -> None:
    """КРАСНЫЙ НА main: там спрашивался ряд «Свердловская область»."""
    raw, asked = _fetch(77)
    assert asked == ["Москва"]
    assert raw is not None
    # 200 000 × 1.0917
    assert abs(raw["median_ppm2"] - 218_340) <= 1


def test_sverdlovsk_corridor_unchanged() -> None:
    raw, asked = _fetch(66)
    assert asked == ["Свердловская область"]
    assert raw is not None
    # 200 000 × 1.0313 — ровно то же число, что давал main
    assert abs(raw["median_ppm2"] - 206_260) <= 1


def test_default_region_is_still_sverdlovsk_byte_identical() -> None:
    """Вызов без region_code (дефолт 66) обязан дать то же, что явный 66."""
    default_raw, default_asked = _fetch(None)
    explicit_raw, explicit_asked = _fetch(66)
    assert default_asked == explicit_asked == ["Свердловская область"]
    assert default_raw == explicit_raw


def test_moscow_corridor_is_higher_than_the_sverdlovsk_one() -> None:
    """Тот самый недобор ~5.9%: направление промаха, а не только имя ряда."""
    moscow, _ = _fetch(77)
    sverdlovsk, _ = _fetch(66)
    assert moscow is not None and sverdlovsk is not None
    ratio = moscow["median_ppm2"] / sverdlovsk["median_ppm2"]
    assert 1.055 < ratio < 1.065  # 1.0917 / 1.0313 = 1.0586


def test_unknown_region_corridor_uses_russia_series() -> None:
    raw, asked = _fetch(99)
    assert asked == ["Россия"]
    assert raw is not None
    assert abs(raw["median_ppm2"] - 212_000) <= 1  # 200 000 × 1.06


# ── монитор: следит за обоими рядами ────────────────────────────────────────

_NOW = datetime(2026, 8, 12, 9, 0, tzinfo=UTC)
_LAST_PULL = datetime(2026, 8, 6, 9, 0, tzinfo=UTC)  # 6 суток < порога 14
_JUN_2026 = date(2026, 6, 1)


class _Row:
    def __init__(self, **kw: Any) -> None:
        self.__dict__.update(kw)


class _Result:
    def __init__(self, row: _Row | None) -> None:
        self._row = row

    def first(self) -> _Row | None:
        return self._row


class _MonDB:
    """Отвечает latest-периодом ПО РЕГИОНУ (main спрашивал только один)."""

    def __init__(self, latest_by_region: dict[str, date]) -> None:
        self._latest = latest_by_region
        self.asked_regions: list[str] = []

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _Result:
        sql = str(stmt)
        params = params or {}
        if "scrape_runs" in sql:
            return _Result(_Row(last_pull=_LAST_PULL))
        if "scrape_schedules" in sql:
            return _Result(_Row(interval_days="7"))
        self.asked_regions.append(params["city"])
        latest = self._latest.get(params["city"]) if params["dash"] == "real_estate_deals" else None
        return _Result(_Row(latest=latest))

    def rollback(self) -> None:  # pragma: no cover — не используется
        pass


def _patch_runs(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
    calls: dict[str, Any] = {"done": None, "failed": None}
    monkeypatch.setattr(mon.runs_mod, "update_heartbeat", lambda *a, **k: None)
    monkeypatch.setattr(
        mon.runs_mod, "mark_done", lambda _db, _rid, c: calls.__setitem__("done", dict(c))
    )
    monkeypatch.setattr(
        mon.runs_mod, "mark_failed", lambda _db, _rid, err, _c: calls.__setitem__("failed", err)
    )
    return calls


@contextlib.contextmanager
def _capture_errors() -> Any:
    """Собрать ERROR-записи монитора (caplog не годится: логгер модульный)."""
    records: list[logging.LogRecord] = []

    class _H(logging.Handler):
        def emit(self, record: logging.LogRecord) -> None:
            records.append(record)

    handler = _H(level=logging.ERROR)
    mon.logger.addHandler(handler)
    try:
        yield records
    finally:
        mon.logger.removeHandler(handler)


def _all_present() -> _MonDB:
    return _MonDB({r: _JUN_2026 for r in est.SBER_MONITORED_REGIONS})


def test_monitor_asks_every_series_the_estimator_can_read(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """КРАСНЫЙ НА main: спрашивался ровно один ряд, московский никто не видел."""
    _patch_runs(monkeypatch)
    db = _all_present()
    mon.check_sber_freshness(db, run_id=1, params={}, now=_NOW)  # type: ignore[arg-type]
    assert set(db.asked_regions) >= set(est.SBER_MONITORED_REGIONS)


def test_monitor_verdict_for_sverdlovsk_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
    """Прежние ключи counters и прежние числа: вердикт идёт за тактом загрузки."""
    calls = _patch_runs(monkeypatch)
    out = mon.check_sber_freshness(_all_present(), run_id=2, params={}, now=_NOW)  # type: ignore[arg-type]
    assert (out["latest_year"], out["latest_month"]) == (2026, 6)
    assert out["age_days"] == 72
    assert out["pull_lag_days"] == 6
    assert out["max_pull_lag_days"] == 14
    assert out["alert"] == 0
    assert out["regions_missing"] == 0
    assert calls["failed"] is None


def _without_moscow() -> _MonDB:
    """Источник переименовал «Москва» → ряда с прежним именем больше нет.

    Остальные обязательные ряды на месте — иначе тесты ниже считали бы пропажу
    Москвы вместе с чужими пропажами. Строятся от карты эстиматора, чтобы новый
    регион в ней не ронял эти тесты по счётчику (ровно так они и покраснели,
    когда в карту приехала область).
    """
    present = {r: _JUN_2026 for r in est.SBER_MONITORED_REGIONS if r != "Москва"}
    return _MonDB(present)


def test_missing_moscow_does_not_suppress_the_sverdlovsk_verdict(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Пропажа чужого ряда НЕ имеет права отключать мониторинг 66.

    КРАСНЫЙ ДО ПРАВКИ: ранний выход с mark_failed срабатывал раньше расчёта, и
    свердловская свежесть не считалась вовсе (latest_*=0, age_days=0, alert=0),
    хотя ряд был на месте и мог быть протухшим.
    """
    calls = _patch_runs(monkeypatch)
    out = mon.check_sber_freshness(_without_moscow(), run_id=3, params={}, now=_NOW)  # type: ignore[arg-type]
    # прежние ключи и прежние числа по свердловскому ряду — как при всех рядах
    assert (out["latest_year"], out["latest_month"]) == (2026, 6)
    assert out["age_days"] == 72
    assert out["pull_lag_days"] == 6
    assert out["max_pull_lag_days"] == 14
    assert calls["done"] == out
    assert calls["failed"] is None


def test_missing_moscow_is_visible_not_silently_green(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Прогон наблюдаем: свой счётчик + ERROR, где назван КОНКРЕТНЫЙ пропавший ряд."""
    _patch_runs(monkeypatch)
    with _capture_errors() as records:
        out = mon.check_sber_freshness(_without_moscow(), run_id=3, params={}, now=_NOW)  # type: ignore[arg-type]
    assert out["regions_missing"] == 1
    assert out["alert_regions_missing"] == 1  # КРАСНЫЙ ДО ПРАВКИ: ключа не было вовсе
    assert out["alert"] == 0  # алерт о такте загрузки не перегружен пропажей ряда
    msg = " ".join(r.getMessage() for r in records)
    assert "Москва" in msg  # назван КОНКРЕТНЫЙ пропавший ряд
    assert "таблица НЕ пуста" in msg  # прежний текст врал про пустую таблицу


def test_missing_sverdlovsk_still_marks_failed_with_truthful_text(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Прежнее поведение ради которого ранний выход и заведён: считать нечего.

    Текст ошибки при этом называет ряд, а не врёт про пустую таблицу.
    """
    calls = _patch_runs(monkeypatch)
    db = _MonDB({"Москва": _JUN_2026, "Россия": _JUN_2026})
    out = mon.check_sber_freshness(db, run_id=6, params={}, now=_NOW)  # type: ignore[arg-type]
    assert calls["done"] is None
    assert calls["failed"] is not None
    assert "Свердловская область" in calls["failed"]
    # КРАСНЫЙ ДО ПРАВКИ: было "sber_price_index empty or unavailable" — таблица непуста
    assert "empty" not in calls["failed"]
    assert out["latest_year"] == 0
    assert out["alert"] == 0


def test_all_series_present_keeps_old_counters_and_adds_new(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Оба ряда на месте: прежние ключи те же, новые — нулевые."""
    calls = _patch_runs(monkeypatch)
    out = mon.check_sber_freshness(_all_present(), run_id=7, params={}, now=_NOW)  # type: ignore[arg-type]
    assert (out["latest_year"], out["latest_month"], out["age_days"]) == (2026, 6, 72)
    assert (out["pull_lag_days"], out["max_pull_lag_days"], out["alert"]) == (6, 14, 0)
    assert out["regions_missing"] == 0
    assert out["alert_regions_missing"] == 0
    assert out["age_days_max"] == 72
    assert calls["failed"] is None


def test_missing_fallback_series_does_not_fail_the_monitor(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """«Россия» — фолбэк: сегодня по нему не считается ни одна сделка → не сбой."""
    calls = _patch_runs(monkeypatch)
    db = _MonDB({r: _JUN_2026 for r in est.SBER_REQUIRED_REGIONS})
    out = mon.check_sber_freshness(db, run_id=4, params={}, now=_NOW)  # type: ignore[arg-type]
    assert calls["failed"] is None
    assert out["regions_missing"] == 1
    assert out["alert"] == 0


def test_lagging_region_shows_up_in_age_days_max(monkeypatch: pytest.MonkeyPatch) -> None:
    """Отставший ряд виден наблюдением, свердловский age_days при этом не сдвинут."""
    _patch_runs(monkeypatch)
    db = _MonDB(
        {
            "Свердловская область": _JUN_2026,
            "Москва": date(2026, 3, 1),
            "Россия": _JUN_2026,
        }
    )
    out = mon.check_sber_freshness(db, run_id=5, params={}, now=_NOW)  # type: ignore[arg-type]
    assert out["age_days"] == 72
    assert out["age_days_max"] == (_NOW.date() - date(2026, 3, 1)).days


# --- круг 3: изоляция чужих рядов по ИСКЛЮЧЕНИЮ, дедупликация, ранний выход -------


class _MonDBFlaky(_MonDB):
    """Соединение рвётся на ОДНОМ ряде, остальные запросы отвечают как обычно.

    Это типовой прод-случай: таймаут/обрыв на середине обхода, а не «БД мертва».
    """

    def __init__(self, latest_by_region: dict[str, date], broken_region: str) -> None:
        super().__init__(latest_by_region)
        self._broken = broken_region

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _Result:
        params = params or {}
        if params.get("city") == self._broken:
            raise TimeoutError(f"connection reset while reading {self._broken}")
        return super().execute(stmt, params)


def test_probe_failure_on_foreign_series_does_not_suppress_the_verdict(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """КРАСНЫЙ ДО ПРАВКИ: обход шёл под ОБЩИМ try.

    TimeoutError на московском ряде улетал во внешний except → mark_failed и
    повторный подъём, вердикт по Свердловской области снова не считался — тот же
    дефект, что чинили по ветке пустой выборки.
    """
    calls = _patch_runs(monkeypatch)
    db = _MonDBFlaky({r: _JUN_2026 for r in est.SBER_MONITORED_REGIONS}, "Москва")
    with _capture_errors() as records:
        out = mon.check_sber_freshness(db, run_id=8, params={}, now=_NOW)  # type: ignore[arg-type]
    # вердикт по свердловскому ряду посчитан и доехал до done-прогона
    assert (out["latest_year"], out["latest_month"], out["age_days"]) == (2026, 6, 72)
    assert (out["pull_lag_days"], out["max_pull_lag_days"], out["alert"]) == (6, 14, 0)
    assert calls["failed"] is None
    assert calls["done"] == out
    # и при этом сбой не проглочен: ряд ненаблюдён, назван в ERROR
    assert out["regions_missing"] == 1
    assert out["alert_regions_missing"] == 1
    assert "Москва" in " ".join(r.getMessage() for r in records)


def test_probe_failure_on_the_default_series_still_fails_the_run(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Обратная сторона: сбой на ряде РЕГИОНА ПО УМОЛЧАНИЮ прячется не должен."""
    calls = _patch_runs(monkeypatch)
    db = _MonDBFlaky(
        {r: _JUN_2026 for r in est.SBER_MONITORED_REGIONS}, est.SBER_TIME_ADJUST_REGION
    )
    with pytest.raises(TimeoutError):
        mon.check_sber_freshness(db, run_id=9, params={}, now=_NOW)  # type: ignore[arg-type]
    assert calls["done"] is None
    assert calls["failed"] is not None


def test_duplicate_series_name_does_not_pin_regions_missing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """КРАСНЫЙ ДО ПРАВКИ: regions_missing = len(КОРТЕЖА) - len(найденных).

    Заводим в карте оценщика ещё один регион, чей ряд совпал с фолбэчной «Россией»:
    кортеж длиннее на элемент, различных рядов столько же — счётчик залипал на 1
    при всех живых рядах, т.е. вечный «пропал ряд» без пропажи.
    """
    calls = _patch_runs(monkeypatch)
    required = (*est.SBER_REQUIRED_REGIONS, est.SBER_TIME_ADJUST_FALLBACK_REGION)
    monkeypatch.setattr(mon, "SBER_REQUIRED_REGIONS", required)
    monkeypatch.setattr(
        mon, "SBER_MONITORED_REGIONS", (*required, est.SBER_TIME_ADJUST_FALLBACK_REGION)
    )
    db = _MonDB({r: _JUN_2026 for r in required})
    out = mon.check_sber_freshness(db, run_id=10, params={}, now=_NOW)  # type: ignore[arg-type]
    assert out["regions_missing"] == 0
    assert out["alert_regions_missing"] == 0
    assert calls["failed"] is None


def test_early_exit_fills_the_new_counters(monkeypatch: pytest.MonkeyPatch) -> None:
    """КРАСНЫЙ ДО ПРАВКИ: в раннем выходе новые ключи отдавали нули.

    Пропажа ВСЕХ обязательных рядов была неотличима от пропажи одного
    свердловского, а найденный ряд не попадал в age_days_max вовсе. Счётчик
    сверяем с длиной карты, а не с литералом: новый регион в ней — не повод
    этому тесту краснеть.
    """
    calls = _patch_runs(monkeypatch)
    db = _MonDB({est.SBER_TIME_ADJUST_FALLBACK_REGION: _JUN_2026})
    out = mon.check_sber_freshness(db, run_id=11, params={}, now=_NOW)  # type: ignore[arg-type]
    assert calls["done"] is None
    assert calls["failed"] is not None
    assert out["regions_missing"] == len(est.SBER_REQUIRED_REGIONS)
    assert out["alert_regions_missing"] == 1  # ДО ПРАВКИ: 0
    assert out["age_days_max"] == 72  # ДО ПРАВКИ: 0 — найденный ряд был не виден
    # прежняя семантика раннего выхода не тронута
    assert (out["latest_year"], out["latest_month"], out["age_days"], out["alert"]) == (0, 0, 0, 0)


# --- круг 4: откат сессии, ERROR до раннего выхода, настоящий канал тревоги -------


class _MonDBDriverFault(_MonDB):
    """Сессия ведёт себя как НАСТОЯЩАЯ: ошибка драйвера деактивирует транзакцию.

    Ровно семантика SQLAlchemy Session — после сбоя инструкции любой следующий запрос
    падает с PendingRollbackError, даже не дойдя до БД, пока кто-нибудь не позовёт
    rollback(). _MonDBFlaky этого не показывает (там сбоит ровно спрошенный ряд),
    поэтому зелёный тест на нём ничего не доказывал про испорченную сессию.
    """

    def __init__(self, latest_by_region: dict[str, date], broken_region: str) -> None:
        super().__init__(latest_by_region)
        self._broken = broken_region
        self._deactivated = False
        self.rollbacks = 0

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _Result:
        if self._deactivated:
            raise PendingRollbackError("Can't reconnect until invalid transaction is rolled back")
        params = params or {}
        if params.get("city") == self._broken:
            self._deactivated = True
            raise OperationalError(
                "SELECT max(period_month) FROM sber_price_index",
                {},
                TimeoutError("canceling statement due to statement timeout"),
            )
        return super().execute(stmt, params)

    def rollback(self) -> None:
        self.rollbacks += 1
        self._deactivated = False


def test_driver_error_on_foreign_series_rolls_the_session_back(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """КРАСНЫЙ ДО ПРАВКИ: в per-region except не было ОТКАТА.

    Прошлый круг изолировал распространение исключения, но не порчу сессии. Свердловский
    ряд прочитан, на московском таймаут, цикл идёт дальше — и первый же запрос ЗА
    ИНТЕРВАЛОМ ЗАГРУЗКИ падает с PendingRollbackError на деактивированной сессии,
    улетает во внешний except, даёт mark_failed и подъём: вердикт по Свердловской
    области снова не доезжает.
    """
    calls = _patch_runs(monkeypatch)
    db = _MonDBDriverFault({r: _JUN_2026 for r in est.SBER_MONITORED_REGIONS}, "Москва")
    with _capture_errors() as records:
        out = mon.check_sber_freshness(db, run_id=12, params={}, now=_NOW)  # type: ignore[arg-type]
    assert db.rollbacks == 1  # ДО ПРАВКИ: 0 — сессия оставалась деактивированной
    # вердикт по свердловскому ряду посчитан и доехал до done-прогона
    assert (out["latest_year"], out["latest_month"], out["age_days"]) == (2026, 6, 72)
    assert (out["pull_lag_days"], out["max_pull_lag_days"], out["alert"]) == (6, 14, 0)
    assert calls["failed"] is None
    assert calls["done"] == out
    # соседний ряд после отката читается как обычно — испорчен был запрос, не БД
    assert out["regions_missing"] == 1
    assert out["alert_regions_missing"] == 1
    assert out["age_days_max"] == 72
    assert "Москва" in " ".join(r.getMessage() for r in records)


def test_failed_rollback_is_not_swallowed(monkeypatch: pytest.MonkeyPatch) -> None:
    """Если не прошёл и откат — мертво соединение целиком, а не один запрос.

    Молча досчитывать вердикт на такой сессии нельзя: наружу должна уйти ошибка отката.
    КРАСНЫЙ ДО ПРАВКИ: отката не было вовсе, наружу уходил PendingRollbackError
    следующего запроса.
    """

    class _DeadSession(_MonDBDriverFault):
        def rollback(self) -> None:
            raise OperationalError("ROLLBACK", {}, TimeoutError("connection is closed"))

    calls = _patch_runs(monkeypatch)
    db = _DeadSession({r: _JUN_2026 for r in est.SBER_MONITORED_REGIONS}, "Москва")
    with pytest.raises(OperationalError):
        mon.check_sber_freshness(db, run_id=13, params={}, now=_NOW)  # type: ignore[arg-type]
    assert calls["done"] is None


def test_every_missing_required_series_is_named_even_on_early_exit(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """КРАСНЫЙ ДО ПРАВКИ: блок с ERROR про пропавший обязательный ряд стоял ПОСЛЕ
    раннего выхода, т.е. в его ветке не выполнялся НИКОГДА.

    Пропали оба обязательных ряда — в мониторинг уходило сообщение только про
    свердловский, хотя под московским 212 937 сделок региона 77 и чинится он отдельно
    (имя ряда в источнике).
    """
    calls = _patch_runs(monkeypatch)
    db = _MonDB({est.SBER_TIME_ADJUST_FALLBACK_REGION: _JUN_2026})
    with _capture_errors() as records:
        out = mon.check_sber_freshness(db, run_id=14, params={}, now=_NOW)  # type: ignore[arg-type]
    msg = " ".join(r.getMessage() for r in records)
    assert "Москва" in msg  # ДО ПРАВКИ: про 77 не уходило ни одного сообщения
    assert est.SBER_TIME_ADJUST_REGION in msg
    # прежняя семантика раннего выхода не тронута
    assert calls["done"] is None
    assert calls["failed"] is not None
    assert "Свердловская область" in calls["failed"]
    assert out["alert_regions_missing"] == 1
    assert (out["latest_year"], out["latest_month"], out["age_days"], out["alert"]) == (0, 0, 0, 0)


def test_missing_required_series_reaches_the_real_alerting_channel(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Настоящий потребитель тревоги — СОБЫТИЕ GlitchTip, а не счётчик прогона.

    alert_regions_missing не читает ни одно правило алертинга; соседи алертят
    ERROR-логом (tests/test_alerts_become_events.py проверяет факт события, а не
    levelno) — проверяем тем же харнессом и тот же факт.
    КРАСНЫЙ ДО ПРАВКИ: на ветке раннего выхода события про «Москва» не было.
    """
    _patch_runs(monkeypatch)
    db = _MonDB({est.SBER_TIME_ADJUST_FALLBACK_REGION: _JUN_2026})
    with glitchtip_events() as events:
        out = mon.check_sber_freshness(db, run_id=15, params={}, now=_NOW)  # type: ignore[arg-type]
    assert any("Москва" in t for t in event_texts(events))
    assert out["alert_regions_missing"] == 1