"""#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")