lekss361
e7a0af5ddc
_http_get_json/_http_post_json → verify=False (const _VERIFY_TLS): геопортал ЕКБ отдаёт цепочку с росс-гос-CA, которого нет в trust-store контейнера → verify=True падал CERTIFICATE_VERIFY_FAILED, весь geoportal-слой (C8b/D9b/#1085) не работал бы с прода. Данные публичные open-data, секреты не передаются. Прецедент: nspd_lite/ekburg_permits/pzz_loader. Прод-probe подтвердил. +регрессия-тест. Refs #1067. Co-authored-by: lekss361 <lekss361@gendsgn.local> Co-committed-by: lekss361 <lekss361@gendsgn.local>
237 lines
10 KiB
Python
237 lines
10 KiB
Python
"""Тесты EKBGeoportalClient (#1067 C7) — WFS-клиент геопортала ЕКБ.
|
||
|
||
Сеть не дёргается: мокаем module-level _http_get_json, проверяем построение CQL/params
|
||
и парсинг WFS GeoJSON. Образцы ответов — форма из live-разведки 2026-06-06
|
||
(.playwright-mcp/ekb-direct-layers-proof.json).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from app.services.scrapers.ekb_geoportal_client import (
|
||
EKB_SEARCH_BY_GEOM_URL,
|
||
EKB_URBAN_CARD_URL,
|
||
WFS_LAYERS,
|
||
EKBFeature,
|
||
EKBGeoportalClient,
|
||
parse_zone_regulation,
|
||
)
|
||
|
||
|
||
def _terzone_fc() -> dict[str, Any]:
|
||
"""WFS FeatureCollection терзоны (форма как у live-ответа Ц-1)."""
|
||
return {
|
||
"type": "FeatureCollection",
|
||
"features": [
|
||
{
|
||
"id": "urban_ter_zone.1",
|
||
"geometry": {"type": "Polygon", "coordinates": []},
|
||
"properties": {
|
||
"urban_index": "Ц-1",
|
||
"full_name": "Ц-1 Общественно-деловая зона городского центра",
|
||
"urban_kind_obj": 42,
|
||
},
|
||
}
|
||
],
|
||
}
|
||
|
||
|
||
def test_features_at_point_builds_intersects_cql(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""features_at_point резолвит typeName + строит CQL INTERSECTS(geoloc, POINT(lon lat))."""
|
||
captured: dict[str, Any] = {}
|
||
|
||
def fake_http(url: str, params: dict[str, str], **kw: Any) -> dict[str, Any]:
|
||
captured["url"] = url
|
||
captured["params"] = params
|
||
return _terzone_fc()
|
||
|
||
monkeypatch.setattr("app.services.scrapers.ekb_geoportal_client._http_get_json", fake_http)
|
||
feats = EKBGeoportalClient().features_at_point("territorial_zone", lon=60.6107, lat=56.8391)
|
||
|
||
assert len(feats) == 1
|
||
assert feats[0].properties["urban_index"] == "Ц-1"
|
||
assert feats[0].crs == "EPSG:4326"
|
||
# typeName резолвлен из WFS_LAYERS
|
||
assert captured["params"]["typeNames"] == "portal:portal_geo_urban_ter_zone_exist"
|
||
assert captured["params"]["request"] == "GetFeature"
|
||
assert captured["params"]["outputFormat"] == "application/json"
|
||
# CQL: X=lon, Y=lat
|
||
assert captured["params"]["CQL_FILTER"] == "INTERSECTS(geoloc, POINT(60.6107 56.8391))"
|
||
assert captured["url"].endswith("/api/gis/ows/3/portal/wfs")
|
||
|
||
|
||
def test_features_at_point_accepts_raw_type_name(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Сырой typeName (не в WFS_LAYERS) прокидывается as-is."""
|
||
captured: dict[str, Any] = {}
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.ekb_geoportal_client._http_get_json",
|
||
lambda url, params, **kw: captured.update(params) or {"features": []},
|
||
)
|
||
EKBGeoportalClient().features_at_point("portal:portal_geo_custom_layer", lon=60.0, lat=56.0)
|
||
assert captured["typeNames"] == "portal:portal_geo_custom_layer"
|
||
|
||
|
||
def test_features_in_bbox_builds_bbox_cql(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""features_in_bbox строит CQL BBOX с EPSG:4326."""
|
||
captured: dict[str, Any] = {}
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.ekb_geoportal_client._http_get_json",
|
||
lambda url, params, **kw: captured.update(params) or _terzone_fc(),
|
||
)
|
||
EKBGeoportalClient().features_in_bbox("krt", (60.5, 56.7, 60.7, 56.9))
|
||
assert captured["typeNames"] == "portal:portal_geo_oks_krt"
|
||
assert captured["CQL_FILTER"] == "BBOX(geoloc, 60.5, 56.7, 60.7, 56.9, 'EPSG:4326')"
|
||
|
||
|
||
def test_zone_index_at_returns_index(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""zone_index_at вытаскивает urban_index терзоны в точке (crosswalk #1059)."""
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.ekb_geoportal_client._http_get_json",
|
||
lambda url, params, **kw: _terzone_fc(),
|
||
)
|
||
assert EKBGeoportalClient().zone_index_at(60.6107, 56.8391) == "Ц-1"
|
||
|
||
|
||
def test_zone_index_at_none_when_empty(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Нет терзоны в точке → None (не падает)."""
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.ekb_geoportal_client._http_get_json",
|
||
lambda url, params, **kw: {"type": "FeatureCollection", "features": []},
|
||
)
|
||
assert EKBGeoportalClient().zone_index_at(0.0, 0.0) is None
|
||
|
||
|
||
def test_ekb_feature_from_raw() -> None:
|
||
"""EKBFeature.from_raw парсит id/geometry/properties; пустые → defaults."""
|
||
f = EKBFeature.from_raw({"id": "x.1", "geometry": {"type": "Point"}, "properties": {"k": "v"}})
|
||
assert f.feature_id == "x.1"
|
||
assert f.geometry == {"type": "Point"}
|
||
assert f.properties == {"k": "v"}
|
||
empty = EKBFeature.from_raw({})
|
||
assert empty.feature_id is None
|
||
assert empty.geometry is None
|
||
assert empty.properties == {}
|
||
|
||
|
||
def test_wfs_layers_catalog_has_key_layers() -> None:
|
||
"""Ключевые ИРД/supply-слои присутствуют в каталоге."""
|
||
for key in ("territorial_zone", "functional_zone", "krt", "ppt", "zouit"):
|
||
assert key in WFS_LAYERS
|
||
assert WFS_LAYERS[key].startswith("portal:")
|
||
|
||
|
||
# ── urbanCard (рич ПЗЗ-регламент) ────────────────────────────────────────────
|
||
|
||
|
||
def _card_fixture() -> dict[str, Any]:
|
||
"""urbanCard ответ (форма live-ответа Ц-1, vault recon)."""
|
||
return {
|
||
"pzzList": [
|
||
{
|
||
"urbanTerZone": "Ц-1 Общественно-деловая зона городского центра",
|
||
"mainTypesPermittedUse": [
|
||
{"name": "2.6 Среднеэтажная жилая застройка", "restriction": ""},
|
||
{"name": "4.1 Деловое управление", "restriction": ""},
|
||
],
|
||
"conditionTypesPermittedUse": [{"name": "4.9 Служебные гаражи"}],
|
||
"additionTypesPermittedUse": [{"name": "12.0.2 Благоустройство территории"}],
|
||
"limitTypesPermittedUse": [
|
||
{
|
||
"typePermittedUsePars": [
|
||
{"otherDescription": "минимальная площадь ЗУ - 1500 кв. м"},
|
||
{"otherDescription": ""},
|
||
]
|
||
}
|
||
],
|
||
}
|
||
]
|
||
}
|
||
|
||
|
||
def test_parse_zone_regulation() -> None:
|
||
"""parse_zone_regulation вытаскивает index + 3 списка ВРИ + предельные параметры."""
|
||
reg = parse_zone_regulation(_card_fixture())
|
||
assert reg is not None
|
||
assert reg.zone_index == "Ц-1"
|
||
assert reg.zone_full_name.startswith("Ц-1 Общественно-деловая")
|
||
assert reg.main_vri == ["2.6 Среднеэтажная жилая застройка", "4.1 Деловое управление"]
|
||
assert reg.conditional_vri == ["4.9 Служебные гаражи"]
|
||
assert reg.auxiliary_vri == ["12.0.2 Благоустройство территории"]
|
||
# пустой otherDescription отброшен → только непустой параметр
|
||
assert reg.limit_params == ["минимальная площадь ЗУ - 1500 кв. м"]
|
||
|
||
|
||
def test_parse_zone_regulation_empty() -> None:
|
||
"""Нет pzzList → None (участок без регламента)."""
|
||
assert parse_zone_regulation({"pzzList": []}) is None
|
||
assert parse_zone_regulation({}) is None
|
||
|
||
|
||
def test_zone_regulation_at_pipeline(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""zone_regulation_at: searchByGeom → urbanCard → parse (мок по URL)."""
|
||
search_resp = {
|
||
"10019": [{"typeAlias": "GknParcel", "key": 111}],
|
||
"17041": [{"typeAlias": "UrbanTerZone", "key": 222}],
|
||
}
|
||
|
||
def fake_post(url: str, body: dict[str, Any], **kw: Any) -> dict[str, Any]:
|
||
if url == EKB_SEARCH_BY_GEOM_URL:
|
||
return search_resp
|
||
if url == EKB_URBAN_CARD_URL:
|
||
assert body["pzz"] == [222] # терзона-ключ прокинут
|
||
assert body["parcel"] == [111]
|
||
return _card_fixture()
|
||
raise AssertionError(f"unexpected url {url}")
|
||
|
||
monkeypatch.setattr("app.services.scrapers.ekb_geoportal_client._http_post_json", fake_post)
|
||
reg = EKBGeoportalClient().zone_regulation_at(60.6107, 56.8391)
|
||
assert reg is not None
|
||
assert reg.zone_index == "Ц-1"
|
||
assert len(reg.main_vri) == 2
|
||
|
||
|
||
def test_zone_regulation_at_none_without_terzone(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Нет UrbanTerZone в searchByGeom → None (urbanCard не дёргается)."""
|
||
monkeypatch.setattr(
|
||
"app.services.scrapers.ekb_geoportal_client._http_post_json",
|
||
lambda url, body, **kw: {"10019": [{"typeAlias": "GknParcel", "key": 1}]},
|
||
)
|
||
assert EKBGeoportalClient().zone_regulation_at(0.0, 0.0) is None
|
||
|
||
|
||
def test_http_helpers_disable_tls_verify(monkeypatch: Any) -> None:
|
||
"""Регрессия: _http_get_json/_http_post_json должны слать verify=False.
|
||
|
||
Росс-гос-CA геопортала не в trust-store контейнера → verify=True падает
|
||
CERTIFICATE_VERIFY_FAILED на проде (см. fix #1067). Ловим молчаливый ре-брейк.
|
||
"""
|
||
from app.services.scrapers import ekb_geoportal_client as mod
|
||
|
||
captured: dict[str, Any] = {}
|
||
|
||
class _Resp:
|
||
def raise_for_status(self) -> None:
|
||
return None
|
||
|
||
def json(self) -> dict[str, Any]:
|
||
return {"ok": True}
|
||
|
||
def _fake_get(url: str, **kw: Any) -> _Resp:
|
||
captured["get_verify"] = kw.get("verify")
|
||
return _Resp()
|
||
|
||
def _fake_post(url: str, **kw: Any) -> _Resp:
|
||
captured["post_verify"] = kw.get("verify")
|
||
return _Resp()
|
||
|
||
monkeypatch.setattr(mod.httpx, "get", _fake_get)
|
||
monkeypatch.setattr(mod.httpx, "post", _fake_post)
|
||
|
||
mod._http_get_json("https://x/y", {"a": "b"})
|
||
mod._http_post_json("https://x/y", {"a": "b"})
|
||
|
||
assert captured["get_verify"] is False
|
||
assert captured["post_verify"] is False
|