"""Тесты для реального ПЗЗ-градрегламента в nspd_zoning ответа analyze (Route 2, #1067). Покрывает merge числовых предельных параметров зоны (КСИТ/max_far, высота, этажность, %застройки, min площадь ЗУ) в nspd_zoning через get_or_fetch_zone_regulation: 1. resolver вернул регламент → nspd_zoning несёт max_far/max_height_m/regulation_source 2. resolver кинул исключение → analyze всё равно 200 и nspd_zoning БЕЗ regulation-полей (hot-path-safe, не падаем) 3. флаг enable_zoning_regulation_in_analyze=False → resolver НЕ вызывается, полей нет PR-A (#financial-zoning-decouple) — резолвер РАЗГЕЙЧЕН от dump-зоны (раньше был gated за `_nspd_zoning is not None`, из-за чего gap-участки без dump-зоны не получали max_far → financial_estimate ~0%). Новые тесты на synth-путь: (a) dump nspd_zoning=None + geoportal резолвит "Ж-2"+max_far → nspd_zoning СИНТЕЗИРОВАН (zone_code="Ж-2", source="ekb-geoportal-urbancard", max_far), gate видит зону как жилую (can_build_mkd=True). (b) dump nspd_zoning=None + geoportal вернул None → nspd_zoning остаётся None, analyze не падает, financial_estimate=None. (c) dump ДАЛ nspd_zoning (рег-номер zone_code + raw_props) + geoportal резолвит → regulation-поля добавлены, zone_code НЕ перетёрт (остаётся рег-номер). (d) geoportal кинул → деградирует в None, analyze не падает. Стратегия mock: как в test_analyze_parcel_meta.py — DB mock через dependency_overrides, тяжёлые сервисы патчим через unittest.mock.patch. """ from __future__ import annotations from typing import Any from unittest.mock import MagicMock, patch from fastapi.testclient import TestClient from app.main import app _CAD = "66:41:0204016:10" _WKT = "POLYGON((60.6 56.838, 60.61 56.838, 60.61 56.845, 60.6 56.845, 60.6 56.838))" _GEOJSON = '{"type":"Polygon","coordinates":[[[60.6,56.838],[60.61,56.838]]]}' _NSPD_ZONING = { "zone_code": "Ц-1", "zone_name": "Общественно-деловая зона", "source": "nspd-quarter-dump", "raw_props": {"type_zone": "Общественно-деловая зона"}, } _REGULATION = { "zone_index": "Ц-1", "max_far": 4.0, "max_height_m": 25.0, "max_floors": 9, "max_building_pct": 100.0, "min_parcel_area_m2": 1500.0, "source": "ekb_geoportal", } # PR-A synth fixtures: жилая зона Ж-2 (gate должен пустить МКД) с числами для финмодели. # #1891: main_vri теперь авторитетный сигнал для gate — Ж-2 здесь МКД-зона (код 2.5), # поэтому добавляем main_vri с МКД-кодом, чтобы synth→gate дал can_build_mkd=True. _REGULATION_ZH2 = { "zone_index": "Ж-2", "max_far": 2.5, "max_height_m": 75.0, "max_floors": 25, "max_building_pct": 40.0, "min_parcel_area_m2": 600.0, "main_vri": ["2.5 Среднеэтажная жилая застройка", "2.6 Многоэтажная жилая застройка"], "source": "ekb_geoportal", } # #1891: ИЖС-зона — main_vri БЕЗ МКД-кодов (только 2.1/2.2). gate.is_residential_zone # по main_vri должен дать False, подавляя ^Ж-regex false-positive (Ж-2 ИЖС ≠ МКД). _REGULATION_IZHS = { "zone_index": "Ж-2", "max_far": 0.4, "max_height_m": 12.0, "max_floors": 3, "max_building_pct": 30.0, "min_parcel_area_m2": 400.0, "main_vri": [ "2.1 Для индивидуального жилищного строительства", "2.2 Для ведения личного подсобного хозяйства (приусадебный земельный участок)", ], "source": "ekb_geoportal", } # Dump-provided зона с кадастровым рег-номером в zone_code (реальная форма dump). _NSPD_ZONING_REGNUM = { "zone_code": "66:41-7.14", "zone_name": "66:41-7.14", "source": "nspd-quarter-dump", "raw_props": {"subcategory": 2, "descr": "66:41-7.14"}, } def _make_mapping(data: dict[str, Any]) -> MagicMock: m = MagicMock() m.__getitem__ = lambda self, k: data[k] m.get = lambda k, default=None: data.get(k, default) return m def _make_db_for_analyze() -> MagicMock: """Mock DB Session для analyze_parcel (минимальный — фокус на nspd_zoning merge).""" db = MagicMock() geom_row = _make_mapping({"geom_geojson": _GEOJSON, "geom_wkb": None, "source": "cad_quarter"}) wkt_row = _make_mapping({"wkt": _WKT}) district_row = _make_mapping( { "district_name": "Октябрьский", "ekb_reference_median": 120000, "quarter_median_12m": None, "quarter_deals_count": 0, "dist_to_center": 1500.0, } ) centroid_row = _make_mapping({"lat": 56.84, "lon": 60.605}) call_idx = [0] responses: list[Any] = [ ("first", geom_row), # 0: geom UNION ALL ("first", wkt_row), # 1: WKT ("first", district_row), # 2: district ("all", []), # 3: POI rows ("all", []), # 4: competitor rows ("all", []), # 5: pipeline rows ("first", centroid_row), # 6: centroid ("all", []), # 7: noise rows ("all", []), # 8: hydrology rows ("all", []), # 9: utilities rows # (далее nspd_zoning merge не ходит в db.execute — резолвер замокан) ("first", None), # 10: parcel_meta ("first", None), # 11: market trend ("first", None), # 12: zoning (begin_nested) ("all", []), # 13: success recommendation ("first", None), # 14: market price ("all", []), # 15: recent permits ("scalar", 0), # 16: geotech_risk ("all", []), # 17: neighbors ("first", None), # 18: overlap ] def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock: idx = call_idx[0] call_idx[0] += 1 if idx >= len(responses): r = MagicMock() r.mappings.return_value.first.return_value = None r.mappings.return_value.all.return_value = [] r.scalar.return_value = 0 return r kind, data = responses[idx] r = MagicMock() r.mappings.return_value.first.return_value = data r.mappings.return_value.all.return_value = data if isinstance(data, list) else [] r.scalar.return_value = data if kind == "scalar" else 0 return r db.execute.side_effect = _execute_side_effect ctx = MagicMock() ctx.__enter__ = MagicMock(return_value=ctx) ctx.__exit__ = MagicMock(return_value=False) db.begin_nested.return_value = ctx return db def _override_db(db: MagicMock): def _get_db_override(): yield db return _get_db_override def _dump_result( nspd_zoning: dict[str, Any] | None, *, dump_available: bool = False, ) -> dict[str, Any]: """Сборка результата get_quarter_dump_data. nspd_zoning — свежая копия (merge мутирует).""" return { "nspd_zoning": dict(nspd_zoning) if nspd_zoning is not None else None, "nspd_zouit_overlaps": [], "nspd_engineering_nearby": [], "nspd_risk_zones": [], "nspd_opportunity_parcels": [], "nspd_red_lines": [], "nspd_dump": { "available": dump_available, "stale": False, "harvest_triggered": False, }, } def _base_patches( *, dump_zoning: dict[str, Any] | None = _NSPD_ZONING, dump_available: bool = False, mock_gate: bool = True, ) -> list[Any]: """Патчи тяжёлых сервисов. dump_zoning — что вернёт get_quarter_dump_data в nspd_zoning (None = gap-участок). dump_available — флаг свежести dump (gate evaluates зону ТОЛЬКО при available=True). mock_gate — если False, реальный compute_gate_verdict (для проверки synth→gate пути). """ patches = [ patch("app.api.v1.parcels.get_air_quality_cached", return_value=None), patch("app.api.v1.parcels.get_weather_cached", return_value=None), patch("app.api.v1.parcels.get_seasonal_weather_cached", return_value=None), patch( "app.api.v1.parcels.get_quarter_dump_data", return_value=_dump_result(dump_zoning, dump_available=dump_available), ), patch("app.api.v1.parcels.compute_velocity", return_value=None), patch( "app.api.v1.parcels.build_ird_analyze_block", return_value={ "ird_overlaps": [], "ird_by_kind": {}, "opportunity_overlaps": [], "planning_projects": [], "functional_zone": None, "krt": [], "zone_regulation": None, }, ), ] if mock_gate: patches.append( patch("app.api.v1.parcels.compute_gate_verdict", return_value={"verdict": "unknown"}) ) return patches def test_zoning_regulation_merged_when_resolved() -> None: """resolver вернул регламент → nspd_zoning несёт max_far/max_height_m/regulation_source.""" from app.core.db import get_db db = _make_db_for_analyze() app.dependency_overrides[get_db] = _override_db(db) patches = [ *_base_patches(), patch( "app.api.v1.parcels.get_or_fetch_zone_regulation", return_value=dict(_REGULATION), ), ] for p in patches: p.start() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() zoning = body["nspd_zoning"] assert zoning is not None # существующие ключи сохранены assert zoning["zone_code"] == "Ц-1" assert zoning["source"] == "nspd-quarter-dump" # regulation-поля добавлены assert zoning["max_far"] == 4.0 assert zoning["max_height_m"] == 25.0 assert zoning["max_floors"] == 9 assert zoning["max_building_pct"] == 100.0 assert zoning["min_parcel_area_m2"] == 1500.0 assert zoning["regulation_zone_index"] == "Ц-1" assert zoning["regulation_source"] == "ekb-geoportal-urbancard" finally: app.dependency_overrides.clear() for p in patches: p.stop() def test_zoning_regulation_graceful_on_resolver_exception() -> None: """resolver кинул → analyze 200, nspd_zoning сохранён БЕЗ regulation-полей (не падаем).""" from app.core.db import get_db db = _make_db_for_analyze() app.dependency_overrides[get_db] = _override_db(db) patches = [ *_base_patches(), patch( "app.api.v1.parcels.get_or_fetch_zone_regulation", side_effect=RuntimeError("geoportal timeout"), ), ] for p in patches: p.start() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() zoning = body["nspd_zoning"] assert zoning is not None assert zoning["zone_code"] == "Ц-1" # существующие ключи целы # regulation-поля НЕ просочились (graceful degrade) assert "max_far" not in zoning assert "max_height_m" not in zoning assert "regulation_source" not in zoning finally: app.dependency_overrides.clear() for p in patches: p.stop() def test_zoning_regulation_skipped_when_flag_off() -> None: """флаг OFF → resolver НЕ вызывается, regulation-полей нет, nspd_zoning без изменений.""" from app.core.db import get_db db = _make_db_for_analyze() app.dependency_overrides[get_db] = _override_db(db) resolver_mock = MagicMock(return_value=dict(_REGULATION)) patches = [ *_base_patches(), patch("app.api.v1.parcels.settings.enable_zoning_regulation_in_analyze", False), patch("app.api.v1.parcels.get_or_fetch_zone_regulation", resolver_mock), ] for p in patches: p.start() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() zoning = body["nspd_zoning"] assert zoning is not None assert zoning["zone_code"] == "Ц-1" assert "max_far" not in zoning # резолв пропущен → полей нет resolver_mock.assert_not_called() # флаг OFF → резолвер не дёрнут finally: app.dependency_overrides.clear() for p in patches: p.stop() # ── PR-A: synth-путь (dump зону не дал, geoportal резолвит) ──────────────────── def test_synth_when_dump_zoning_none_and_geoportal_resolves() -> None: """(a) dump nspd_zoning=None + geoportal "Ж-2"+main_vri(МКД) → synth + gate видит жилую.""" from app.core.db import get_db db = _make_db_for_analyze() app.dependency_overrides[get_db] = _override_db(db) patches = [ # dump_available=True + реальный gate → проверяем synth→gate путь end-to-end. *_base_patches(dump_zoning=None, dump_available=True, mock_gate=False), patch( "app.api.v1.parcels.get_or_fetch_zone_regulation", return_value=dict(_REGULATION_ZH2), ), ] for p in patches: p.start() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() zoning = body["nspd_zoning"] # synth построил минимальный dict из geoportal-результата assert zoning is not None assert zoning["zone_code"] == "Ж-2" # zone_index → zone_code (для ^Ж regex) assert zoning["source"] == "ekb-geoportal-urbancard" assert zoning["zone_name"] is None assert zoning["max_far"] == 2.5 assert zoning["regulation_zone_index"] == "Ж-2" assert zoning["regulation_source"] == "ekb-geoportal-urbancard" # #1891: main_vri проброшен из regulation в synth-зону assert "2.5 Среднеэтажная жилая застройка" in zoning["main_vri"] # gate увидел synth-зону как жилую по main_vri → can_build_mkd=True assert body["gate_verdict"]["can_build_mkd"] is True finally: app.dependency_overrides.clear() for p in patches: p.stop() def test_synth_izhs_main_vri_suppresses_residential_false_positive() -> None: """#1891: dump None + geoportal "Ж-2" ИЖС (main_vri без МКД-кодов) → gate НЕ жилая. Раньше synth клал zone_code="Ж-2" → gate ^Ж regex давал can_build_mkd=True даже на ИЖС-зоне (FALSE POSITIVE). Теперь main_vri (авторитетный) перебивает regex → gate видит non-residential → can_build_mkd is NOT True. """ from app.core.db import get_db db = _make_db_for_analyze() app.dependency_overrides[get_db] = _override_db(db) patches = [ *_base_patches(dump_zoning=None, dump_available=True, mock_gate=False), patch( "app.api.v1.parcels.get_or_fetch_zone_regulation", return_value=dict(_REGULATION_IZHS), ), ] for p in patches: p.start() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() zoning = body["nspd_zoning"] assert zoning is not None assert zoning["zone_code"] == "Ж-2" # display-индекс цел # main_vri проброшен — авторитетный сигнал «ИЖС, МКД нельзя» assert "2.1 Для индивидуального жилищного строительства" in zoning["main_vri"] # FIX: false-positive подавлен — gate НЕ даёт can_build_mkd True на ИЖС assert body["gate_verdict"]["can_build_mkd"] is not True finally: app.dependency_overrides.clear() for p in patches: p.stop() def test_no_synth_when_dump_none_and_geoportal_none() -> None: """(b) dump nspd_zoning=None + geoportal None → nspd_zoning None, financial None, не падаем.""" from app.core.db import get_db db = _make_db_for_analyze() app.dependency_overrides[get_db] = _override_db(db) patches = [ *_base_patches(dump_zoning=None, dump_available=True, mock_gate=False), patch("app.api.v1.parcels.get_or_fetch_zone_regulation", return_value=None), ] for p in patches: p.start() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() assert body["nspd_zoning"] is None # synth НЕ сработал (geoportal вернул None) assert body["financial_estimate"] is None # нет чисел зоны → финмодель None finally: app.dependency_overrides.clear() for p in patches: p.stop() def test_dump_provided_zone_code_not_overwritten_by_synth() -> None: """(c) dump ДАЛ зону (рег-номер) + geoportal резолвит → regulation добавлен, code цел.""" from app.core.db import get_db db = _make_db_for_analyze() app.dependency_overrides[get_db] = _override_db(db) patches = [ *_base_patches(dump_zoning=_NSPD_ZONING_REGNUM, dump_available=True, mock_gate=True), patch( "app.api.v1.parcels.get_or_fetch_zone_regulation", return_value=dict(_REGULATION_ZH2), ), ] for p in patches: p.start() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() zoning = body["nspd_zoning"] assert zoning is not None # zone_code НЕ перетёрт synth'ом — остаётся dump-рег-номер + raw_props assert zoning["zone_code"] == "66:41-7.14" assert zoning["source"] == "nspd-quarter-dump" assert zoning["raw_props"]["subcategory"] == 2 # regulation-поля ДОБАВЛЕНЫ из geoportal assert zoning["max_far"] == 2.5 assert zoning["regulation_zone_index"] == "Ж-2" assert zoning["regulation_source"] == "ekb-geoportal-urbancard" finally: app.dependency_overrides.clear() for p in patches: p.stop() def test_synth_path_graceful_when_geoportal_raises() -> None: """(d) dump None + geoportal кинул → деградирует в None, analyze не падает.""" from app.core.db import get_db db = _make_db_for_analyze() app.dependency_overrides[get_db] = _override_db(db) patches = [ *_base_patches(dump_zoning=None, dump_available=True, mock_gate=False), patch( "app.api.v1.parcels.get_or_fetch_zone_regulation", side_effect=RuntimeError("geoportal timeout"), ), ] for p in patches: p.start() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() assert body["nspd_zoning"] is None # synth НЕ сработал — резолвер кинул assert body["financial_estimate"] is None finally: app.dependency_overrides.clear() for p in patches: p.stop()