lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(#29,#232): wire cad_parcels.permitted_use in analyze + cad_zouit fallback (G2+G3)

Browse source 
TASK A (#29 G2): add parcel_meta to analyze response
- New ParcelMeta Pydantic schema in app/schemas/parcel.py
- SELECT from cad_parcels WHERE cad_num=:c in analyze_parcel() (step 9f)
- Returns permitted_use_established_by_document, land_record_category_type,
  land_record_subtype, cost_value; None when row absent
- Tests: test_analyze_parcel_meta.py (found + not-found cases)

TASK B (#232 G3): cad_zouit fallback in _get_zouit_overlaps
- When nspd_quarter_dumps has zouit_count==0, fall back to ST_Intersects
  query on cad_zouit (3483 rows, GIST indexed)
- Overlaps tagged with source='cad_zouit'; format compatible with NSPD path
- gate_verdict.py: BLOCKER_TYPE_ZONE_KEYWORDS tuple for keyword-based
  classification (охранная зона / трубопровод / электр / газ -> blocker;
  СЗЗ -> warning); NSPD subcategory path preserved backward-compat
- Tests: 6 new test cases in test_gate_verdict.py covering cad_zouit path
  and backward-compat for NSPD subcategory path

Updated db.execute call sequence in test_analyze_*.py (index shift +1 at pos 10).
This commit is contained in:
lekss361 2026-05-17 08:17:22 +03:00
parent 56bf28758c
commit f70be684da
9 changed files with 551 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
backend/app/api/v1/parcels.py
View file

@ -23,6 +23,7 @@ from app.schemas.parcel import (
ConnectionPointsResponse,
OpportunityParcel,
ParcelDetail,
ParcelMeta,
ParcelSearchRequest,
ParcelSearchResponse,
RedLine,
@ -1627,6 +1628,35 @@ def analyze_parcel(
# Independent dict per request — never mutate module singleton.
nspd_dump_data = make_empty_result()
# 9f) Parcel meta — ВРИ и кадастровые метаданные из cad_parcels (#29 G2)
parcel_meta: ParcelMeta | None = None
try:
pm_row = (
db.execute(
text("""
SELECT permitted_use_established_by_document AS permitted_use,
land_record_category_type AS land_category,
land_record_subtype AS land_subtype,
cost_value AS cad_cost
FROM cad_parcels
WHERE cad_num = CAST(:c AS text)
LIMIT 1
"""),
{"c": cad_num},
)
.mappings()
.first()
)
if pm_row:
parcel_meta = ParcelMeta(
permitted_use=pm_row["permitted_use"],
land_category=pm_row["land_category"],
land_subtype=pm_row["land_subtype"],
cad_cost=float(pm_row["cad_cost"]) if pm_row["cad_cost"] is not None else None,
)
except Exception as e:
logger.warning("parcel_meta query failed for %s: %s", cad_num, e)
# 10) Market trend — динамика цен ДДУ в радиусе 3 км за 6 vs предыдущие 6 месяцев
market_trend: dict[str, Any] | None = None
try:
@ -2051,6 +2081,8 @@ def analyze_parcel(
"market_trend": market_trend,
# #33 D2: квартальная ценовая статистика из mv_quarter_price_per_m2
"market_price": market_price,
# #29 G2: кадастровые метаданные участка (ВРИ, категория, кад. стоимость)
"parcel_meta": parcel_meta.model_dump() if parcel_meta is not None else None,
# #105 Phase 5: РНС/РВЭ в квартале (quarter prefix match; после Phase 3 → spatial 500m)
"recent_permits_in_quarter": recent_permits,
"permits_summary": permits_summary,

16
backend/app/schemas/parcel.py
View file

@ -315,6 +315,22 @@ class MarketPrice(BaseModel):
source: Literal["quarter_mv", "no_data"] = "no_data"
# ── Analyze endpoint parcel meta (#29 G2) ────────────────────────────────────
class ParcelMeta(BaseModel):
"""Кадастровые метаданные участка из cad_parcels (#29 G2).
Поля берутся из NSPD bulk-ingest (issue #168). Строки сырые, без нормализации.
"""
permitted_use: str | None = None # permitted_use_established_by_document
land_category: str | None = None # land_record_category_type
land_subtype: str | None = None # land_record_subtype
cad_cost: float | None = None # cost_value (кадастровая стоимость, руб.)
source: str = "cad_parcels"
# ── Analyze endpoint inline weights (#201) ────────────────────────────────────

52
backend/app/services/site_finder/gate_verdict.py
View file

@ -18,11 +18,25 @@ RESIDENTIAL_KEYWORDS = ("жил",)
# ── ЗОУИТ taxonomy ────────────────────────────────────────────────────────────
# Subcategory codes которые блокируют МКД (охранные зоны ЛЭП/газа/трубопровода)
# Subcategory codes которые блокируют МКД (охранные зоны ЛЭП/газа/трубопровода).
# Используется для overlaps из NSPD dump (subcategory INT там заполнен).
BLOCKER_SUBCATEGORIES: dict[int, str] = {
17: "Инженерные коммуникации (охранная зона ЛЭП/газа/трубопровода)",
}
# Keyword-based blockers для cad_zouit fallback (#232).
# cad_zouit.subcategory = 100% NULL, поэтому классификация по type_zone substring.
# Порядок: lowercase, substring match (any wins → blocker).
BLOCKER_TYPE_ZONE_KEYWORDS: tuple[str, ...] = (
"охранная зона",
"трубопровод",
"электр",
"газ",
)
# type_zone substrings, которые дают warning (не blocker) из cad_zouit.
WARNING_TYPE_ZONE_KEYWORDS: tuple[str, ...] = ("сзз", "санитарно-защитная")
# Engineering nearby search radius (метры) — совпадает с quarter_dump_lookup.py
ENGINEERING_NEARBY_THRESHOLD_M = 200
@ -136,6 +150,42 @@ def compute_gate_verdict(
# Check 2 — ЗОУИТ overlaps
checks.append("ЗОУИТ пересечения")
for overlap in nspd_zouit_overlaps or []:
src = overlap.get("source", "nspd-quarter-dump")
if src == "cad_zouit":
# cad_zouit fallback path: classify by type_zone keywords (#232).
# subcategory = NULL в cad_zouit, поэтому subcategory-based logic не применяется.
type_zone_lower = (overlap.get("type_zone") or overlap.get("layer") or "").lower()
if any(kw in type_zone_lower for kw in BLOCKER_TYPE_ZONE_KEYWORDS):
blockers.append(
Blocker(
code="ZOUIT_CAD_BLOCKER",
detail=(
f"Охранная зона ({overlap.get('type_zone', '')}): "
f"{overlap.get('name', '')}"
),
)
)
elif any(kw in type_zone_lower for kw in WARNING_TYPE_ZONE_KEYWORDS):
warnings.append(
Warning(
code="ZOUIT_CAD_SZZ",
detail=(
f"СЗЗ ({overlap.get('type_zone', '')}): " f"{overlap.get('name', '')}"
),
)
)
else:
warnings.append(
Warning(
code="ZOUIT_CAD_OTHER",
detail=(
f"ЗОУИТ cad ({overlap.get('type_zone', '')}): "
f"{overlap.get('name', '')}"
),
)
)
else:
# NSPD dump path: subcategory-based logic (backward-compat).
sub = overlap.get("subcategory")
if isinstance(sub, int) and sub in BLOCKER_SUBCATEGORIES:
blockers.append(

60
backend/app/services/site_finder/quarter_dump_lookup.py
View file

@ -296,10 +296,10 @@ def _get_zouit_overlaps(
Проверяем 5 групп: zouit_okn, zouit_engineering, zouit_natural,
zouit_protected, zouit_other.
layer_counts — денормализованные счётчики. Если zouit_count == 0 —
пропускаем heavy jsonb_array_elements scan.
пропускаем nspd dump scan и делаем fallback на cad_zouit (#232).
"""
if layer_counts is not None and layer_counts.get("zouit_count", 1) == 0:
return []
nspd_has_zouit = layer_counts is None or layer_counts.get("zouit_count", 1) != 0
if nspd_has_zouit:
try:
rows = db.execute(
text(
@ -339,6 +339,7 @@ def _get_zouit_overlaps(
"subcategory": props.get("subcategory") or props.get("type_zone"),
"name": props.get("name") or props.get("object_name"),
"raw_props": props,
"source": "nspd-quarter-dump",
}
)
return result
@ -346,6 +347,59 @@ def _get_zouit_overlaps(
logger.warning("nspd zouit query failed for quarter=%s: %s", quarter, e)
return []
# Fallback: cad_zouit (#232) — nspd dump пуст по ЗОУИТ.
# cad_zouit.subcategory = 100% NULL (verified), используем type_zone keywords.
return _get_cad_zouit_overlaps(db, parcel_wkt)
def _get_cad_zouit_overlaps(db: Session, parcel_wkt: str) -> list[dict[str, Any]]:
"""Fallback: читать ЗОУИТ из cad_zouit когда nspd_quarter_dumps пуст (#232).
Использует GIST index cad_zouit_geom_gist через ST_Intersects.
Возвращает тот же формат что nspd-dump path (с доп. полем source='cad_zouit').
"""
try:
rows = db.execute(
text(
"""
SELECT type_zone,
category_name,
name_by_doc AS name,
reg_numb_border,
id AS zouit_id
FROM cad_zouit
WHERE ST_Intersects(geom, ST_GeomFromText(:wkt, 4326))
ORDER BY id
"""
),
{"wkt": parcel_wkt},
).fetchall()
result: list[dict[str, Any]] = []
for r in rows:
type_zone: str = r[0] or ""
result.append(
{
"group_key": "cad_zouit",
"layer": type_zone,
"subcategory": None, # subcategory = 100% NULL в cad_zouit
"name": r[2],
"raw_props": {
"type_zone": type_zone,
"category_name": r[1],
"reg_numb_border": r[3],
"zouit_id": r[4],
},
"source": "cad_zouit",
"type_zone": type_zone,
}
)
logger.info("cad_zouit fallback: found %d overlaps for wkt prefix", len(result))
return result
except Exception as e:
logger.warning("cad_zouit fallback query failed: %s", e)
return []
def _get_engineering_nearby(
db: Session,

30
backend/tests/api/v1/test_analyze_inline_weights.py
View file

@ -58,14 +58,15 @@ def _make_db_for_analyze(
7. Noise rows → .mappings().all()
8. Hydrology → .mappings().all()
9. Utilities → .mappings().all()
10. Market trend → .mappings().first()
11. Zoning (begin_nested) → .mappings().first()
12. Success recommendation (begin_nested) → .mappings().all()
13. Market price (begin_nested) → .mappings().first()
14. Recent permits (begin_nested) → .mappings().all() ← NEW #105 Phase 5
15. _geotech_risk (industrial count) → .scalar()
16. _neighbors_summary (neighbor_rows) → .mappings().all()
17. _neighbors_summary (overlap_row) → .mappings().first()
10. parcel_meta (cad_parcels) → .mappings().first() ← #29 G2
11. Market trend → .mappings().first()
12. Zoning (begin_nested) → .mappings().first()
13. Success recommendation (begin_nested) → .mappings().all()
14. Market price (begin_nested) → .mappings().first()
15. Recent permits (begin_nested) → .mappings().all() ← #105 Phase 5
16. _geotech_risk (industrial count) → .scalar()
17. _neighbors_summary (neighbor_rows) → .mappings().all()
18. _neighbors_summary (overlap_row) → .mappings().first()
begin_nested() — возвращаем context manager чтобы поддержать `with` statement.
"""
@ -106,12 +107,13 @@ def _make_db_for_analyze(
("all", []), # 7: noise rows
("all", []), # 8: hydrology rows
("all", []), # 9: utilities rows
("first", None), # 10: market trend
("first", None), # 11: zoning (inside begin_nested)
("all", []), # 12: success recommendation (inside begin_nested)
("scalar", 0), # 13: geotech_risk industrial count
("all", []), # 14: neighbors
("first", None), # 15: overlap
("first", None), # 10: parcel_meta (cad_parcels) ← #29 G2
("first", None), # 11: market trend
("first", None), # 12: zoning (inside begin_nested)
("all", []), # 13: success recommendation (inside begin_nested)
("scalar", 0), # 14: geotech_risk industrial count
("all", []), # 15: neighbors
("first", None), # 16: overlap
]
def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:

35
backend/tests/api/v1/test_analyze_market_price.py
View file

@ -8,7 +8,7 @@
Стратегия mock: аналогична test_analyze_inline_weights.py — DB mock через
dependency_overrides, тяжёлые сервисы патчим через unittest.mock.patch.
Порядок db.execute calls в analyze_parcel (v3.7 + #33):
Порядок db.execute calls в analyze_parcel (v3.7 + #33 + #29 G2):
0. UNION ALL geom + source → .mappings().first()
1. WKT query → .mappings().first()
2. District → .mappings().first()
@ -19,14 +19,15 @@ dependency_overrides, тяжёлые сервисы патчим через unit
7. Noise rows → .mappings().all()
8. Hydrology → .mappings().all()
9. Utilities → .mappings().all()
10. Market trend → .mappings().first()
11. Zoning (begin_nested) → .mappings().first()
12. Success recommendation (begin_nested) → .mappings().all()
13. Market price (begin_nested) → .mappings().first() ← NEW #33
14. Recent permits (begin_nested) → .mappings().all() ← NEW #105 Phase 5
15. _geotech_risk (industrial count) → .scalar()
16. _neighbors_summary (neighbor_rows) → .mappings().all()
17. _neighbors_summary (overlap_row) → .mappings().first()
10. parcel_meta (cad_parcels) → .mappings().first() ← #29 G2
11. Market trend → .mappings().first()
12. Zoning (begin_nested) → .mappings().first()
13. Success recommendation (begin_nested) → .mappings().all()
14. Market price (begin_nested) → .mappings().first() ← #33
15. Recent permits (begin_nested) → .mappings().all() ← #105 Phase 5
16. _geotech_risk (industrial count) → .scalar()
17. _neighbors_summary (neighbor_rows) → .mappings().all()
18. _neighbors_summary (overlap_row) → .mappings().first()
"""
from __future__ import annotations
@ -104,13 +105,15 @@ def _make_db_for_analyze(
("all", []), # 7: noise rows
("all", []), # 8: hydrology rows
("all", []), # 9: utilities rows
("first", None), # 10: market trend
("first", None), # 11: zoning (begin_nested)
("all", []), # 12: success recommendation (begin_nested)
("first", mp_mock), # 13: market price (begin_nested) ← #33
("scalar", 0), # 14: geotech_risk
("all", []), # 15: neighbors
("first", None), # 16: overlap
("first", None), # 10: parcel_meta (cad_parcels) ← #29 G2
("first", None), # 11: market trend
("first", None), # 12: zoning (begin_nested)
("all", []), # 13: success recommendation (begin_nested)
("first", mp_mock), # 14: market price (begin_nested) ← #33
("all", []), # 15: recent permits (begin_nested)
("scalar", 0), # 16: geotech_risk
("all", []), # 17: neighbors
("first", None), # 18: overlap
]
def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:

210
backend/tests/api/v1/test_analyze_parcel_meta.py Normal file
View file

@ -0,0 +1,210 @@
"""Тесты для parcel_meta в POST /api/v1/parcels/{cad_num}/analyze (#29 G2).
Покрывает:
1. analyze когда cad_parcels has row → parcel_meta содержит permitted_use/land_category/cad_cost
2. analyze когда cad_parcels row отсутствует → parcel_meta == None
Стратегия mock: аналогична test_analyze_market_price.py — DB mock через
dependency_overrides, тяжёлые сервисы патчим через unittest.mock.patch.
Порядок db.execute calls в analyze_parcel (с #29 G2):
0. UNION ALL geom + source → .mappings().first()
1. WKT query → .mappings().first()
2. District → .mappings().first()
3. POI rows → .mappings().all()
4. Competitor rows → .mappings().all()
5. Pipeline rows → .mappings().all()
6. Centroid lat/lon → .mappings().first()
7. Noise rows → .mappings().all()
8. Hydrology → .mappings().all()
9. Utilities → .mappings().all()
10. parcel_meta (cad_parcels) → .mappings().first() ← NEW #29 G2
11. Market trend → .mappings().first()
12. Zoning (begin_nested) → .mappings().first()
13. Success recommendation (begin_nested) → .mappings().all()
14. Market price (begin_nested) → .mappings().first()
15. Recent permits (begin_nested) → .mappings().all()
16. _geotech_risk (industrial count) → .scalar()
17. _neighbors_summary (neighbor_rows) → .mappings().all()
18. _neighbors_summary (overlap_row) → .mappings().first()
"""
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]]]}'
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(
parcel_meta_row: dict[str, Any] | None = None,
) -> MagicMock:
"""Сконструировать mock DB Session для analyze_parcel с фокусом на parcel_meta.
parcel_meta_row=None → имитирует "строки нет в cad_parcels" → parcel_meta=None.
parcel_meta_row={...} → имитирует найденную строку.
"""
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})
pm_mock = _make_mapping(parcel_meta_row) if parcel_meta_row is not None else None
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
("first", pm_mock), # 10: parcel_meta ← #29 G2
("first", None), # 11: market trend
("first", None), # 12: zoning (begin_nested)
("all", []), # 13: success recommendation (begin_nested)
("first", None), # 14: market price (begin_nested)
("all", []), # 15: recent permits (begin_nested)
("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
_PATCHES = [
patch("app.api.v1.parcels._fetch_air_quality_sync", return_value=None),
patch("app.api.v1.parcels._fetch_weather_sync", return_value=None),
patch("app.api.v1.parcels._fetch_seasonal_weather_sync", return_value=None),
patch(
"app.api.v1.parcels.get_quarter_dump_data",
return_value={
"nspd_zoning": None,
"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"}),
]
def _start_patches() -> None:
for p in _PATCHES:
p.start()
def _stop_patches() -> None:
for p in _PATCHES:
p.stop()
def test_parcel_meta_found_in_cad_parcels() -> None:
"""analyze → cad_parcels has row → parcel_meta содержит permitted_use/land_category/cad_cost."""
from app.core.db import get_db
pm_data: dict[str, Any] = {
"permitted_use": "многоквартирный дом",
"land_category": "Земли населённых пунктов",
"land_subtype": "жилая застройка",
"cad_cost": 5_000_000.0,
}
db = _make_db_for_analyze(parcel_meta_row=pm_data)
app.dependency_overrides[get_db] = _override_db(db)
_start_patches()
try:
client = TestClient(app)
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
assert resp.status_code == 200, resp.text
body = resp.json()
assert "parcel_meta" in body, "parcel_meta отсутствует в ответе"
pm = body["parcel_meta"]
assert pm is not None, "parcel_meta должен быть не None при наличии строки"
assert pm["permitted_use"] == "многоквартирный дом"
assert pm["land_category"] == "Земли населённых пунктов"
assert pm["land_subtype"] == "жилая застройка"
assert pm["cad_cost"] == 5_000_000.0
assert pm["source"] == "cad_parcels"
finally:
app.dependency_overrides.clear()
_stop_patches()
def test_parcel_meta_none_when_cad_parcels_missing() -> None:
"""analyze → cad_parcels row отсутствует → parcel_meta == None."""
from app.core.db import get_db
db = _make_db_for_analyze(parcel_meta_row=None)
app.dependency_overrides[get_db] = _override_db(db)
_start_patches()
try:
client = TestClient(app)
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
assert resp.status_code == 200, resp.text
body = resp.json()
assert "parcel_meta" in body, "parcel_meta ключ должен присутствовать в ответе"
assert body["parcel_meta"] is None, "parcel_meta должен быть None при отсутствии строки"
finally:
app.dependency_overrides.clear()
_stop_patches()

36
backend/tests/api/v1/test_analyze_recent_permits.py
View file

@ -5,7 +5,7 @@
2. analyze с quarter без permits → recent_permits=[], summary={rns_count=0, ...}
3. analyze 404 на invalid cad (no regression)
Порядок db.execute calls в analyze_parcel (после #105 Phase 5):
Порядок db.execute calls в analyze_parcel (после #105 Phase 5 + #29 G2):
0. UNION ALL geom + source → .mappings().first()
1. WKT query → .mappings().first()
2. District → .mappings().first()
@ -16,14 +16,15 @@
7. Noise rows → .mappings().all()
8. Hydrology → .mappings().all()
9. Utilities → .mappings().all()
10. Market trend → .mappings().first()
11. Zoning (begin_nested) → .mappings().first()
12. Success recommendation (begin_nested) → .mappings().all()
13. Market price (begin_nested) → .mappings().first()
14. Recent permits (begin_nested) → .mappings().all() ← NEW #105
15. _geotech_risk (industrial count) → .scalar()
16. _neighbors_summary (neighbor_rows) → .mappings().all()
17. _neighbors_summary (overlap_row) → .mappings().first()
10. parcel_meta (cad_parcels) → .mappings().first() ← #29 G2
11. Market trend → .mappings().first()
12. Zoning (begin_nested) → .mappings().first()
13. Success recommendation (begin_nested) → .mappings().all()
14. Market price (begin_nested) → .mappings().first()
15. Recent permits (begin_nested) → .mappings().all() ← #105
16. _geotech_risk (industrial count) → .scalar()
17. _neighbors_summary (neighbor_rows) → .mappings().all()
18. _neighbors_summary (overlap_row) → .mappings().first()
"""
from __future__ import annotations
@ -95,14 +96,15 @@ def _make_db_for_analyze(
("all", []), # 7: noise rows
("all", []), # 8: hydrology rows
("all", []), # 9: utilities rows
("first", None), # 10: market trend
("first", None), # 11: zoning (begin_nested)
("all", []), # 12: success recommendation (begin_nested)
("first", None), # 13: market price (begin_nested)
("all", raw_permits), # 14: recent permits (begin_nested) ← #105
("scalar", 0), # 15: geotech_risk
("all", []), # 16: neighbors
("first", None), # 17: overlap
("first", None), # 10: parcel_meta (cad_parcels) ← #29 G2
("first", None), # 11: market trend
("first", None), # 12: zoning (begin_nested)
("all", []), # 13: success recommendation (begin_nested)
("first", None), # 14: market price (begin_nested)
("all", raw_permits), # 15: recent permits (begin_nested) ← #105
("scalar", 0), # 16: geotech_risk
("all", []), # 17: neighbors
("first", None), # 18: overlap
]
def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:

92
backend/tests/test_gate_verdict.py
View file

@ -1,6 +1,9 @@
"""Tests for gate_verdict aggregator — pure function, no DB."""
from app.services.site_finder.gate_verdict import compute_gate_verdict, is_residential_zone
from app.services.site_finder.gate_verdict import (
compute_gate_verdict,
is_residential_zone,
)
# ── is_residential_zone unit tests ────────────────────────────────────────────
@ -115,3 +118,90 @@ def test_pzz_unknown_warning_when_no_zoning():
verdict = compute_gate_verdict(None, [], [{"name": "ТП-1"}], {"available": True})
assert any(w["code"] == "PZZ_UNKNOWN" for w in verdict["warnings"])
assert verdict["can_build_mkd"] is True
# ── cad_zouit fallback path (#232) ────────────────────────────────────────────
def test_cad_zouit_ohranaya_zona_truboprovodov_blocks():
"""cad_zouit overlap с 'охранная зона трубопроводов' → ZOUIT_CAD_BLOCKER → Нельзя."""
nspd_zoning = {"zone_code": "Ж-2", "zone_name": "Жилая"}
overlaps = [
{
"source": "cad_zouit",
"type_zone": "Охранная зона трубопроводов",
"layer": "Охранная зона трубопроводов",
"name": "Газопровод высокого давления",
}
]
verdict = compute_gate_verdict(nspd_zoning, overlaps, [], {"available": True})
assert verdict["can_build_mkd"] is False
assert verdict["verdict_label"] == "Нельзя"
assert any(b["code"] == "ZOUIT_CAD_BLOCKER" for b in verdict["blockers"])
def test_cad_zouit_elektr_blocks():
"""cad_zouit overlap с 'электр' substring в type_zone → blocker."""
nspd_zoning = {"zone_code": "Ж-3", "zone_name": "Жилая"}
overlaps = [
{
"source": "cad_zouit",
"type_zone": "Охранная зона электросетевого объекта",
"layer": "Охранная зона электросетевого объекта",
"name": "ЛЭП 110 кВ",
}
]
verdict = compute_gate_verdict(nspd_zoning, overlaps, [], {"available": True})
assert verdict["can_build_mkd"] is False
assert any(b["code"] == "ZOUIT_CAD_BLOCKER" for b in verdict["blockers"])
def test_cad_zouit_szz_is_warning_not_blocker():
"""cad_zouit overlap с 'СЗЗ' → ZOUIT_CAD_SZZ warning, не blocker."""
nspd_zoning = {"zone_code": "Ж-2", "zone_name": "Жилая"}
overlaps = [
{
"source": "cad_zouit",
"type_zone": "Санитарно-защитная зона предприятия",
"layer": "Санитарно-защитная зона предприятия",
"name": "СЗЗ завода",
}
]
verdict = compute_gate_verdict(nspd_zoning, overlaps, [{"name": "ТП-1"}], {"available": True})
assert verdict["can_build_mkd"] is True
assert verdict["verdict_label"] == "С ограничениями"
assert any(w["code"] == "ZOUIT_CAD_SZZ" for w in verdict["warnings"])
assert verdict["blockers"] == []
def test_cad_zouit_other_is_warning():
"""cad_zouit overlap без blocker/СЗЗ keyword → ZOUIT_CAD_OTHER warning."""
nspd_zoning = {"zone_code": "Ж-2", "zone_name": "Жилая"}
overlaps = [
{
"source": "cad_zouit",
"type_zone": "Зона публичного сервитута",
"layer": "Зона публичного сервитута",
"name": None,
}
]
verdict = compute_gate_verdict(nspd_zoning, overlaps, [{"name": "ТП-1"}], {"available": True})
assert verdict["can_build_mkd"] is True
assert any(w["code"] == "ZOUIT_CAD_OTHER" for w in verdict["warnings"])
assert verdict["blockers"] == []
def test_nspd_subcategory_path_backward_compat():
"""NSPD dump path (source != 'cad_zouit') продолжает работать через subcategory."""
nspd_zoning = {"zone_code": "Ж-3", "zone_name": "Жилая"}
overlaps = [
{
"source": "nspd-quarter-dump",
"subcategory": 17,
"name": "ЛЭП 110кВ",
"layer": "37578",
}
]
verdict = compute_gate_verdict(nspd_zoning, overlaps, [], {"available": True})
assert verdict["can_build_mkd"] is False
assert any(b["code"] == "ZOUIT_OVERLAP_SUB17" for b in verdict["blockers"])