110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""Тесты refresh_zone_regulations (#1059/C8b) — прогрев кэша регламента зон ЕКБ.
|
|
|
|
Сеть/БД не дёргаются: мокаем EKBGeoportalClient + SessionLocal + upsert_zone_regulation.
|
|
Проверяем дедуп по urban_index, представительную точку, skip пустых/битых, счётчики.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.services.scrapers.ekb_geoportal_client import ZoneRegulation
|
|
from app.workers.tasks import zone_regulation_refresh as zrr
|
|
|
|
|
|
class _Feat:
|
|
def __init__(self, geometry: dict[str, Any] | None, properties: dict[str, Any]) -> None:
|
|
self.geometry = geometry
|
|
self.properties = properties
|
|
|
|
|
|
def _poly() -> dict[str, Any]:
|
|
return {"type": "Polygon", "coordinates": [[[60, 56], [60.1, 56], [60.1, 56.1], [60, 56]]]}
|
|
|
|
|
|
def _reg(idx: str) -> ZoneRegulation:
|
|
return ZoneRegulation(
|
|
zone_index=idx,
|
|
zone_full_name=f"{idx} зона",
|
|
main_vri=[],
|
|
conditional_vri=[],
|
|
auxiliary_vri=[],
|
|
limit_params=[],
|
|
)
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, feats: list[_Feat]) -> None:
|
|
self._feats = feats
|
|
self.resolved: list[tuple[float, float]] = []
|
|
|
|
def features_in_bbox(self, layer: str, bbox: Any) -> list[_Feat]:
|
|
return self._feats
|
|
|
|
def zone_regulation_at(self, lon: float, lat: float) -> ZoneRegulation:
|
|
self.resolved.append((lon, lat))
|
|
return _reg("Ц-1")
|
|
|
|
|
|
class _DB:
|
|
def __init__(self) -> None:
|
|
self.committed = False
|
|
self.closed = False
|
|
|
|
def commit(self) -> None:
|
|
self.committed = True
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def _patch(monkeypatch: Any, feats: list[_Feat]) -> tuple[_FakeClient, list[Any]]:
|
|
client = _FakeClient(feats)
|
|
upserts: list[Any] = []
|
|
monkeypatch.setattr(zrr, "EKBGeoportalClient", lambda *a, **kw: client)
|
|
monkeypatch.setattr(zrr, "SessionLocal", lambda: _DB())
|
|
monkeypatch.setattr(
|
|
zrr, "upsert_zone_regulation", lambda db, reg, **kw: upserts.append(reg) or {}
|
|
)
|
|
return client, upserts
|
|
|
|
|
|
def test_dedup_by_urban_index(monkeypatch: Any) -> None:
|
|
"""Две фичи одного индекса → резолвим/кэшируем зону один раз."""
|
|
feats = [
|
|
_Feat(_poly(), {"urban_index": "Ц-1"}),
|
|
_Feat(_poly(), {"urban_index": "Ц-1"}),
|
|
_Feat(_poly(), {"urban_index": "Ж-2"}),
|
|
]
|
|
client, upserts = _patch(monkeypatch, feats)
|
|
|
|
result = zrr.refresh_zone_regulations()
|
|
|
|
assert result == {"zones": 2, "cached": 2}
|
|
assert len(client.resolved) == 2 # уникальных индексов
|
|
assert len(upserts) == 2
|
|
|
|
|
|
def test_skips_blank_index_and_bad_geometry(monkeypatch: Any) -> None:
|
|
feats = [
|
|
_Feat(_poly(), {"urban_index": ""}), # пустой индекс
|
|
_Feat(None, {"urban_index": "Ц-3"}), # нет геометрии
|
|
]
|
|
client, upserts = _patch(monkeypatch, feats)
|
|
|
|
result = zrr.refresh_zone_regulations()
|
|
|
|
assert result == {"zones": 1, "cached": 0} # Ц-3 в seen, но geom=None → не резолвится
|
|
assert client.resolved == []
|
|
assert upserts == []
|
|
|
|
|
|
def test_representative_point_inside_polygon() -> None:
|
|
lon, lat = zrr._representative_point(_poly()) # type: ignore[misc]
|
|
assert 60.0 <= lon <= 60.1
|
|
assert 56.0 <= lat <= 56.1
|
|
|
|
|
|
def test_representative_point_none_on_bad_geom() -> None:
|
|
assert zrr._representative_point(None) is None
|
|
assert zrr._representative_point({"type": "Polygon", "coordinates": "garbage"}) is None
|