329 lines
12 KiB
Python
329 lines
12 KiB
Python
"""Тесты для inline POI-weights в POST /api/v1/parcels/{cad_num}/analyze (#201).
|
||
|
||
Покрывает:
|
||
1. POST /analyze без body → system defaults (no regression)
|
||
2. POST /analyze с inline weights → applied (source = "inline")
|
||
3. POST /analyze с невалидной категорией → 422
|
||
4. POST /analyze с весом вне диапазона → 422
|
||
5. POST /analyze с body.weights + profile_id → body.weights wins (priority)
|
||
|
||
Стратегия mock: DB патчим через dependency_overrides, тяжёлые service-функции
|
||
(weather, velocity, dump и т.д.) патчим через unittest.mock.patch — чтобы не
|
||
дублировать все 18 db.execute call'ов в каждом тесте.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
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]]]}'
|
||
|
||
|
||
# ── Mock factories ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_mapping(data: dict[str, Any]) -> MagicMock:
|
||
"""Создать mock-строку (mapping) с __getitem__ + .get() для dict-like доступа."""
|
||
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(
|
||
geom_found: bool = True,
|
||
district_found: bool = True,
|
||
poi_rows: list[Any] | None = None,
|
||
) -> MagicMock:
|
||
"""Сконструировать mock DB Session для analyze_parcel.
|
||
|
||
Порядок db.execute calls в analyze_parcel:
|
||
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. 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()
|
||
|
||
begin_nested() — возвращаем context manager чтобы поддержать `with` statement.
|
||
"""
|
||
db = MagicMock()
|
||
|
||
geom_row = (
|
||
_make_mapping({"geom_geojson": _GEOJSON, "geom_wkb": None, "source": "cad_quarter"})
|
||
if geom_found
|
||
else None
|
||
)
|
||
wkt_row = _make_mapping({"wkt": _WKT}) if geom_found else None
|
||
district_row = (
|
||
_make_mapping(
|
||
{
|
||
"district_name": "Октябрьский",
|
||
"median_price_per_m2": 120000,
|
||
"dist_to_center": 1500.0,
|
||
}
|
||
)
|
||
if district_found
|
||
else None
|
||
)
|
||
centroid_row = _make_mapping({"lat": 56.84, "lon": 60.605})
|
||
|
||
_poi_rows = poi_rows or []
|
||
|
||
# Счётчик вызовов execute — разводим first() / all() / scalar() по очерёдности
|
||
call_idx = [0]
|
||
# Ответы в порядке вызовов:
|
||
responses: list[Any] = [
|
||
("first", geom_row), # 0: geom UNION ALL
|
||
("first", wkt_row), # 1: WKT
|
||
("first", district_row), # 2: district
|
||
("all", _poi_rows), # 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", 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
|
||
]
|
||
|
||
def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
|
||
idx = call_idx[0]
|
||
call_idx[0] += 1
|
||
if idx >= len(responses):
|
||
# Безопасный fallback для непредусмотренных вызовов
|
||
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
|
||
|
||
# begin_nested() → context manager, остальные execute внутри него проходят
|
||
# через тот же side_effect (because db.execute is the same mock).
|
||
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
|
||
|
||
|
||
# Патчим тяжёлые внешние вызовы (weather / velocity / nspd-dump),
|
||
# чтобы тесты не зависели от сети и не требовали полного mock DB.
|
||
_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_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() -> list[Any]:
|
||
started = [p.start() for p in _PATCHES]
|
||
return started
|
||
|
||
|
||
def _stop_patches() -> None:
|
||
for p in _PATCHES:
|
||
p.stop()
|
||
|
||
|
||
# ── Тесты ─────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_analyze_no_body_uses_system_defaults() -> None:
|
||
"""POST /analyze без body → source = 'system', нет регрессии."""
|
||
from app.core.db import get_db
|
||
|
||
db = _make_db_for_analyze()
|
||
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 "weights_profile" in body
|
||
assert body["weights_profile"]["source"] == "system"
|
||
assert body["weights_profile"]["inline_weights"] is None
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
_stop_patches()
|
||
|
||
|
||
def test_analyze_inline_weights_applied() -> None:
|
||
"""POST /analyze с body.weights → source = 'inline', веса применены."""
|
||
from app.core.db import get_db
|
||
|
||
db = _make_db_for_analyze()
|
||
app.dependency_overrides[get_db] = _override_db(db)
|
||
_start_patches()
|
||
try:
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
f"/api/v1/parcels/{_CAD}/analyze",
|
||
json={"weights": {"kindergarten": 2.5}},
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
wp = body["weights_profile"]
|
||
assert wp["source"] == "inline"
|
||
assert wp["inline_weights"] == {"kindergarten": 2.5}
|
||
# applied weights содержат inline override поверх defaults
|
||
assert wp["weights_applied"]["kindergarten"] == pytest.approx(2.5)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
_stop_patches()
|
||
|
||
|
||
def test_analyze_invalid_category_returns_422() -> None:
|
||
"""POST /analyze с невалидной POI-категорией → 422."""
|
||
from app.core.db import get_db
|
||
|
||
db = _make_db_for_analyze()
|
||
app.dependency_overrides[get_db] = _override_db(db)
|
||
_start_patches()
|
||
try:
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
f"/api/v1/parcels/{_CAD}/analyze",
|
||
json={"weights": {"nonexistent_category": 1.0}},
|
||
)
|
||
assert resp.status_code == 422, resp.text
|
||
detail = resp.json()["detail"]
|
||
assert "nonexistent_category" in detail
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
_stop_patches()
|
||
|
||
|
||
def test_analyze_weight_out_of_range_returns_422() -> None:
|
||
"""POST /analyze с весом вне [-2, 3] → 422."""
|
||
from app.core.db import get_db
|
||
|
||
db = _make_db_for_analyze()
|
||
app.dependency_overrides[get_db] = _override_db(db)
|
||
_start_patches()
|
||
try:
|
||
client = TestClient(app)
|
||
# Слишком большой вес
|
||
resp = client.post(
|
||
f"/api/v1/parcels/{_CAD}/analyze",
|
||
json={"weights": {"school": 99.9}},
|
||
)
|
||
assert resp.status_code == 422, resp.text
|
||
detail = resp.json()["detail"]
|
||
assert "school" in detail
|
||
|
||
# Слишком маленький вес
|
||
resp2 = client.post(
|
||
f"/api/v1/parcels/{_CAD}/analyze",
|
||
json={"weights": {"park": -5.0}},
|
||
)
|
||
assert resp2.status_code == 422, resp2.text
|
||
assert "park" in resp2.json()["detail"]
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
_stop_patches()
|
||
|
||
|
||
def test_inline_weights_rejects_nan() -> None:
|
||
"""NaN weight должен вернуть 422, а не propagate в score."""
|
||
from app.core.db import get_db
|
||
|
||
db = _make_db_for_analyze()
|
||
app.dependency_overrides[get_db] = _override_db(db)
|
||
_start_patches()
|
||
try:
|
||
client = TestClient(app)
|
||
# Отправляем raw JSON с NaN — httpx.Client не умеет encode float('nan'),
|
||
# поэтому используем content= с явным bytes-телом.
|
||
raw_body = b'{"weights": {"school": NaN}}'
|
||
resp = client.post(
|
||
f"/api/v1/parcels/{_CAD}/analyze",
|
||
content=raw_body,
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
assert (
|
||
resp.status_code == 422
|
||
), f"Ожидали 422 для NaN-weight, получили {resp.status_code}: {resp.text}"
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
_stop_patches()
|
||
|
||
|
||
def test_analyze_inline_weights_beats_profile_id() -> None:
|
||
"""body.weights + profile_id → body.weights имеет приоритет (source = 'inline')."""
|
||
from app.core.db import get_db
|
||
|
||
db = _make_db_for_analyze()
|
||
app.dependency_overrides[get_db] = _override_db(db)
|
||
_start_patches()
|
||
try:
|
||
client = TestClient(app)
|
||
# Передаём и profile_id=1, и inline weights — inline должен победить
|
||
resp = client.post(
|
||
f"/api/v1/parcels/{_CAD}/analyze?profile_id=1",
|
||
json={"weights": {"metro_stop": 2.0}},
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
wp = resp.json()["weights_profile"]
|
||
assert wp["source"] == "inline", f"Ожидали source='inline', получили '{wp['source']}'"
|
||
assert wp["weights_applied"]["metro_stop"] == pytest.approx(2.0)
|
||
# profile_id всё ещё присутствует в ответе для трассировки
|
||
assert wp["profile_id"] == 1
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
_stop_patches()
|