Light1YT
33d7829221
Resolve numeric ПЗЗ limits (max_far/КСИТ, max_height_m, max_floors, max_building_pct, min_parcel_area_m2) by parcel centroid via the coordinate resolver get_or_fetch_zone_regulation (cache-first; bounded 3s live geoportal) and merge into the /analyze nspd_zoning object. Powers the ПТИЦА cockpit's real КСИТ/height/density (was placeholder). Flag-gated (enable_zoning_regulation_in_analyze, default on) and hot-path-safe: any exception/timeout → null fields, analyze never 500s or slow-fails (3s cap). Additive — existing nspd_zoning keys untouched. 198 tests pass. Note: zone_index_at is a live WFS call per analyze even on warm cache (~0.3-0.5s healthy / 3s cap degraded) — optimization tracked as follow-up.
261 lines
9.9 KiB
Python
261 lines
9.9 KiB
Python
"""Тесты для реального ПЗЗ-градрегламента в 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()
|