gendesign/backend/tests/api/v1/test_parcel_utility_infrastructure.py
Light1YT 42ed1f8a93
All checks were successful
CI / changes (pull_request) Successful in 6s
CI / frontend-tests (pull_request) Successful in 54s
CI / openapi-codegen-check (pull_request) Successful in 2m12s
CI / backend-tests (pull_request) Successful in 11m19s
feat(site-finder): OSM engineering-networks loader + endpoint (#1746)
No-B2B open-data путь (Overpass) для инженерных сетей ЕКБ: ЛЭП/подстанции/
опоры, трубопроводы (газ/вода/канализация), ЦТП, вышки связи.

- Миграция 100_osm_utility_infrastructure_ekb.sql: таблица + GIST-индекс,
  UNIQUE(osm_type, osm_id) для идемпотентного UPSERT, geom(Geometry,4326)
  (линии И точки).
- Loader utility_infrastructure_loader.py: Overpass-запросы (rate-limit 1 req/s),
  классификация в infrastructure_kind (power/water/gas/heat/communication/sewage),
  per-element SAVEPOINT UPSERT. Зеркалит noise_loader (геом-хелперы переиспользованы).
- Celery task + weekly beat (понедельник 04:00 МСК, после noise-sync).
- Endpoint GET /api/v1/parcels/{cad}/utility-infrastructure: инж.сети в радиусе
  центроида участка (ST_DWithin) + summary по видам сети. Зеркалит connection-points.
- Тесты: loader (Overpass-парсинг/UPSERT/SAVEPOINT/фильтры) + endpoint (mock DB).
- Codegen: api-types.ts регенерирован (новый endpoint + 3 схемы).
2026-06-27 00:42:02 +05:00

136 lines
4.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Тесты для GET /{cad_num}/utility-infrastructure (#1746).
FastAPI TestClient + mock сервиса get_nearby_utility_infrastructure — без Postgres.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import patch
import pytest
from fastapi.testclient import TestClient
from app.main import app
_VALID_CAD = "66:41:0204016:10"
def _make_feature(
osm_id: int = 1,
kind: str = "power",
distance_m: float = 120.5,
osm_type: str = "way",
) -> dict[str, Any]:
return {
"osm_id": osm_id,
"osm_type": osm_type,
"infrastructure_kind": kind,
"name": "ЛЭП-110",
"source_tag": "power=line",
"distance_m": distance_m,
"geometry_geojson": {"type": "LineString", "coordinates": [[60.6, 56.8], [60.61, 56.81]]},
}
def _make_response(
features: list[dict[str, Any]] | None = None,
nearest: float | None = 120.5,
nearest_by_kind: dict[str, float | None] | None = None,
) -> dict[str, Any]:
if features is None:
features = [_make_feature()]
if nearest_by_kind is None:
nearest_by_kind = {"power": 120.5}
return {
"features": features,
"summary": {
"total_features": len(features),
"nearest_distance_m": nearest,
"nearest_by_kind": nearest_by_kind,
},
}
def test_utility_infra_returns_features() -> None:
"""Нормальный ответ: фичи + summary с ближайшими по видам сети."""
full = _make_response(
features=[
_make_feature(1, "power", 80.0, "way"),
_make_feature(2, "water", 150.0, "node"),
],
nearest=80.0,
nearest_by_kind={"power": 80.0, "water": 150.0},
)
with patch(
"app.api.v1.parcels.get_nearby_utility_infrastructure",
return_value=full,
):
client = TestClient(app)
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/utility-infrastructure")
assert r.status_code == 200, r.text
body = r.json()
assert body["summary"]["total_features"] == 2
assert body["summary"]["nearest_distance_m"] == pytest.approx(80.0)
assert body["summary"]["nearest_by_kind"]["power"] == pytest.approx(80.0)
assert body["summary"]["nearest_by_kind"]["water"] == pytest.approx(150.0)
assert len(body["features"]) == 2
assert body["features"][0]["infrastructure_kind"] == "power"
assert body["features"][0]["geometry_geojson"]["type"] == "LineString"
def test_utility_infra_empty() -> None:
"""Нет сетей в радиусе → пустой список, total=0, nearest=null."""
empty = _make_response(features=[], nearest=None, nearest_by_kind={})
with patch(
"app.api.v1.parcels.get_nearby_utility_infrastructure",
return_value=empty,
):
client = TestClient(app)
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/utility-infrastructure")
assert r.status_code == 200, r.text
body = r.json()
assert body["features"] == []
assert body["summary"]["total_features"] == 0
assert body["summary"]["nearest_distance_m"] is None
assert body["summary"]["nearest_by_kind"] == {}
def test_utility_infra_parcel_not_found_404() -> None:
"""Участок не найден (ValueError из сервиса) → 404."""
with patch(
"app.api.v1.parcels.get_nearby_utility_infrastructure",
side_effect=ValueError("Участок '66:41:9999999:1' не найден в БД"),
):
client = TestClient(app)
r = client.get("/api/v1/parcels/66:41:9999999:1/utility-infrastructure")
assert r.status_code == 404
assert "не найден" in r.json()["detail"]
def test_utility_infra_passes_radius() -> None:
"""radius_m из query передаётся в сервис."""
with patch("app.api.v1.parcels.get_nearby_utility_infrastructure") as mock_svc:
mock_svc.return_value = _make_response()
client = TestClient(app)
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/utility-infrastructure?radius_m=1000")
assert r.status_code == 200
call = mock_svc.call_args
assert call[0][2] == 1000 or call[1].get("radius_m") == 1000
def test_utility_infra_radius_out_of_range_422() -> None:
"""radius_m=10 (< min 50) → 422."""
client = TestClient(app)
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/utility-infrastructure?radius_m=10")
assert r.status_code == 422
def test_utility_infra_radius_too_large_422() -> None:
"""radius_m=5000 (> max 2000) → 422."""
client = TestClient(app)
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/utility-infrastructure?radius_m=5000")
assert r.status_code == 422