test(ptica): снятие SAVEPOINT теперь роняет тест во всех местах кластера A (#2464)

Мутационный прогон 19.08 показал, что у трёх закрытых пунктов кластера снятие
`with db.begin_nested():` не роняет НИ ОДНОГО теста:

    saturation.py        снят SAVEPOINT -> 465 passed
    zone_regulation.py   снят SAVEPOINT -> 195 passed
    pat_lookup.py        снят SAVEPOINT ->  21 passed

Код там правильный — беззащитны регрессии. Причина общая: существующие проверки
построены на MagicMock, у которого нет aborted-состояния, поэтому второй execute
проходит при любом коде. Такая проверка ловит только проглоченное исключение.

Добавлен двойник tests/support/pg_session.py, воспроизводящий семантику Postgres:
упавший запрос переводит транзакцию в aborted; следующий execute падает с тем же
текстом, что отдаёт Postgres; выход из begin_nested с исключением = ROLLBACK TO
SAVEPOINT и снимает aborted.

Проверяется СВОЙСТВО «сессия вызывающего осталась рабочей», а не наличие вызова
begin_nested: второе зафиксировало бы реализацию, первое переживёт любой
корректный способ починки. Плюс два контроля на сам двойник — без них проверки
были бы зелёными по построению.

Заодно закрыт пробел в connection_capacity_lookup. Его собственный тест краснеет
только при снятии ВСЕХ пяти SAVEPOINT, потому что проверяет
`begin_nested.call_count >= 1` — «хоть один был», а не «этот запрос обёрнут».
Снятие одного из пяти (обычный случай при рефакторинге одной функции) проходило
незамеченным. Новые проверки адресные: по одной на функцию.

Прицел важен: первая версия теста для _query_gas_city_grs роняла ПЕРВЫЙ execute,
а он принадлежит _query_gas_outlet_counts, которую эта функция зовёт внутри себя —
то есть тест сторожил соседа. Исправлено на fail_on=(2,).

Проверено мутацией: снятие SAVEPOINT в любом из трёх модулей роняет ровно свою
проверку; в connection_capacity_lookup поштучное снятие каждого из пяти — тоже
(5 из 5, до правки 4 из 5).

pytest tests/services: 3070 passed, 14 skipped, rc=0

backend/tests/services/site_finder/test_2464a_savepoint_regressions.py

@@ -0,0 +1,185 @@
+"""#2464 кластер A: снятие SAVEPOINT обязано ронять тест.
+
+Три места кластера отмечены закрытыми, код в них правильный — но мутационный
+прогон 19.08.2026 показал, что снятие `with db.begin_nested():` не роняет ни одного
+теста:
+
+    saturation.py        снят SAVEPOINT -> 465 passed
+    zone_regulation.py   снят SAVEPOINT -> 195 passed
+    pat_lookup.py        снят SAVEPOINT ->  21 passed
+
+Причина общая: существующие проверки построены на `MagicMock`, у которого нет
+aborted-состояния, поэтому второй `execute` проходит при любом коде. Такая проверка
+ловит только проглоченное исключение, но не отсутствие SAVEPOINT.
+
+Здесь используется двойник из `tests/support/pg_session.py`, воспроизводящий
+семантику Postgres. Проверяется СВОЙСТВО — «после сбоя сессия вызывающего осталась
+рабочей», — а не наличие вызова `begin_nested` в коде: последнее зафиксировало бы
+реализацию, а первое переживёт любой корректный способ починки.
+
+Отдельно проверяется сам двойник: без SAVEPOINT он ОБЯЗАН отравляться. Без этого
+контроля тесты ниже были бы зелёными по построению — ровно та ловушка, которую они
+призваны закрыть.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.exc import OperationalError
+
+from app.services.scrapers.ekb_geoportal_client import ZoneRegulation
+from app.services.site_finder.pat_lookup import parcel_pat_subzones
+from app.services.site_finder.saturation import compute_district_saturation
+from app.services.site_finder.zone_regulation import upsert_zone_regulation
+from tests.support.pg_session import AbortedTransactionError, PostgresLikeSession
+
+_WKT = "POLYGON((60.6 56.83, 60.61 56.83, 60.61 56.84, 60.6 56.84, 60.6 56.83))"
+
+
+def _sqla_error() -> OperationalError:
+    """Обработчики ловят конкретные типы SQLAlchemy, а не голый Exception."""
+    return OperationalError("SELECT 1", {}, Exception("relation does not exist"))
+
+
+# ── контроль на сам двойник ────────────────────────────────────────────────────
+
+
+def test_double_poisons_the_session_without_a_savepoint() -> None:
+    db = PostgresLikeSession()
+    with pytest.raises(RuntimeError):
+        db.execute("boom")
+    with pytest.raises(AbortedTransactionError):
+        db.execute("SELECT 1")
+
+
+def test_double_recovers_when_the_savepoint_unwinds() -> None:
+    db = PostgresLikeSession()
+    with pytest.raises(RuntimeError), db.begin_nested():
+        db.execute("boom")
+    db.execute("SELECT 1")  # не должно бросить
+    assert db.savepoints_entered == 1
+
+
+# ── сами регрессии ─────────────────────────────────────────────────────────────
+
+
+def test_saturation_leaves_the_caller_session_usable() -> None:
+    """`compute_district_saturation` делит сессию с analyze_parcel."""
+    db = PostgresLikeSession()
+
+    assert compute_district_saturation(db, "Чкаловский") is None
+
+    db.execute("SELECT 1")  # следующий блок analyze_parcel
+    assert db.calls == 2
+
+
+def test_pat_subzones_leaves_the_caller_session_usable() -> None:
+    """`parcel_pat_subzones` вызывается из /analyze на общей request-сессии."""
+    db = PostgresLikeSession(error=_sqla_error())
+
+    assert parcel_pat_subzones(db, _WKT) == []
+
+    db.execute("SELECT 1")
+    assert db.calls == 2
+
+
+def test_zone_regulation_upsert_leaves_the_caller_session_usable() -> None:
+    """`upsert_zone_regulation` пишет в кэш посреди чужой транзакции."""
+    db = PostgresLikeSession(error=_sqla_error())
+    reg = ZoneRegulation(
+        zone_index="Ж-5",
+        zone_full_name="Ж-5 зона",
+        main_vri=[],
+        conditional_vri=[],
+        auxiliary_vri=[],
+        limit_params=[],
+    )
+
+    assert upsert_zone_regulation(db, reg) is None
+
+    db.execute("SELECT 1")
+    assert db.calls == 2
+
+
+# ── контроль: без сбоя поведение прежнее ───────────────────────────────────────
+
+
+def test_no_failure_means_no_savepoint_side_effects() -> None:
+    """Исправная БД: pat_lookup отдаёт строки, лишних откатов нет."""
+    db = PostgresLikeSession(
+        fail_on=(),
+        rows=[
+            {
+                "subzone_no": "ПАТ-1",
+                "name": "Кольцово",
+                "restriction": "высота",
+                "aerodrome": "USSS",
+            }
+        ],
+    )
+
+    out = parcel_pat_subzones(db, _WKT)
+
+    assert out and out[0]["subzone_no"] == "ПАТ-1"
+    assert db.aborted is False
+
+
+# ── connection_capacity_lookup: пять SAVEPOINT, снятие ОДНОГО не ловилось ───────
+#
+# У файла есть свой тест (`test_connection_capacity_savepoint.py`), и он краснеет,
+# если снять ВСЕ пять. Но проверка там — `assert db.begin_nested.call_count >= 1`,
+# то есть «хоть один savepoint был», а не «этот запрос обёрнут». Мутационный прогон
+# 19.08: снятие одного из пяти проходит незамеченным, а рефакторинг одной функции —
+# ровно такой случай.
+#
+# Проверки ниже адресные: у каждой своя функция и свой двойник, поэтому снятие
+# SAVEPOINT в любой ОДНОЙ из них роняет ровно свою проверку.
+
+
+def test_gas_city_grs_leaves_the_caller_session_usable() -> None:
+    """Роняем ВТОРОЙ запрос: первый принадлежит `_query_gas_outlet_counts`, который
+    эта функция зовёт внутри себя и у которого свой SAVEPOINT. Целиться надо в
+    запрос самой city_grs, иначе проверка сторожила бы соседа."""
+    from app.services.site_finder.connection_capacity_lookup import _query_gas_city_grs
+
+    db = PostgresLikeSession(fail_on=(2,), error=_sqla_error())
+    out = _query_gas_city_grs(db)
+
+    assert out["city_grs"] == []
+    db.execute("SELECT 1")
+
+
+def test_gas_outlet_counts_leaves_the_caller_session_usable() -> None:
+    from app.services.site_finder.connection_capacity_lookup import _query_gas_outlet_counts
+
+    db = PostgresLikeSession(error=_sqla_error())
+    _query_gas_outlet_counts(db)
+
+    db.execute("SELECT 1")
+
+
+def test_gas_outlet_points_leaves_the_caller_session_usable() -> None:
+    from app.services.site_finder.connection_capacity_lookup import _query_gas_outlet_points
+
+    db = PostgresLikeSession(error=_sqla_error())
+    assert _query_gas_outlet_points(db, _WKT) == []
+
+    db.execute("SELECT 1")
+
+
+def test_heat_latest_leaves_the_caller_session_usable() -> None:
+    from app.services.site_finder.connection_capacity_lookup import _query_heat_latest
+
+    db = PostgresLikeSession(error=_sqla_error())
+    _query_heat_latest(db)
+
+    db.execute("SELECT 1")
+
+
+def test_nearby_network_zones_leaves_the_caller_session_usable() -> None:
+    from app.services.site_finder.connection_capacity_lookup import _query_nearby_network_zones
+
+    db = PostgresLikeSession(error=_sqla_error())
+    assert _query_nearby_network_zones(db, _WKT) == []
+
+    db.execute("SELECT 1")

backend/tests/support/pg_session.py

@@ -0,0 +1,99 @@
+"""Сессия-двойник с настоящей семантикой aborted-транзакции Postgres.
+
+Зачем отдельный помощник, а не MagicMock: у мока нет состояния «транзакция
+оборвана», поэтому второй `execute` в тесте проходит независимо от того, обёрнут
+первый в SAVEPOINT или нет. Проверка на отравление сессии на таком моке зелена по
+построению — она может поймать только проглоченное исключение.
+
+Замер 19.08.2026 (мутационный прогон по кластеру A эпика #2464): у трёх мест из
+одиннадцати снятие SAVEPOINT не роняет НИ ОДНОГО теста — saturation (465 passed),
+zone_regulation (195), pat_lookup (21). Код там правильный; беззащитны регрессии.
+
+Семантика, которую двойник воспроизводит:
+  * упавший `execute` переводит транзакцию в aborted;
+  * любой следующий `execute` в aborted-состоянии падает с тем же текстом, что
+    отдаёт Postgres, — до конца транзакции;
+  * выход из `begin_nested()` с исключением = ROLLBACK TO SAVEPOINT: снимает
+    aborted, внешняя транзакция остаётся рабочей.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from typing import Any
+
+_ABORTED_MESSAGE = "current transaction is aborted, commands ignored until end of transaction block"
+
+
+class AbortedTransactionError(RuntimeError):
+    """Аналог psycopg InFailedSqlTransaction."""
+
+
+class PostgresLikeSession:
+    """Минимальная сессия: aborted-состояние + рабочий SAVEPOINT.
+
+    fail_on: номера вызовов `execute` (с единицы), которые должны упасть.
+    error: исключение, которым падать (важно — обработчики ловят конкретные типы
+    SQLAlchemy, а не голый Exception).
+    rows / row: что отдавать при успешном вызове.
+    """
+
+    def __init__(
+        self,
+        *,
+        fail_on: tuple[int, ...] = (1,),
+        error: BaseException | None = None,
+        rows: list[Any] | None = None,
+        row: Any = None,
+    ) -> None:
+        self.calls = 0
+        self.aborted = False
+        self.savepoints_entered = 0
+        self._fail_on = fail_on
+        self._error = error or RuntimeError("simulated DB failure")
+        self._rows = rows if rows is not None else []
+        self._row = row
+
+    @contextmanager
+    def begin_nested(self):  # type: ignore[no-untyped-def]
+        self.savepoints_entered += 1
+        try:
+            yield
+        except BaseException:
+            # ROLLBACK TO SAVEPOINT — внешняя транзакция снова пригодна.
+            self.aborted = False
+            raise
+
+    def execute(self, *_args: Any, **_kwargs: Any) -> _Result:
+        if self.aborted:
+            raise AbortedTransactionError(_ABORTED_MESSAGE)
+        self.calls += 1
+        if self.calls in self._fail_on:
+            self.aborted = True
+            raise self._error
+        return _Result(self._rows, self._row)
+
+    # ── прочее, что могут дёрнуть обработчики ────────────────────────────────
+    def commit(self) -> None:
+        if self.aborted:
+            raise AbortedTransactionError(_ABORTED_MESSAGE)
+
+    def rollback(self) -> None:
+        self.aborted = False
+
+
+class _Result:
+    def __init__(self, rows: list[Any], row: Any) -> None:
+        self._rows, self._row = rows, row
+
+    def mappings(self) -> _Result:
+        return self
+
+    def all(self) -> list[Any]:
+        return self._rows
+
+    def first(self) -> Any:
+        return self._row
+
+    def scalar(self) -> Any:
+        return self._row