223 lines
8.4 KiB
Python
223 lines
8.4 KiB
Python
"""Тесты для 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"}),
|
||
# enable_ird_analyze=True by default — мокаем чтобы не делать сетевые вызовы в тестах
|
||
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 _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()
|