392 lines
16 KiB
Python
392 lines
16 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.
|
||
|
||
Mock диспетчеризует строки по СИГНАТУРЕ SQL, а не по позиционному индексу.
|
||
Это критично: weight-validation (422) в обработчике срабатывает ПОСЛЕ резолва
|
||
геометрии (parcels.py:~1213 geom → ~1382 validation), а несколько тестов шлют
|
||
ДВА POST подряд на один mock — монотонный счётчик «съедал» бы geom-row на втором
|
||
запросе → handler уходил в inline NSPD fetch (202 «fetching») вместо 422.
|
||
Сигнатурный матчинг возвращает geom/wkt/district/centroid независимо от числа
|
||
предыдущих вызовов. Прочие запросы обёрнуты в try/except SAVEPOINT → дефолтный
|
||
пустой mock их не валит.
|
||
|
||
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": "Октябрьский",
|
||
"ekb_reference_median": 120000,
|
||
"quarter_median_12m": None,
|
||
"quarter_deals_count": 0,
|
||
"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 []
|
||
|
||
def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
|
||
sql = " ".join(str(args[0]).split()) if args else ""
|
||
|
||
first_val: Any = None
|
||
all_val: list[Any] = []
|
||
|
||
if "AS geom_geojson" in sql:
|
||
first_val = geom_row
|
||
elif "AS wkt" in sql:
|
||
first_val = wkt_row
|
||
elif "AS ekb_reference_median" in sql and "district_name" in sql:
|
||
first_val = district_row
|
||
elif "AS lon" in sql and "AS lat" in sql:
|
||
first_val = centroid_row
|
||
# POI rows — категория/имя/координаты в радиусе.
|
||
elif "category" in sql and "ST_Distance" in sql and "p.geom" in sql:
|
||
all_val = _poi_rows
|
||
|
||
r = MagicMock()
|
||
r.mappings.return_value.first.return_value = first_val
|
||
r.mappings.return_value.all.return_value = all_val
|
||
r.scalar.return_value = 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.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={
|
||
"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"}),
|
||
# 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() -> 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()
|
||
|
||
|
||
def test_analyze_missing_profile_is_not_labelled_profile() -> None:
|
||
"""#2811: profile_id задан, профиль НЕ найден → метка НЕ смеет быть 'profile'.
|
||
|
||
Три способа промахнуться мимо профиля (все три воспроизведены живым запросом
|
||
на проде 2026-08-10): owner не передан вовсе, чужой профиль, удалённый id.
|
||
В mock-БД профилей нет — значит применились системные веса, и ответ обязан
|
||
это признать, а не утверждать, что считал по профилю.
|
||
"""
|
||
from app.core.db import get_db
|
||
from app.services.site_finder.weight_profiles import _SYSTEM_POI_WEIGHTS
|
||
|
||
for qs in ("profile_id=999999", "profile_id=999999&profile_user_id=nobody"):
|
||
db = _make_db_for_analyze() # профилей нет → get_profile/get_default_profile → None
|
||
app.dependency_overrides[get_db] = _override_db(db)
|
||
_start_patches()
|
||
try:
|
||
client = TestClient(app)
|
||
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze?{qs}")
|
||
assert resp.status_code == 200, resp.text
|
||
wp = resp.json()["weights_profile"]
|
||
# sanity: веса и правда системные, промах реальный
|
||
assert wp["weights_applied"]["tram_stop"] == pytest.approx(
|
||
_SYSTEM_POI_WEIGHTS["tram_stop"]
|
||
)
|
||
assert wp["source"] != "profile", (
|
||
f"?{qs}: применились системные веса, а метка source='profile' — "
|
||
"ответ утверждает то, чего не было (#2811)"
|
||
)
|
||
assert wp["source"] == "system"
|
||
# «что просили» не теряется: запрошенный id + явный признак промаха
|
||
assert wp["profile_id"] == 999999
|
||
assert wp["requested_profile_applied"] is False
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
_stop_patches()
|
||
|
||
|
||
def test_analyze_found_profile_keeps_label_and_flag() -> None:
|
||
"""Обратная сторона: профиль найден → source='profile', флаг промаха False."""
|
||
from datetime import UTC, datetime
|
||
|
||
import app.services.site_finder.weight_profiles as wp_module
|
||
from app.core.db import get_db
|
||
from app.services.site_finder.weight_profiles import WeightProfile
|
||
|
||
profile = WeightProfile(
|
||
id=7,
|
||
user_id="user-1",
|
||
profile_name="test",
|
||
weights={"tram_stop": -0.4},
|
||
is_default=False,
|
||
description=None,
|
||
created_at=datetime.now(UTC),
|
||
updated_at=datetime.now(UTC),
|
||
)
|
||
db = _make_db_for_analyze()
|
||
app.dependency_overrides[get_db] = _override_db(db)
|
||
_start_patches()
|
||
original = wp_module.get_profile
|
||
wp_module.get_profile = lambda _db, uid, pid: profile
|
||
try:
|
||
client = TestClient(app)
|
||
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze?profile_id=7&profile_user_id=user-1")
|
||
assert resp.status_code == 200, resp.text
|
||
wp = resp.json()["weights_profile"]
|
||
assert wp["source"] == "profile"
|
||
assert wp["requested_profile_applied"] is True
|
||
assert wp["weights_applied"]["tram_stop"] == pytest.approx(-0.4)
|
||
finally:
|
||
wp_module.get_profile = original
|
||
app.dependency_overrides.clear()
|
||
_stop_patches()
|