lekss361
3921180d07
Some checks failed
Co-authored-by: lekss361 <lekss361@gendsgn.local> Co-committed-by: lekss361 <lekss361@gendsgn.local>
213 lines
8 KiB
Python
213 lines
8 KiB
Python
"""Тесты sync_functional_zones (#1058) — ГИСОГД-СО WFS → ekb_genplan_functional_zone.
|
||
|
||
Сеть/БД не дёргаются: мокаем fetch_functional_zones + SessionLocal, проверяем
|
||
маппинг properties → колонки, globalid-дедуп, upsert-идемпотентность.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import contextmanager
|
||
from typing import Any
|
||
|
||
from app.workers.tasks import genplan_zones_sync
|
||
|
||
# ── Фейковые данные ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_feature(
|
||
globalid: str = "aaaa-1111",
|
||
full_name: str | None = "Жилые зоны",
|
||
classid: str | None = "Ж",
|
||
status: str | None = "Существующий",
|
||
area_ha: float | None = 12.5,
|
||
doc_ref: str | None = "REF-001",
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"geometry": {
|
||
"type": "Polygon",
|
||
"coordinates": [[[60.5, 56.8], [60.6, 56.8], [60.5, 56.8]]],
|
||
},
|
||
"properties": {
|
||
"urban10_globalid": globalid,
|
||
"full_name": full_name,
|
||
"urban10_classid_v": "Жилые зоны (fallback)",
|
||
"urban10_classid": classid,
|
||
"urban10_status_v": status,
|
||
"urban10_area": area_ha,
|
||
"info_set_key_ref": doc_ref,
|
||
},
|
||
}
|
||
|
||
|
||
# ── Фейковая БД ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
class _FakeDB:
|
||
def __init__(self) -> None:
|
||
self.executed: list[tuple[Any, dict[str, Any] | None]] = []
|
||
self.committed = False
|
||
self.closed = False
|
||
|
||
@contextmanager
|
||
def begin_nested(self) -> Any:
|
||
yield
|
||
|
||
def execute(self, sql: Any, params: dict[str, Any] | None = None) -> Any:
|
||
self.executed.append((sql, params))
|
||
return None
|
||
|
||
def commit(self) -> None:
|
||
self.committed = True
|
||
|
||
def close(self) -> None:
|
||
self.closed = True
|
||
|
||
|
||
# ── Тесты ────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_sync_maps_properties_to_columns(monkeypatch: Any) -> None:
|
||
"""Проверяем маппинг properties → параметры UPSERT."""
|
||
feat = _make_feature(
|
||
globalid="uuid-abc",
|
||
full_name="Жилые зоны",
|
||
classid="Ж",
|
||
status="Существующий",
|
||
area_ha=10.5,
|
||
doc_ref="DOC-777",
|
||
)
|
||
db = _FakeDB()
|
||
monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat])
|
||
monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
|
||
|
||
result = genplan_zones_sync.sync_functional_zones()
|
||
|
||
assert result == {"zones": 1}
|
||
assert db.committed is True
|
||
assert db.closed is True
|
||
assert len(db.executed) == 1
|
||
_, params = db.executed[0]
|
||
assert params is not None
|
||
assert params["globalid"] == "uuid-abc"
|
||
assert params["zone_type"] == "Жилые зоны"
|
||
assert params["zone_class_code"] == "Ж"
|
||
assert params["status"] == "Существующий"
|
||
assert params["area_ha"] == 10.5
|
||
assert params["doc_ref"] == "DOC-777"
|
||
assert isinstance(params["geojson"], str)
|
||
assert isinstance(params["raw_props"], str)
|
||
|
||
|
||
def test_sync_fallback_to_classid_v_when_no_full_name(monkeypatch: Any) -> None:
|
||
"""full_name=None → zone_type берётся из urban10_classid_v."""
|
||
feat = _make_feature(full_name=None)
|
||
db = _FakeDB()
|
||
monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat])
|
||
monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
|
||
|
||
genplan_zones_sync.sync_functional_zones()
|
||
|
||
_, params = db.executed[0]
|
||
assert params is not None
|
||
assert params["zone_type"] == "Жилые зоны (fallback)"
|
||
|
||
|
||
def test_sync_skips_feature_without_globalid(monkeypatch: Any) -> None:
|
||
"""Фича без globalid → пропускается (defensive), не вызывает UPSERT."""
|
||
feat: dict[str, Any] = {
|
||
"geometry": {"type": "Polygon", "coordinates": []},
|
||
"properties": {"full_name": "Жилые зоны"}, # нет urban10_globalid
|
||
}
|
||
db = _FakeDB()
|
||
monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat])
|
||
monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
|
||
|
||
result = genplan_zones_sync.sync_functional_zones()
|
||
|
||
assert result == {"zones": 0}
|
||
assert db.executed == []
|
||
assert db.committed is True
|
||
|
||
|
||
def test_sync_skips_feature_without_geometry(monkeypatch: Any) -> None:
|
||
"""Фича без geometry → пропускается."""
|
||
feat: dict[str, Any] = {
|
||
"geometry": None,
|
||
"properties": {
|
||
"urban10_globalid": "uuid-xyz",
|
||
"full_name": "Производственные зоны",
|
||
},
|
||
}
|
||
db = _FakeDB()
|
||
monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat])
|
||
monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
|
||
|
||
result = genplan_zones_sync.sync_functional_zones()
|
||
|
||
assert result == {"zones": 0}
|
||
assert db.executed == []
|
||
|
||
|
||
def test_sync_globalid_dedup_two_identical_features(monkeypatch: Any) -> None:
|
||
"""Две фичи с одинаковым globalid — оба UPSERT (ON CONFLICT дедуп на уровне БД)."""
|
||
feat = _make_feature(globalid="dup-uuid")
|
||
db = _FakeDB()
|
||
monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat, feat])
|
||
monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
|
||
|
||
result = genplan_zones_sync.sync_functional_zones()
|
||
|
||
# Оба вызова UPSERT прошли (ON CONFLICT DO UPDATE разрулит на уровне БД).
|
||
assert result == {"zones": 2}
|
||
assert len(db.executed) == 2
|
||
|
||
|
||
def test_sync_savepoint_per_row_error_does_not_abort_batch(monkeypatch: Any) -> None:
|
||
"""SAVEPOINT per-row: ошибка в одной фиче не валит батч остальных."""
|
||
call_count = 0
|
||
|
||
class _FaultyDB(_FakeDB):
|
||
@contextmanager
|
||
def begin_nested(self) -> Any:
|
||
nonlocal call_count
|
||
call_count += 1
|
||
if call_count == 1:
|
||
raise Exception("upsert error row 1")
|
||
yield
|
||
|
||
features = [_make_feature(globalid=f"uuid-{i}") for i in range(3)]
|
||
db = _FaultyDB()
|
||
monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: features)
|
||
monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
|
||
|
||
result = genplan_zones_sync.sync_functional_zones()
|
||
|
||
# Строка 1 упала (savepoint), строки 2 и 3 записались.
|
||
assert result == {"zones": 2}
|
||
assert db.committed is True
|
||
assert db.closed is True
|
||
|
||
|
||
def test_sync_empty_wfs_response(monkeypatch: Any) -> None:
|
||
"""WFS вернул 0 фич → zones=0, commit без ошибок."""
|
||
db = _FakeDB()
|
||
monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [])
|
||
monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
|
||
|
||
result = genplan_zones_sync.sync_functional_zones()
|
||
|
||
assert result == {"zones": 0}
|
||
assert db.committed is True
|
||
assert db.closed is True
|
||
|
||
|
||
def test_upsert_sql_wraps_geom_with_st_makevalid() -> None:
|
||
"""_UPSERT_SQL содержит ST_MakeValid вокруг ST_SetSRID (#1156).
|
||
|
||
Лёгкая проверка без PostGIS: строковое представление compiled SQL
|
||
должно включать ST_MakeValid — гарантирует что backfill-логика
|
||
присутствует в тексте запроса.
|
||
"""
|
||
sql_text = str(genplan_zones_sync._UPSERT_SQL)
|
||
assert "ST_MakeValid" in sql_text
|
||
assert "ST_SetSRID" in sql_text
|