gendesign/backend/tests/test_nspd_client.py
lekss361 be143880d0 feat(scrapers): NSPD client foundation (#94) — search_by_cad + WMS + layers
Foundation для G1 #28 ПЗЗ, G2 #29 ВРИ, G3 #30 ЗОУИТ, E1 #51 cad_parcels,
I3 #44 engineering, и поддержки on-demand #93.

Backend (`app/services/scrapers/nspd_client.py`):
- `NSPDClient` с 4 core методами:
  - `search_by_cad(cad, thematic_id)` — REST /api/geoportal/v2/search
    возвращает GeoJSON + ВРИ + land_category + cost_value за 1 запрос
  - `get_feature_info(layer_id, lon, lat, buffer_m=100)` — WMS GetFeatureInfo
    на точке: какие feature'ы layer'а покрывают (lon,lat)
  - `get_features_in_bbox(layer_id, bbox_3857)` — bulk через WMS workaround
    (WFS GetCapabilities → 404, поэтому большой bbox + центральная I/J точка)
  - `list_layers(theme_id)` — каталог слоёв в теме (PKK=1, ARN=665)
- Typed responses: NSPDFeature, NSPDSearchResult, NSPDLayer (frozen dataclasses)
- LAYERS catalog: 32 layer-id с семантическими именами (territorial_zones=875838,
  zouit_engineering=37578, risk_flooding=872205, etc) — TIER 1-6 per #94
- Coordinate helpers: lonlat_to_3857(), bbox_around_point_m()
- Reuse: HEADERS + SSL ctx + fetch_geoportal из existing nspd_lite (WAF-compatible
  urllib trick). Rate limit через rate_ms.
- WAF/Rate-limit: raises NspdLiteWafError на 403/429 — caller backoff.

Tests (`backend/tests/test_nspd_client.py`): 14 unit tests, no network.
- LAYERS catalog sanity (territorial_zones=875838 закрывает G1)
- Coordinate transforms (zero, ЕКБ center, bbox)
- NSPDFeature.from_raw parsing (full / missing fields)
- NSPDSearchResult helpers (empty/first)
- _walk_layer_tree (flat / nested / empty)
- search_by_cad через monkeypatch fetch_geoportal

Также включён micro-fix из PR #95 review (cosmetic):
- parcels.py inline comment "до 25s" → "до 15s" (matched _INLINE_FETCH_WAIT_S)

Scope NOT в этом PR (отдельные follow-up issues, sequential rule):
- Schema migrations (cad_parcels_geom +columns, nspd_territorial_zones,
  nspd_zouit, nspd_red_lines, nspd_risk_zones tables) → отдельный PR
- Celery tasks (sync_territorial_zones_bbox, sync_zouit_*, sync_risk_*) →
  отдельный PR использует client
- Admin endpoints (trigger / status) → отдельный PR
- Замена on-demand fetch на nspd_client → PR after schema ready

Closes part of #94 (foundation only — sub-issues for schema/tasks).
2026-05-12 09:34:42 +03:00

197 lines
6.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.

"""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,
_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]