"""Тесты для реального ПЗЗ-градрегламента в 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 НЕ вызывается, полей нет Стратегия mock: как в test_analyze_parcel_meta.py — DB mock через dependency_overrides, тяжёлые сервисы патчим через unittest.mock.patch. get_quarter_dump_data замокан так, чтобы вернуть непустой nspd_zoning (иначе merge пропускается по guard `_nspd_zoning is not None`). """ 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", } 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": "Октябрьский", "median_price_per_m2": 120000, "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 _base_patches() -> list[Any]: """Патчи тяжёлых сервисов; get_quarter_dump_data отдаёт непустой nspd_zoning.""" return [ 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={ # dict(_NSPD_ZONING) — свежая копия per-test, merge мутирует её in-place "nspd_zoning": dict(_NSPD_ZONING), "nspd_zouit_overlaps": [], "nspd_engineering_nearby": [], "nspd_risk_zones": [], "nspd_opportunity_parcels": [], "nspd_red_lines": [], "nspd_dump": {"available": False, "stale": False, "harvest_triggered": False}, }, ), patch("app.api.v1.parcels.compute_velocity", return_value=None), patch("app.api.v1.parcels.compute_gate_verdict", return_value={"verdict": "unknown"}), 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, }, ), ] 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()