gendesign/backend/tests/workers/test_krt_geometry_sync.py

"""Тесты krt_geometry_sync (#1130) — WFS portal_geo_oks_krt → ekb_krt_geometry.

Сеть/БД не дёргаются: мокаем EKBGeoportalClient + SessionLocal, проверяем upsert-идемпотентность
(одни и те же feature_id → ON CONFLICT → UPDATE), skip фич без геометрии, graceful WFS-ошибку.
"""

from __future__ import annotations

from contextlib import contextmanager
from typing import Any

import pytest

from app.workers.tasks import krt_geometry_sync


class _Feat:
    """EKBFeature-двойник."""

    def __init__(
        self,
        feature_id: str | None,
        geometry: dict[str, Any] | None,
        properties: dict[str, Any],
    ) -> None:
        self.feature_id = feature_id
        self.geometry = geometry
        self.properties = properties


_GEOM = {
    "type": "Polygon",
    "coordinates": [[[60.6, 56.8], [60.7, 56.8], [60.7, 56.9], [60.6, 56.8]]],
}

_FEAT_1 = _Feat(
    "portal_geo_oks_krt.7",
    _GEOM,
    {"num_oks_krt": 7, "full_name": "КРТ ул. Тверитина", "status": "В работе", "area": 3.5},
)
_FEAT_2 = _Feat(
    "portal_geo_oks_krt.12",
    _GEOM,
    {"num_oks_krt": 12, "full_name": "КРТ ул. Ленина", "status": "Завершён", "area": 1.2},
)


class _FakeClient:
    def __init__(self, features: list[_Feat], *, boom: bool = False) -> None:
        self._features = features
        self._boom = boom
        self.bbox_calls: list[tuple[str, Any]] = []

    def features_in_bbox(self, layer: str, bbox: Any) -> list[_Feat]:
        self.bbox_calls.append((layer, bbox))
        if self._boom:
            raise RuntimeError("geoportal timeout")
        return self._features


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 rollback(self) -> None:
        pass

    def close(self) -> None:
        self.closed = True


def _patch(monkeypatch: Any, features: list[_Feat], *, boom: bool = False) -> _FakeDB:
    fake_client = _FakeClient(features, boom=boom)
    fake_db = _FakeDB()
    monkeypatch.setattr(krt_geometry_sync, "EKBGeoportalClient", lambda *a, **kw: fake_client)
    monkeypatch.setattr(krt_geometry_sync, "SessionLocal", lambda: fake_db)
    return fake_db


def test_upsert_two_features(monkeypatch: Any) -> None:
    """Два WFS-feature → два UPSERT-вызова; возвращает {"krt": 2}."""
    fake_db = _patch(monkeypatch, [_FEAT_1, _FEAT_2])

    result = krt_geometry_sync.sync_krt_geometry()

    assert result == {"krt": 2}
    assert fake_db.committed
    assert fake_db.closed
    # По одному execute на каждую фичу
    assert len(fake_db.executed) == 2


def test_upsert_idempotent_same_feature_id(monkeypatch: Any) -> None:
    """Одна и та же feature_id дважды → 2 execute (ON CONFLICT DO UPDATE — идемпотентно)."""
    fake_db = _patch(monkeypatch, [_FEAT_1, _FEAT_1])

    result = krt_geometry_sync.sync_krt_geometry()

    # Воркер не дедуплицирует на стороне Python — ON CONFLICT в SQL.
    assert result == {"krt": 2}
    assert len(fake_db.executed) == 2


def test_skip_feature_without_geometry(monkeypatch: Any) -> None:
    """Фича без геометрии пропускается (failed++), остальные обрабатываются."""
    no_geom = _Feat("portal_geo_oks_krt.99", None, {"num_oks_krt": 99})
    fake_db = _patch(monkeypatch, [no_geom, _FEAT_1])

    result = krt_geometry_sync.sync_krt_geometry()

    # Только _FEAT_1 дошёл до UPSERT
    assert result == {"krt": 1}
    assert len(fake_db.executed) == 1


def test_empty_wfs_returns_zero(monkeypatch: Any) -> None:
    """WFS вернул 0 фич → {"krt": 0}, никаких DB-вызовов."""
    fake_db = _patch(monkeypatch, [])

    result = krt_geometry_sync.sync_krt_geometry()

    assert result == {"krt": 0}
    assert len(fake_db.executed) == 0


def test_wfs_error_propagates(monkeypatch: Any) -> None:
    """Сбой WFS (boom=True) → Exception пробрасывается (harvest не должен глотать ошибку)."""
    _patch(monkeypatch, [], boom=True)

    with pytest.raises(RuntimeError, match="geoportal timeout"):
        krt_geometry_sync.sync_krt_geometry()


def test_upsert_params_contain_feature_id_and_geom(monkeypatch: Any) -> None:
    """Параметры UPSERT содержат feature_id, geom_json (GeoJSON), num_oks_krt."""
    fake_db = _patch(monkeypatch, [_FEAT_1])

    krt_geometry_sync.sync_krt_geometry()

    assert len(fake_db.executed) == 1
    _, params = fake_db.executed[0]
    assert params is not None
    assert params["feature_id"] == "portal_geo_oks_krt.7"
    assert params["num_oks_krt"] == 7
    assert params["status"] == "В работе"
    assert params["full_name"] == "КРТ ул. Тверитина"
    assert "geom_json" in params
    import json

    geom = json.loads(params["geom_json"])
    assert geom["type"] == "Polygon"


def test_savepoint_skips_bad_row(monkeypatch: Any) -> None:
    """UPSERT падает для одной строки (SAVEPOINT) → остальные обработаны, задача не падает."""
    call_count = 0

    class _FakeDBWithError(_FakeDB):
        @contextmanager
        def begin_nested(self) -> Any:
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise RuntimeError("constraint violation")
            yield

    fake_client = _FakeClient([_FEAT_1, _FEAT_2])
    fake_db = _FakeDBWithError()
    monkeypatch.setattr(krt_geometry_sync, "EKBGeoportalClient", lambda *a, **kw: fake_client)
    monkeypatch.setattr(krt_geometry_sync, "SessionLocal", lambda: fake_db)

    result = krt_geometry_sync.sync_krt_geometry()

    # Первая строка упала в SAVEPOINT, вторая прошла
    assert result == {"krt": 1}
    assert fake_db.committed