NSPDClient.list_layers() парсил _walk_layer_tree по id/children, но НСПД layers-theme-tree отдаёт слои в плоском layers[] с ключом layerId → list_layers() молча возвращал [] для любой темы. Fix: сначала плоский data["layers"], иначе fallback на _walk_layer_tree; defensive None-guard на missing id. +тест на flat-форму, backward-compat сохранён. Refs #1067. Co-authored-by: lekss361 <lekss361@gendsgn.local> Co-committed-by: lekss361 <lekss361@gendsgn.local>
750 lines
27 KiB
Python
750 lines
27 KiB
Python
"""Unit tests для nspd_client (#94). Без сетевых запросов — только parsing.
|
||
|
||
Network calls (search_by_cad / get_feature_info / list_layers) проверяются
|
||
вручную через `python -c "from app.services.scrapers.nspd_client import
|
||
NSPDClient; print(NSPDClient().search_by_cad('66:41:0204016:10'))"` —
|
||
проверяется в production smoke test после deploy.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from app.services.scrapers.nspd_client import (
|
||
LAYERS,
|
||
NSPDClient,
|
||
NSPDFeature,
|
||
NSPDLayer,
|
||
NSPDSearchResult,
|
||
QuarterDump,
|
||
_geojson_bbox_3857,
|
||
_walk_layer_tree,
|
||
bbox_around_point_m,
|
||
lonlat_to_3857,
|
||
)
|
||
|
||
|
||
def test_layers_catalog_has_critical_ids() -> None:
|
||
"""Sanity check: TIER 1 critical layers зарегистрированы."""
|
||
# Из #94: TIER 1 включает 875838 (территориальные зоны / ПЗЗ G1)
|
||
assert LAYERS["territorial_zones"] == 875838
|
||
assert LAYERS["parcels"] == 36048
|
||
assert LAYERS["buildings"] == 36049
|
||
assert LAYERS["quarters"] == 36071
|
||
# TIER 2 ЗОУИТ (G3)
|
||
assert LAYERS["zouit_okn"] == 37577
|
||
assert LAYERS["zouit_engineering"] == 37578
|
||
assert LAYERS["zouit_natural"] == 37580
|
||
|
||
|
||
def test_lonlat_to_3857_zero() -> None:
|
||
"""0,0 → ~0,0 (with tiny floating error)."""
|
||
x, y = lonlat_to_3857(0.0, 0.0)
|
||
assert abs(x) < 1.0
|
||
assert abs(y) < 1.0
|
||
|
||
|
||
def test_lonlat_to_3857_ekb_center() -> None:
|
||
"""ЕКБ центр (60.605, 56.838) — known approximate projection."""
|
||
x, y = lonlat_to_3857(60.605, 56.838)
|
||
# Sanity: положительный x (восточная долгота), положительный y (северная)
|
||
assert x > 6_700_000
|
||
assert x < 6_800_000
|
||
assert y > 7_700_000
|
||
assert y < 7_800_000
|
||
|
||
|
||
def test_bbox_around_point_m_basic() -> None:
|
||
"""100m buffer вокруг точки → bbox с side ≈ 200m."""
|
||
xmin, ymin, xmax, ymax = bbox_around_point_m(60.605, 56.838, 100)
|
||
assert abs((xmax - xmin) - 200) < 1.0
|
||
assert abs((ymax - ymin) - 200) < 1.0
|
||
|
||
|
||
def test_nspd_feature_from_raw_minimal() -> None:
|
||
"""Парсинг минимального GeoJSON Feature."""
|
||
raw = {
|
||
"id": "feat-1",
|
||
"geometry": {"type": "Point", "coordinates": [1, 2]},
|
||
"properties": {"cad_num": "66:41:0204016:10"},
|
||
}
|
||
feat = NSPDFeature.from_raw(raw)
|
||
assert feat.feature_id == "feat-1"
|
||
assert feat.geometry is not None
|
||
assert feat.geometry["type"] == "Point"
|
||
assert feat.properties["cad_num"] == "66:41:0204016:10"
|
||
assert feat.crs == "EPSG:3857"
|
||
|
||
|
||
def test_nspd_feature_from_raw_missing_id() -> None:
|
||
"""Feature без id → feature_id is None."""
|
||
feat = NSPDFeature.from_raw({"geometry": None, "properties": {}})
|
||
assert feat.feature_id is None
|
||
assert feat.geometry is None
|
||
assert feat.properties == {}
|
||
|
||
|
||
def test_nspd_search_result_empty() -> None:
|
||
"""Empty result → first is None, is_empty True."""
|
||
result = NSPDSearchResult(cad_num="x", features=[], raw={})
|
||
assert result.first is None
|
||
assert result.is_empty is True
|
||
|
||
|
||
def test_nspd_search_result_nonempty() -> None:
|
||
"""First feature accessible через .first."""
|
||
feat = NSPDFeature.from_raw({"id": "x", "geometry": None, "properties": {}})
|
||
result = NSPDSearchResult(cad_num="x", features=[feat], raw={})
|
||
assert result.first is feat
|
||
assert result.is_empty is False
|
||
|
||
|
||
def test_walk_layer_tree_flat() -> None:
|
||
"""Flat list of layers — все попадают в output."""
|
||
tree: list[dict[str, Any]] = [
|
||
{"id": 1, "title": "A", "children": []},
|
||
{"id": 2, "title": "B", "children": []},
|
||
]
|
||
out = _walk_layer_tree(tree)
|
||
ids = sorted(x["id"] for x in out)
|
||
assert ids == [1, 2]
|
||
|
||
|
||
def test_walk_layer_tree_nested() -> None:
|
||
"""Nested tree — все leaves вытаскиваются."""
|
||
tree = {
|
||
"id": 1,
|
||
"title": "root",
|
||
"children": [
|
||
{"id": 2, "title": "child", "children": []},
|
||
{
|
||
"id": 3,
|
||
"title": "branch",
|
||
"children": [
|
||
{"id": 4, "title": "grandchild", "children": []},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
out = _walk_layer_tree(tree)
|
||
ids = sorted(x["id"] for x in out)
|
||
# Все уровни включены — root, child, branch, grandchild
|
||
assert ids == [1, 2, 3, 4]
|
||
|
||
|
||
def test_walk_layer_tree_empty() -> None:
|
||
"""Empty input → empty output, no crash."""
|
||
assert _walk_layer_tree({}) == []
|
||
assert _walk_layer_tree([]) == []
|
||
assert _walk_layer_tree(None) == [] # type: ignore[arg-type]
|
||
|
||
|
||
def test_nspd_client_search_by_cad_parses_response(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""search_by_cad оборачивает fetch_geoportal в NSPDSearchResult."""
|
||
fake_raw = {
|
||
"data": {
|
||
"type": "FeatureCollection",
|
||
"features": [
|
||
{
|
||
"id": "f1",
|
||
"geometry": {"type": "Polygon", "coordinates": []},
|
||
"properties": {
|
||
"cad_num": "66:41:0204016:10",
|
||
"land_record_area": 1500,
|
||
"permitted_use_established_by_document": "Многоэтажная жилая",
|
||
},
|
||
}
|
||
],
|
||
}
|
||
}
|
||
|
||
def fake_fetch(query: str, **kwargs: Any) -> dict[str, Any]:
|
||
assert query == "66:41:0204016:10"
|
||
return fake_raw
|
||
|
||
# Patch the imported reference in nspd_client (it was imported by name)
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client.fetch_geoportal",
|
||
fake_fetch,
|
||
)
|
||
|
||
client = NSPDClient()
|
||
result = client.search_by_cad("66:41:0204016:10")
|
||
assert result.cad_num == "66:41:0204016:10"
|
||
assert len(result.features) == 1
|
||
assert result.first is not None
|
||
assert result.first.properties["permitted_use_established_by_document"] == "Многоэтажная жилая"
|
||
assert result.raw is fake_raw # preserved для debugging
|
||
|
||
|
||
def test_nspd_client_search_by_cad_empty(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Empty response → is_empty True, no crash."""
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client.fetch_geoportal",
|
||
lambda *a, **kw: {"data": {"type": "FeatureCollection", "features": []}},
|
||
)
|
||
result = NSPDClient().search_by_cad("99:99:9999999")
|
||
assert result.is_empty is True
|
||
assert result.first is None
|
||
|
||
|
||
def test_nspd_layer_dataclass() -> None:
|
||
"""NSPDLayer immutable, hashable (frozen + slots)."""
|
||
layer = NSPDLayer(layer_id=875838, title="Терр.зоны", layer_type="wms", metadata={})
|
||
assert layer.layer_id == 875838
|
||
# Frozen
|
||
with pytest.raises(AttributeError):
|
||
layer.title = "x" # type: ignore[misc]
|
||
|
||
|
||
# ── Mock tests for WMS/layers methods (no network) ──────────────────────────
|
||
|
||
|
||
def test_get_feature_info_parses_response(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""get_feature_info вызывает _http_get_json с правильным URL + парсит features."""
|
||
captured_urls: list[str] = []
|
||
|
||
def fake_http(url: str, **kwargs: Any) -> dict[str, Any]:
|
||
captured_urls.append(url)
|
||
return {
|
||
"type": "FeatureCollection",
|
||
"features": [
|
||
{
|
||
"id": "tz-1",
|
||
"geometry": {"type": "Polygon", "coordinates": []},
|
||
"properties": {"zone_code": "Ж-1"},
|
||
}
|
||
],
|
||
}
|
||
|
||
monkeypatch.setattr("app.services.scrapers.nspd_client._http_get_json", fake_http)
|
||
client = NSPDClient()
|
||
feats = client.get_feature_info(875838, lon=60.605, lat=56.838, buffer_m=100)
|
||
assert len(feats) == 1
|
||
assert feats[0].properties["zone_code"] == "Ж-1"
|
||
# Verify URL structure
|
||
assert len(captured_urls) == 1
|
||
url = captured_urls[0]
|
||
assert "/api/aeggis/v4/875838/wms?" in url
|
||
assert "REQUEST=GetFeatureInfo" in url
|
||
assert "INFO_FORMAT=application%2Fjson" in url
|
||
assert "STYLES=" in url # required per NSPD WAF
|
||
assert "CRS=EPSG%3A3857" in url
|
||
# bbox should reflect 100m buffer (200m side)
|
||
assert "BBOX=" in url
|
||
|
||
|
||
def test_get_features_in_bbox_url_builds_correctly(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""get_features_in_bbox использует переданный bbox без преобразования."""
|
||
captured_urls: list[str] = []
|
||
|
||
def fake_http(url: str, **kwargs: Any) -> dict[str, Any]:
|
||
captured_urls.append(url)
|
||
return {"features": []}
|
||
|
||
monkeypatch.setattr("app.services.scrapers.nspd_client._http_get_json", fake_http)
|
||
client = NSPDClient()
|
||
bbox = (6700000.0, 7700000.0, 6800000.0, 7800000.0) # ЕКБ region in 3857
|
||
feats = client.get_features_in_bbox(875838, bbox)
|
||
assert feats == []
|
||
assert len(captured_urls) == 1
|
||
url = captured_urls[0]
|
||
# bbox passed directly into BBOX param (URL-encoded comma = %2C)
|
||
assert "BBOX=6700000.0%2C7700000.0%2C6800000.0%2C7800000.0" in url
|
||
# Large width/height для bulk fetch
|
||
assert "WIDTH=4096" in url
|
||
assert "HEIGHT=4096" in url
|
||
# I/J pointer в центр виртуального тайла
|
||
assert "I=2048" in url
|
||
assert "J=2048" in url
|
||
|
||
|
||
def test_list_layers_walks_tree_response(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""list_layers вызывает /layers-theme-tree и распаковывает в NSPDLayer."""
|
||
fake_tree = [
|
||
{
|
||
"id": 1,
|
||
"title": "PKK тема",
|
||
"children": [
|
||
{
|
||
"id": 875838,
|
||
"title": "Территориальные зоны",
|
||
"layerType": "wms",
|
||
"children": [],
|
||
},
|
||
{
|
||
"id": 37577,
|
||
"title": "ЗОУИТ ОКН",
|
||
"layerType": "wms",
|
||
"children": [],
|
||
},
|
||
],
|
||
}
|
||
]
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client._http_get_json",
|
||
lambda url, **kw: fake_tree,
|
||
)
|
||
client = NSPDClient()
|
||
layers = client.list_layers(theme_id=1)
|
||
layer_ids = [layer.layer_id for layer in layers]
|
||
# Root + 2 children = 3 leaves
|
||
assert 875838 in layer_ids
|
||
assert 37577 in layer_ids
|
||
# Sample layer metadata preserved
|
||
tz_layer = next(layer for layer in layers if layer.layer_id == 875838)
|
||
assert tz_layer.title == "Территориальные зоны"
|
||
assert tz_layer.layer_type == "wms"
|
||
|
||
|
||
def test_list_layers_handles_data_wrapper(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Shape defense — NSPD иногда оборачивает в {"data": [...]}."""
|
||
fake_wrapped = {
|
||
"data": [{"id": 36048, "title": "ЗУ", "children": []}],
|
||
"meta": {"count": 1},
|
||
}
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client._http_get_json",
|
||
lambda url, **kw: fake_wrapped,
|
||
)
|
||
layers = NSPDClient().list_layers(theme_id=1)
|
||
assert len(layers) >= 1
|
||
assert any(layer.layer_id == 36048 for layer in layers)
|
||
|
||
|
||
def test_list_layers_handles_garbage_response(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Не dict/list response → empty list + warning log (no crash)."""
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client._http_get_json",
|
||
lambda url, **kw: "garbage string",
|
||
)
|
||
assert NSPDClient().list_layers(theme_id=1) == []
|
||
|
||
|
||
def test_list_layers_parses_flat_layers_response(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Актуальная форма NSPD: {"layers": [...flat...], "tree": ...} с ключом layerId.
|
||
|
||
Регрессия на баг: walker искал id/children, а реальный ответ держит слои в
|
||
плоском layers[] (layerId/title/layerName/layerType) — list_layers молча
|
||
возвращал [] для любой темы (см.
|
||
research/NSPD_IRD_Layers_Automation_Recon_Jun06).
|
||
"""
|
||
fake_resp = {
|
||
"layers": [
|
||
{
|
||
"layerId": 36945,
|
||
"title": "Кадастровые округа",
|
||
"layerName": "ЕГРН. Округа",
|
||
"layerType": "wms",
|
||
},
|
||
{
|
||
"layerId": 875838,
|
||
"title": "Территориальные зоны",
|
||
"layerName": "ПЗЗ",
|
||
"layerType": "wms",
|
||
},
|
||
{
|
||
"layerId": 37577,
|
||
"title": "ЗОУИТ ОКН",
|
||
"layerName": "ЗОУИТ ОКН",
|
||
"layerType": "wms",
|
||
},
|
||
],
|
||
"tree": {"categories": []},
|
||
}
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client._http_get_json",
|
||
lambda url, **kw: fake_resp,
|
||
)
|
||
layers = NSPDClient().list_layers(theme_id=1)
|
||
assert sorted(layer.layer_id for layer in layers) == [36945, 37577, 875838]
|
||
tz = next(layer for layer in layers if layer.layer_id == 875838)
|
||
assert tz.title == "Территориальные зоны"
|
||
assert tz.layer_type == "wms"
|
||
assert tz.metadata["layerName"] == "ПЗЗ"
|
||
|
||
|
||
# ── _geojson_bbox_3857 tests ──────────────────────────────────────────────────
|
||
|
||
|
||
def test_geojson_bbox_3857_polygon() -> None:
|
||
"""Простой Polygon → bbox охватывает все vertices."""
|
||
geom = {
|
||
"type": "Polygon",
|
||
"coordinates": [[[0, 0], [100, 0], [100, 100], [0, 100], [0, 0]]],
|
||
}
|
||
result = _geojson_bbox_3857(geom)
|
||
assert result == (0.0, 0.0, 100.0, 100.0)
|
||
|
||
|
||
def test_geojson_bbox_3857_multipolygon() -> None:
|
||
"""MultiPolygon с двумя кусками → bbox охватывает оба."""
|
||
geom = {
|
||
"type": "MultiPolygon",
|
||
"coordinates": [
|
||
[[[0, 0], [10, 0], [10, 10], [0, 10], [0, 0]]],
|
||
[[[20, 30], [50, 30], [50, 60], [20, 60], [20, 30]]],
|
||
],
|
||
}
|
||
result = _geojson_bbox_3857(geom)
|
||
assert result is not None
|
||
xmin, ymin, xmax, ymax = result
|
||
assert xmin == 0.0
|
||
assert ymin == 0.0
|
||
assert xmax == 50.0
|
||
assert ymax == 60.0
|
||
|
||
|
||
def test_geojson_bbox_3857_linestring() -> None:
|
||
"""LineString из 2 точек → bbox прямоугольник вокруг них."""
|
||
geom = {
|
||
"type": "LineString",
|
||
"coordinates": [[10, 20], [30, 40]],
|
||
}
|
||
result = _geojson_bbox_3857(geom)
|
||
assert result == (10.0, 20.0, 30.0, 40.0)
|
||
|
||
|
||
def test_geojson_bbox_3857_point() -> None:
|
||
"""Point → bbox с шириной и высотой 0 (один пиксель)."""
|
||
geom = {
|
||
"type": "Point",
|
||
"coordinates": [60.0, 56.0],
|
||
}
|
||
result = _geojson_bbox_3857(geom)
|
||
assert result == (60.0, 56.0, 60.0, 56.0)
|
||
|
||
|
||
def test_geojson_bbox_3857_empty() -> None:
|
||
"""Polygon с пустым coordinates → None."""
|
||
geom = {"type": "Polygon", "coordinates": []}
|
||
assert _geojson_bbox_3857(geom) is None
|
||
|
||
|
||
def test_geojson_bbox_3857_no_type() -> None:
|
||
"""Нет поля type → None (не можем определить геометрию)."""
|
||
geom: dict[str, Any] = {"coordinates": [[0, 0]]}
|
||
assert _geojson_bbox_3857(geom) is None
|
||
|
||
|
||
# ── QuarterDump tests ─────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_feat(fid: str = "x") -> NSPDFeature:
|
||
"""Вспомогательная фабрика NSPDFeature для тестов."""
|
||
return NSPDFeature.from_raw({"id": fid, "geometry": None, "properties": {}})
|
||
|
||
|
||
def test_quarter_dump_total_features() -> None:
|
||
"""total_features корректно суммирует features по всем layers."""
|
||
dump = QuarterDump(
|
||
quarter_cad="66:41:0204016",
|
||
quarter=_make_feat("q"),
|
||
parcels=[_make_feat("p1"), _make_feat("p2"), _make_feat("p3")], # 3
|
||
buildings=[_make_feat("b1")], # 1
|
||
territorial_zones=[_make_feat("tz1"), _make_feat("tz2")], # 2
|
||
red_lines=[], # 0
|
||
engineering_structures=[_make_feat("e1")], # 1
|
||
zouit={
|
||
"okn": [_make_feat("ok1"), _make_feat("ok2")], # 2
|
||
"natural": [_make_feat("nat1")], # 1
|
||
},
|
||
risks={
|
||
"flooding": [_make_feat("fl1"), _make_feat("fl2"), _make_feat("fl3")], # 3
|
||
"landslide": [_make_feat("ls1")], # 1
|
||
},
|
||
opportunity={}, # 0
|
||
layers_fetched=("search", "parcels", "buildings"),
|
||
bbox_3857=(6700000.0, 7700000.0, 6800000.0, 7800000.0),
|
||
fetched_at_utc="2026-05-12T00:00:00+00:00",
|
||
)
|
||
# 3+1+2+0+1 (core) + 2+1 (zouit) + 3+1 (risks) = 14
|
||
assert dump.total_features == 14
|
||
|
||
|
||
def test_quarter_dump_frozen() -> None:
|
||
"""QuarterDump frozen=True: попытка assign → AttributeError."""
|
||
dump = QuarterDump(
|
||
quarter_cad="66:41:0204016",
|
||
quarter=None,
|
||
parcels=[],
|
||
buildings=[],
|
||
territorial_zones=[],
|
||
red_lines=[],
|
||
engineering_structures=[],
|
||
zouit={},
|
||
risks={},
|
||
opportunity={},
|
||
layers_fetched=("search",),
|
||
bbox_3857=None,
|
||
fetched_at_utc="2026-05-12T00:00:00+00:00",
|
||
)
|
||
with pytest.raises(AttributeError):
|
||
dump.parcels = [] # type: ignore[misc]
|
||
|
||
|
||
# ── search_by_quarter mock tests ──────────────────────────────────────────────
|
||
|
||
# Квартал в районе ЕКБ (EPSG:3857 метры)
|
||
_QUARTER_COORDS = [
|
||
[6700000.0, 7700000.0],
|
||
[6800000.0, 7700000.0],
|
||
[6800000.0, 7800000.0],
|
||
[6700000.0, 7800000.0],
|
||
[6700000.0, 7700000.0],
|
||
]
|
||
_QUARTER_BBOX = (6700000.0, 7700000.0, 6800000.0, 7800000.0)
|
||
|
||
_FAKE_QUARTER_SEARCH = {
|
||
"data": {
|
||
"type": "FeatureCollection",
|
||
"features": [
|
||
{
|
||
"id": "q-1",
|
||
"geometry": {"type": "Polygon", "coordinates": [_QUARTER_COORDS]},
|
||
"properties": {"cad_num": "66:41:0204016"},
|
||
}
|
||
],
|
||
}
|
||
}
|
||
|
||
_LAYER_FEATURE_COUNTS: dict[str, int] = {
|
||
"parcels": 4,
|
||
"buildings": 2,
|
||
"territorial_zones": 1,
|
||
"red_lines": 0,
|
||
"engineering_structures": 3,
|
||
}
|
||
|
||
|
||
def _make_fake_http(
|
||
layer_feature_counts: dict[str, int] | None = None,
|
||
) -> Any:
|
||
"""Возвращает fake _http_get_json который генерирует N features по layer ID.
|
||
|
||
layer_feature_counts: {layer_name: count} — если None, возвращает 1 feature.
|
||
"""
|
||
counts = layer_feature_counts or {}
|
||
# Строим reverse-map layer_id → count для подстановки в URL
|
||
from app.services.scrapers.nspd_client import LAYERS as _LAYERS
|
||
|
||
id_to_name: dict[int, str] = {v: k for k, v in _LAYERS.items()}
|
||
|
||
def fake_http(url: str, **kwargs: Any) -> dict[str, Any]:
|
||
# Извлекаем layer_id из URL вида /api/aeggis/v4/{id}/wms?...
|
||
parts = url.split("/")
|
||
layer_id_str = parts[6] if len(parts) > 6 else "0"
|
||
try:
|
||
layer_id = int(layer_id_str.split("?")[0])
|
||
except ValueError:
|
||
layer_id = 0
|
||
layer_name = id_to_name.get(layer_id, "unknown")
|
||
n = counts.get(layer_name, 1)
|
||
return {
|
||
"type": "FeatureCollection",
|
||
"features": [
|
||
{"id": f"{layer_name}-{i}", "geometry": None, "properties": {}} for i in range(n)
|
||
],
|
||
}
|
||
|
||
return fake_http
|
||
|
||
|
||
def _make_fake_grid_walk(
|
||
layer_feature_counts: dict[str, int] | None = None,
|
||
) -> Any:
|
||
"""Возвращает fake `NSPDClient.get_features_in_bbox_grid`.
|
||
|
||
После Sub-PR B (#260) area/linear layers (territorial_zones, red_lines,
|
||
engineering_structures, zouit_*, risk_*) идут через grid-walk вместо
|
||
legacy `_http_get_json` → тесты обязаны мокать оба пути, иначе грид-walk
|
||
бьёт по живому NSPD API.
|
||
"""
|
||
counts = layer_feature_counts or {}
|
||
from app.services.scrapers.nspd_client import LAYERS as _LAYERS
|
||
from app.services.scrapers.nspd_client import NSPDFeature
|
||
|
||
id_to_name: dict[int, str] = {v: k for k, v in _LAYERS.items()}
|
||
|
||
def fake_grid(
|
||
self: Any,
|
||
layer_id: int,
|
||
bbox: tuple[float, float, float, float],
|
||
*,
|
||
grid_n: int = 7,
|
||
step_m: float = 50.0,
|
||
) -> list[NSPDFeature]:
|
||
layer_name = id_to_name.get(layer_id, "unknown")
|
||
n = counts.get(layer_name, 1)
|
||
return [
|
||
NSPDFeature.from_raw({"id": f"{layer_name}-{i}", "geometry": None, "properties": {}})
|
||
for i in range(n)
|
||
]
|
||
|
||
return fake_grid
|
||
|
||
|
||
def test_search_by_quarter_core_only(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""core_only (include_zouit=False, include_risks=False): 1 search + 5 bulk."""
|
||
search_calls: list[str] = []
|
||
|
||
def fake_fetch(query: str, **kwargs: Any) -> dict[str, Any]:
|
||
search_calls.append(query)
|
||
return _FAKE_QUARTER_SEARCH
|
||
|
||
monkeypatch.setattr("app.services.scrapers.nspd_client.fetch_geoportal", fake_fetch)
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client._http_get_json",
|
||
_make_fake_http(_LAYER_FEATURE_COUNTS),
|
||
)
|
||
# Sub-PR B (#260): area layers идут через grid-walk — мокаем оба пути.
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid",
|
||
_make_fake_grid_walk(_LAYER_FEATURE_COUNTS),
|
||
)
|
||
|
||
result = NSPDClient().search_by_quarter(
|
||
"66:41:0204016", include_zouit=False, include_risks=False
|
||
)
|
||
|
||
# Quarter populated
|
||
assert result.quarter is not None
|
||
assert result.quarter.feature_id == "q-1"
|
||
|
||
# Core layer lengths match mock counts
|
||
assert len(result.parcels) == 4
|
||
assert len(result.buildings) == 2
|
||
assert len(result.territorial_zones) == 1
|
||
assert len(result.red_lines) == 0
|
||
assert len(result.engineering_structures) == 3
|
||
|
||
# ЗОУИТ / risks пусты
|
||
assert result.zouit == {}
|
||
assert result.risks == {}
|
||
|
||
# layers_fetched: search + 5 core (red_lines тоже: _fetch_layer вызывается,
|
||
# но возвращает пустой список — имя всё равно добавляется). tuple, not list
|
||
# — frozen-dataclass immutability invariant.
|
||
assert result.layers_fetched == (
|
||
"search",
|
||
"parcels",
|
||
"buildings",
|
||
"territorial_zones",
|
||
"red_lines",
|
||
"engineering_structures",
|
||
)
|
||
|
||
# bbox вычислен из четырёхугольного polygon
|
||
assert result.bbox_3857 == _QUARTER_BBOX
|
||
|
||
|
||
def test_search_by_quarter_with_zouit(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""include_zouit=True → zouit dict содержит 5 ключей."""
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client.fetch_geoportal",
|
||
lambda *a, **kw: _FAKE_QUARTER_SEARCH,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client._http_get_json",
|
||
_make_fake_http(),
|
||
)
|
||
# Sub-PR B (#260): zouit_* и area layers идут через grid-walk.
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid",
|
||
_make_fake_grid_walk(),
|
||
)
|
||
|
||
result = NSPDClient().search_by_quarter(
|
||
"66:41:0204016", include_zouit=True, include_risks=False
|
||
)
|
||
|
||
assert set(result.zouit.keys()) == {"okn", "engineering", "natural", "protected", "other"}
|
||
assert result.risks == {}
|
||
|
||
# layers_fetched содержит search + 5 core + 5 zouit = 11 записей
|
||
assert len(result.layers_fetched) == 11
|
||
|
||
|
||
def test_search_by_quarter_empty_quarter(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Пустой FeatureCollection → quarter=None, все layer feature lists пустые.
|
||
|
||
При bbox=None `_fetch_layer` возвращает [] без HTTP вызовов. ЗОУИТ-словарь
|
||
всё равно создаётся (с пустыми списками) — ключи присутствуют, features нет.
|
||
"""
|
||
http_calls: list[str] = []
|
||
|
||
def fake_http(url: str, **kwargs: Any) -> dict[str, Any]:
|
||
http_calls.append(url)
|
||
return {"type": "FeatureCollection", "features": []}
|
||
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client.fetch_geoportal",
|
||
lambda *a, **kw: {"data": {"type": "FeatureCollection", "features": []}},
|
||
)
|
||
monkeypatch.setattr("app.services.scrapers.nspd_client._http_get_json", fake_http)
|
||
|
||
result = NSPDClient().search_by_quarter("66:41:9999999", include_zouit=True)
|
||
|
||
assert result.quarter is None
|
||
assert result.parcels == []
|
||
assert result.buildings == []
|
||
assert result.territorial_zones == []
|
||
assert result.red_lines == []
|
||
assert result.engineering_structures == []
|
||
|
||
# include_zouit=True: ключи созданы, но все списки пустые (bbox=None → _fetch_layer → [])
|
||
assert set(result.zouit.keys()) == {"okn", "engineering", "natural", "protected", "other"}
|
||
assert all(v == [] for v in result.zouit.values())
|
||
|
||
assert result.risks == {}
|
||
assert result.bbox_3857 is None
|
||
|
||
# Никаких bulk HTTP запросов — bbox=None ⇒ ранний выход из _fetch_layer
|
||
assert http_calls == []
|
||
# layers_fetched содержит только "search" (tuple, no bulk fetches happened)
|
||
assert result.layers_fetched == ("search",)
|
||
|
||
|
||
def test_search_by_quarter_layers_fetched_with_risks(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""include_risks=True (без zouit) → layers_fetched включает 11 risk layer имён."""
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client.fetch_geoportal",
|
||
lambda *a, **kw: _FAKE_QUARTER_SEARCH,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client._http_get_json",
|
||
_make_fake_http(),
|
||
)
|
||
# Sub-PR B (#260): risk_* и area layers идут через grid-walk.
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid",
|
||
_make_fake_grid_walk(),
|
||
)
|
||
|
||
result = NSPDClient().search_by_quarter(
|
||
"66:41:0204016", include_zouit=False, include_risks=True
|
||
)
|
||
|
||
assert result.zouit == {}
|
||
assert set(result.risks.keys()) == {
|
||
"flooding_underground",
|
||
"flooding",
|
||
"swampification",
|
||
"landslide",
|
||
"abrasion",
|
||
"erosion_water",
|
||
"erosion_linear",
|
||
"erosion_wind",
|
||
"desertification",
|
||
"clutter",
|
||
"burns",
|
||
}
|
||
|
||
# layers_fetched: search + 5 core + 11 risks = 17
|
||
assert len(result.layers_fetched) == 17
|
||
# Убедимся что risk имена действительно присутствуют в layers_fetched
|
||
assert "risk_flooding" in result.layers_fetched
|
||
assert "risk_landslide" in result.layers_fetched
|
||
assert "risk_burns" in result.layers_fetched
|