"""Тесты 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