gendesign/backend/tests/services/test_cadastre_bulk.py

"""Тесты для bulk_harvest.py и grid_geometry.py.

Использует mock DB (через unittest.mock) и mock NSPDBulkClient.
Не требует реального PostgreSQL или NSPD сети.
"""

from __future__ import annotations

from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from app.schemas.nspd_bulk import NSPDBulkFeature, QuarterSnapshot
from app.services.cadastre.bulk_harvest import (
    CAT_ENK,
    CAT_QUARTER_STATS,
    CAT_ZOUIT,
    _derive_quarter_cad,
    _safe_int,
    _safe_numeric,
    upsert_features,
    upsert_quarter_geom_from_feature,
)
from app.services.cadastre.grid_geometry import (
    generate_grid_click_points,
    quarter_bbox_3857,
)

# ── Фикстуры ─────────────────────────────────────────────────────────────────


def _make_parcel_feature(cad_num: str = "66:41:0303161:1") -> NSPDBulkFeature:
    """Создать NSPDBulkFeature для ЗУ (category_id=36368)."""
    return NSPDBulkFeature.model_validate(
        {
            "id": 1,
            "type": "Feature",
            "geometry": {
                "type": "Polygon",
                "coordinates": [
                    [
                        [6090000, 7590000],
                        [6090100, 7590000],
                        [6090100, 7590100],
                        [6090000, 7590100],
                        [6090000, 7590000],
                    ]
                ],
            },
            "properties": {
                "category": 36368,
                "options": {
                    "cad_num": cad_num,
                    "quarter_cad_number": "66:41:0303161",
                    "land_record_area": 1234.56,
                    "cost_value": 5000000.0,
                    "readable_address": "г Екатеринбург, ул Тестовая",
                    "status": "Учтённый",
                    "ownership_type": "частная",
                },
            },
        }
    )


def _make_building_feature(cad_num: str = "66:41:0303161:1:1") -> NSPDBulkFeature:
    """Создать NSPDBulkFeature для здания (category_id=36369)."""
    return NSPDBulkFeature.model_validate(
        {
            "id": 2,
            "type": "Feature",
            "geometry": {
                "type": "Polygon",
                "coordinates": [
                    [
                        [6090000, 7590000],
                        [6090050, 7590000],
                        [6090050, 7590050],
                        [6090000, 7590050],
                        [6090000, 7590000],
                    ]
                ],
            },
            "properties": {
                "category": 36369,
                "options": {
                    "cad_num": cad_num,
                    "quarter_cad_number": "66:41:0303161",
                    "purpose": "Жилое",
                    "floors": "9",
                    "area": 5000.0,
                    "cost_value": 20000000.0,
                    "readable_address": "г Екатеринбург, ул Тестовая, д 1",
                },
            },
        }
    )


def _make_zouit_feature(reg_numb: str = "66-0.1-1.1-1234") -> NSPDBulkFeature:
    """Создать NSPDBulkFeature для ЗОУИТ (category_id=36940)."""
    return NSPDBulkFeature.model_validate(
        {
            "id": 3,
            "type": "Feature",
            "geometry": {
                "type": "MultiPolygon",
                "coordinates": [
                    [
                        [
                            [6090000, 7590000],
                            [6091000, 7590000],
                            [6091000, 7591000],
                            [6090000, 7591000],
                            [6090000, 7590000],
                        ]
                    ]
                ],
            },
            "properties": {
                "category": 36940,
                "categoryName": "ЗОУИТ",
                "options": {
                    "reg_numb_border": reg_numb,
                    "type_zone": "Охранная зона ЛЭП",
                    "registration_date": "2020-01-01",
                },
            },
        }
    )


# ── Вспомогательные функции ───────────────────────────────────────────────────


def test_derive_quarter_cad_4_segments() -> None:
    assert _derive_quarter_cad("66:41:0303161:1") == "66:41:0303161"


def test_derive_quarter_cad_5_segments() -> None:
    assert _derive_quarter_cad("66:41:0303161:1:1") == "66:41:0303161"


def test_derive_quarter_cad_3_segments_unchanged() -> None:
    assert _derive_quarter_cad("66:41:0303161") == "66:41:0303161"


def test_safe_numeric_valid() -> None:
    assert _safe_numeric("1234.56") == pytest.approx(1234.56)
    assert _safe_numeric(100) == pytest.approx(100.0)


def test_safe_numeric_none_returns_none() -> None:
    assert _safe_numeric(None) is None


def test_safe_numeric_invalid_returns_none() -> None:
    assert _safe_numeric("не число") is None


def test_safe_int_valid() -> None:
    assert _safe_int("9") == 9
    assert _safe_int(5.7) == 5


def test_safe_int_none_returns_none() -> None:
    assert _safe_int(None) is None


# ── Grid geometry ────────────────────────────────────────────────────────────


def test_generate_grid_click_points_count() -> None:
    """15×15 grid → 225 точек."""
    bbox = (6090000.0, 7590000.0, 6090750.0, 7590750.0)
    points = generate_grid_click_points(bbox, grid_size=15)
    assert len(points) == 225


def test_generate_grid_click_points_bbox_coverage() -> None:
    """Все sub-bbox покрывают исходный bbox без перекрытий."""
    bbox = (0.0, 0.0, 1500.0, 1500.0)
    points = generate_grid_click_points(bbox, grid_size=3)
    # 3×3 = 9 ячеек
    assert len(points) == 9
    # Первая ячейка начинается с 0,0
    first_bbox, first_click = points[0]
    assert first_bbox[0] == pytest.approx(0.0)
    assert first_bbox[1] == pytest.approx(0.0)
    # click_xy в центре тайла (256, 256) для tile_size=512
    assert first_click == (256, 256)


def test_generate_grid_click_points_custom_grid() -> None:
    bbox = (0.0, 0.0, 1000.0, 1000.0)
    points = generate_grid_click_points(bbox, grid_size=5, tile_size=256)
    assert len(points) == 25
    # Каждая ячейка имеет правильный размер 200м×200м
    cell_bbox, _ = points[0]
    assert cell_bbox[2] - cell_bbox[0] == pytest.approx(200.0)
    assert cell_bbox[3] - cell_bbox[1] == pytest.approx(200.0)


# ── quarter_bbox_3857 sanity check ───────────────────────────────────────────


def _mock_db_with_bbox(
    xmin: float | None, ymin: float | None, xmax: float | None, ymax: float | None
) -> MagicMock:
    """Mock SQLAlchemy session.execute().mappings().first() возвращает заданный bbox."""
    mock_db = MagicMock()
    row = None if xmin is None else {"xmin": xmin, "ymin": ymin, "xmax": xmax, "ymax": ymax}
    mock_db.execute.return_value.mappings.return_value.first.return_value = row
    return mock_db


def test_quarter_bbox_3857_healthy_geom() -> None:
    """Real ЕКБ-квартал ~750м — должен пройти sanity check."""
    db = _mock_db_with_bbox(6735845.0, 8329000.0, 6736595.0, 8329750.0)
    bbox = quarter_bbox_3857(db, "66:41:0303161")
    assert bbox is not None
    assert bbox == (6735845.0, 8329000.0, 6736595.0, 8329750.0)


def test_quarter_bbox_3857_skips_broken_micro_geom() -> None:
    """width < 100m — broken micro-precision geom (72% rows в cad_quarters_geom)."""
    # width=60.7m — реальный broken quarter 66:41:0101005 после ST_Transform
    db = _mock_db_with_bbox(60.0, 60.0, 120.7, 120.7)
    assert quarter_bbox_3857(db, "66:41:0101005") is None


def test_quarter_bbox_3857_skips_oversized_geom() -> None:
    """width > 10000m — broken/regional geom (не городской квартал)."""
    db = _mock_db_with_bbox(0.0, 0.0, 15000.0, 15000.0)
    assert quarter_bbox_3857(db, "66:41:9999999") is None


def test_quarter_bbox_3857_skips_zero_height() -> None:
    """height < 100m — симметрия по другой оси."""
    db = _mock_db_with_bbox(0.0, 0.0, 500.0, 50.0)
    assert quarter_bbox_3857(db, "66:41:0000001") is None


def test_quarter_bbox_3857_returns_none_when_quarter_missing() -> None:
    """Квартал не найден в cad_quarters_geom → None (xmin IS NULL)."""
    db = _mock_db_with_bbox(None, None, None, None)
    assert quarter_bbox_3857(db, "66:99:0000000") is None


# ── upsert_features routing ──────────────────────────────────────────────────


def test_upsert_features_routes_parcels_and_buildings() -> None:
    """upsert_features должен вызывать upsert_parcel и upsert_building."""
    parcel = _make_parcel_feature()
    building = _make_building_feature()

    db = MagicMock()
    db.execute = MagicMock(return_value=MagicMock())
    db.begin_nested = MagicMock(return_value=MagicMock(__enter__=MagicMock(), __exit__=MagicMock()))

    with (
        patch("app.services.cadastre.bulk_harvest.upsert_parcel") as mock_parcel,
        patch("app.services.cadastre.bulk_harvest.upsert_building") as mock_building,
    ):
        counts = upsert_features(db, [parcel, building], source="search")

    mock_parcel.assert_called_once_with(db, parcel, source="search")
    mock_building.assert_called_once_with(db, building, source="search")
    assert counts["parcels"] == 1
    assert counts["buildings"] == 1


def test_upsert_features_routes_zouit() -> None:
    """ЗОУИТ (36940) маршрутируется в upsert_zouit."""
    zouit = _make_zouit_feature()
    db = MagicMock()

    with patch("app.services.cadastre.bulk_harvest.upsert_zouit") as mock_zouit:
        counts = upsert_features(db, [zouit], source="search")

    mock_zouit.assert_called_once()
    assert counts["zouit"] == 1


def test_upsert_features_unknown_category_counts_as_skipped() -> None:
    """Неизвестный category_id → skipped, не raise."""
    unknown_feature = NSPDBulkFeature.model_validate(
        {
            "type": "Feature",
            "geometry": None,
            "properties": {"category": 99999, "options": {}},
        }
    )
    db = MagicMock()
    counts = upsert_features(db, [unknown_feature], source="search")
    assert counts["skipped"] == 1


def test_upsert_features_quarter_stats_not_counted() -> None:
    """category_id=36381 (quarter stats) — не считается ни в одну таблицу."""
    stats_feature = NSPDBulkFeature.model_validate(
        {
            "type": "Feature",
            "geometry": None,
            "properties": {"category": 36381, "options": {"cnt_land": 5}},
        }
    )
    db = MagicMock()
    counts = upsert_features(db, [stats_feature], source="search")
    # Все счётчики должны быть 0, skipped тоже (36381 — особый случай)
    assert counts["parcels"] == 0
    assert counts["buildings"] == 0
    assert counts["skipped"] == 0


# ── upsert_parcel SQL params ─────────────────────────────────────────────────


def test_upsert_parcel_uses_cast_not_double_colon() -> None:
    """Проверить что SQL использует CAST(:x AS type) а не :x::type (SQLAlchemy ловушка).

    Используем inspect самого SQL через db.execute calls.
    """
    from app.services.cadastre.bulk_harvest import upsert_parcel

    db = MagicMock()
    executed_sqls: list[str] = []

    def capture_execute(stmt: Any, params: Any = None) -> MagicMock:
        executed_sqls.append(str(stmt))
        return MagicMock()

    db.execute = capture_execute

    feature = _make_parcel_feature()
    upsert_parcel(db, feature, source="search")

    assert len(executed_sqls) == 1
    sql = executed_sqls[0]
    # Проверяем что нет `:x::type` паттерна (только CAST(:x AS ...))
    assert "::jsonb" not in sql, "SQL содержит ::jsonb — нужно CAST(:x AS jsonb)"
    assert "::date" not in sql, "SQL содержит ::date — нужно CAST(:x AS date)"
    assert "CAST(:raw_props AS jsonb)" in sql


# ── harvest_quarter integration (async, mocked) ───────────────────────────────


@pytest.mark.asyncio
async def test_harvest_quarter_phase_done_early_return() -> None:
    """Если phase_state='done' — harvest_quarter возвращается ранним выходом."""
    from app.services.cadastre.bulk_harvest import harvest_quarter

    db = MagicMock()
    db.execute = MagicMock(
        return_value=MagicMock(
            mappings=MagicMock(
                return_value=MagicMock(
                    first=MagicMock(return_value={"phase_state": {"phase": "done"}})
                )
            )
        )
    )

    client = AsyncMock()
    progress_calls: list[Any] = []

    await harvest_quarter(
        db=db,
        client=client,
        quarter="66:41:0303161",
        job_id=1,
        update_progress=lambda s: progress_calls.append(s),
    )

    # client.search_by_quarter не должен вызываться
    client.search_by_quarter.assert_not_called()
    # Нет progress calls
    assert len(progress_calls) == 0


@pytest.mark.asyncio
async def test_harvest_quarter_calls_upsert_features() -> None:
    """harvest_quarter Phase 1: search_by_quarter → upsert_features."""
    from app.services.cadastre.bulk_harvest import harvest_quarter

    parcel = _make_parcel_feature()
    snapshot = QuarterSnapshot(
        quarter="66:41:0303161",
        fetched_at="2026-05-15T10:00:00+00:00",
        features=[parcel],
        meta_counts={36368: 5, 36369: 3},  # нет overflow
    )

    db = MagicMock()
    # phase_state = None → начинаем с нуля
    db.execute = MagicMock(
        return_value=MagicMock(
            mappings=MagicMock(
                return_value=MagicMock(first=MagicMock(return_value={"phase_state": None}))
            )
        )
    )
    db.commit = MagicMock()

    client = AsyncMock()
    client.search_by_quarter = AsyncMock(return_value=snapshot)

    progress_states: list[dict[str, Any]] = []

    with patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert:
        mock_upsert.return_value = {
            "parcels": 1,
            "buildings": 0,
            "constructions": 0,
            "oncs": 0,
            "enks": 0,
            "zouit": 0,
            "skipped": 0,
        }
        result = await harvest_quarter(
            db=db,
            client=client,
            quarter="66:41:0303161",
            job_id=1,
            update_progress=lambda s: progress_states.append(s),
        )

    client.search_by_quarter.assert_awaited_once_with("66:41:0303161")
    mock_upsert.assert_called()
    assert result.parcels_upserted == 1

    # Проверить что последний progress = done
    assert progress_states[-1]["phase"] == "done"


# ── Dedup тест для grid-walk ─────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_grid_walk_dedup_same_cad_num() -> None:
    """Если wms_feature_info возвращает одинаковый cad_num дважды — 1 upsert."""
    from app.services.cadastre.bulk_harvest import _grid_walk_category

    parcel = _make_parcel_feature("66:41:0303161:1")
    # Одинаковый feature для каждой grid-точки

    db = MagicMock()
    db.execute = MagicMock(
        return_value=MagicMock(
            mappings=MagicMock(
                return_value=MagicMock(
                    first=MagicMock(
                        return_value={
                            "xmin": 6090000.0,
                            "ymin": 7590000.0,
                            "xmax": 6090750.0,
                            "ymax": 7590750.0,
                        }
                    )
                )
            )
        )
    )
    db.begin_nested = MagicMock(
        return_value=MagicMock(__enter__=MagicMock(return_value=None), __exit__=MagicMock())
    )
    db.commit = MagicMock()

    client = AsyncMock()
    # Каждый wms_feature_info вызов возвращает один и тот же парцел
    client.wms_feature_info = AsyncMock(return_value=[parcel])

    with patch("app.services.cadastre.bulk_harvest.quarter_bbox_3857") as mock_bbox:
        mock_bbox.return_value = (6090000.0, 7590000.0, 6090750.0, 7590750.0)
        with patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert:
            mock_upsert.return_value = {
                "parcels": 1,
                "buildings": 0,
                "constructions": 0,
                "oncs": 0,
                "enks": 0,
                "zouit": 0,
                "skipped": 0,
            }

            # Используем grid_size=2 для скорости теста (4 ячейки)
            upserted, requests = await _grid_walk_category(
                db=db,
                client=client,
                quarter="66:41:0303161",
                layer_id=36368,
                grid_size=2,
            )

    # Несмотря на 4 запроса, upsert_features вызывается 1 раз (дедуп по cad_num)
    assert mock_upsert.call_count == 1
    assert upserted == 1
    assert requests == 4  # 2×2 grid


# ── ЗОУИТ dedup по (reg_numb_border, category_id) ───────────────────────────


def test_zouit_dedup_different_category_ids() -> None:
    """Один reg_numb_border с разными category_id → 2 записи в upsert."""
    zouit_1 = NSPDBulkFeature.model_validate(
        {
            "type": "Feature",
            "geometry": None,
            "properties": {
                "category": 36940,
                "categoryName": "ЗОУИТ 1",
                "options": {"reg_numb_border": "66-0.1-1.1-1234", "type_zone": "тип А"},
            },
        }
    )
    zouit_2 = NSPDBulkFeature.model_validate(
        {
            "type": "Feature",
            "geometry": None,
            "properties": {
                "category": 469039,
                "categoryName": "ЗОУИТ 2",
                "options": {"reg_numb_border": "66-0.1-1.1-1234", "type_zone": "тип Б"},
            },
        }
    )

    db = MagicMock()
    calls: list[Any] = []
    db.execute = MagicMock(
        side_effect=lambda stmt, params=None: calls.append(params) or MagicMock()
    )

    with patch("app.services.cadastre.bulk_harvest.upsert_zouit") as mock_zouit:
        counts = upsert_features(db, [zouit_1, zouit_2], source="search")

    # Обе записи маршрутируются в upsert_zouit
    assert mock_zouit.call_count == 2
    assert counts["zouit"] == 2


# ── per-category snapshot (Phase 1.5) ────────────────────────────────────────


@pytest.mark.asyncio
async def test_harvest_quarter_calls_per_cat_probe_for_zouit_when_meta_nonzero() -> None:
    """Phase 1.5: если в meta snapshot есть ЗОУИТ (36940), делаем per-cat probe."""
    from app.services.cadastre.bulk_harvest import harvest_quarter

    zouit_cat = next(iter(CAT_ZOUIT))  # один из ЗОУИТ categoryId

    # Snapshot Phase 1: 1 парцел + meta с ЗОУИТ
    parcel = _make_parcel_feature()
    zouit_feature = _make_zouit_feature()
    base_snapshot = QuarterSnapshot(
        quarter="66:41:0303161",
        fetched_at="2026-05-15T10:00:00+00:00",
        features=[parcel],
        meta_counts={36368: 2, zouit_cat: 5},  # ЗОУИТ totalCount=5 > 0
    )
    # Per-category probe snapshot для ЗОУИТ
    cat_snapshot = QuarterSnapshot(
        quarter="66:41:0303161",
        fetched_at="2026-05-15T10:00:01+00:00",
        features=[zouit_feature],
        meta_counts={zouit_cat: 5},
    )

    db = MagicMock()
    db.execute = MagicMock(
        return_value=MagicMock(
            mappings=MagicMock(
                return_value=MagicMock(first=MagicMock(return_value={"phase_state": None}))
            )
        )
    )
    db.commit = MagicMock()

    client = AsyncMock()
    # Первый вызов (без category_id) → базовый snapshot
    # Последующие — per-category probes
    client.search_by_quarter = AsyncMock(side_effect=[base_snapshot, cat_snapshot])

    with patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert:
        mock_upsert.return_value = {
            "parcels": 0,
            "buildings": 0,
            "constructions": 0,
            "oncs": 0,
            "enks": 0,
            "zouit": 1,
            "skipped": 0,
        }
        result = await harvest_quarter(
            db=db,
            client=client,
            quarter="66:41:0303161",
            job_id=1,
            update_progress=lambda s: None,
        )

    # search_by_quarter должен быть вызван минимум дважды:
    # 1) без category_id (базовый)
    # 2) с category_id=zouit_cat (per-cat probe)
    assert client.search_by_quarter.await_count >= 2
    call_args_list = client.search_by_quarter.await_args_list
    # Первый вызов без category_id
    first_call_kwargs = call_args_list[0].kwargs
    assert first_call_kwargs.get("category_id") is None
    # Один из последующих вызовов с category_id ∈ CAT_ZOUIT
    per_cat_calls = [c for c in call_args_list[1:] if c.kwargs.get("category_id") in CAT_ZOUIT]
    assert len(per_cat_calls) >= 1, "Ожидался per-cat probe для ЗОУИТ"
    assert result.zouit_upserted >= 1


@pytest.mark.asyncio
async def test_harvest_quarter_skips_per_cat_probe_when_meta_zero() -> None:
    """Phase 1.5: если meta не содержит ЗОУИТ/ЕНК — per-cat probe не делается."""
    from app.services.cadastre.bulk_harvest import harvest_quarter

    parcel = _make_parcel_feature()
    # Snapshot без ЗОУИТ/ЕНК в meta (только ЗУ и Здания)
    base_snapshot = QuarterSnapshot(
        quarter="66:41:0303161",
        fetched_at="2026-05-15T10:00:00+00:00",
        features=[parcel],
        meta_counts={36368: 2, 36369: 3},  # нет ЗОУИТ / ЕНК
    )

    db = MagicMock()
    db.execute = MagicMock(
        return_value=MagicMock(
            mappings=MagicMock(
                return_value=MagicMock(first=MagicMock(return_value={"phase_state": None}))
            )
        )
    )
    db.commit = MagicMock()

    client = AsyncMock()
    client.search_by_quarter = AsyncMock(return_value=base_snapshot)

    with patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert:
        mock_upsert.return_value = {
            "parcels": 1,
            "buildings": 0,
            "constructions": 0,
            "oncs": 0,
            "enks": 0,
            "zouit": 0,
            "skipped": 0,
        }
        await harvest_quarter(
            db=db,
            client=client,
            quarter="66:41:0303161",
            job_id=1,
            update_progress=lambda s: None,
        )

    # Только один вызов search_by_quarter (базовый), никаких per-cat probe
    assert client.search_by_quarter.await_count == 1
    call_kwargs = client.search_by_quarter.await_args_list[0].kwargs
    assert call_kwargs.get("category_id") is None


@pytest.mark.asyncio
async def test_harvest_quarter_per_cat_probe_enk_called_when_meta_nonzero() -> None:
    """Phase 1.5: если в meta есть ЕНК (39663) — вызываем per-cat probe с category_id=39663."""
    from app.services.cadastre.bulk_harvest import harvest_quarter

    parcel = _make_parcel_feature()
    base_snapshot = QuarterSnapshot(
        quarter="66:41:0303161",
        fetched_at="2026-05-15T10:00:00+00:00",
        features=[parcel],
        meta_counts={36368: 2, CAT_ENK: 3},  # ЕНК в meta
    )
    enk_snapshot = QuarterSnapshot(
        quarter="66:41:0303161",
        fetched_at="2026-05-15T10:00:01+00:00",
        features=[],
        meta_counts={CAT_ENK: 3},
    )

    db = MagicMock()
    db.execute = MagicMock(
        return_value=MagicMock(
            mappings=MagicMock(
                return_value=MagicMock(first=MagicMock(return_value={"phase_state": None}))
            )
        )
    )
    db.commit = MagicMock()

    client = AsyncMock()
    client.search_by_quarter = AsyncMock(side_effect=[base_snapshot, enk_snapshot])

    with patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert:
        mock_upsert.return_value = {
            "parcels": 1,
            "buildings": 0,
            "constructions": 0,
            "oncs": 0,
            "enks": 0,
            "zouit": 0,
            "skipped": 0,
        }
        await harvest_quarter(
            db=db,
            client=client,
            quarter="66:41:0303161",
            job_id=1,
            update_progress=lambda s: None,
        )

    # Ожидаем вызов с category_id=CAT_ENK
    per_cat_enk_calls = [
        c
        for c in client.search_by_quarter.await_args_list
        if c.kwargs.get("category_id") == CAT_ENK
    ]
    assert len(per_cat_enk_calls) >= 1, "Ожидался per-cat probe с category_id=39663 (ENK)"


# ── auto-heal cad_quarters_geom ──────────────────────────────────────────────


def _make_quarter_stats_feature(
    geom_type: str = "Polygon",
    geom_coords: list[Any] | None = None,
) -> NSPDBulkFeature:
    """Создать NSPDBulkFeature с category_id=36381 (квартальная статистика)."""
    if geom_coords is None:
        # 200m × 200m полигон в EPSG:3857 — здоровый квартал
        geom_coords = [
            [
                [6090000.0, 7590000.0],
                [6090200.0, 7590000.0],
                [6090200.0, 7590200.0],
                [6090000.0, 7590200.0],
                [6090000.0, 7590000.0],
            ]
        ]
    return NSPDBulkFeature.model_validate(
        {
            "id": 99,
            "type": "Feature",
            "geometry": {"type": geom_type, "coordinates": geom_coords},
            "properties": {
                "category": CAT_QUARTER_STATS,
                "options": {
                    "cnt_land": 5,
                    "cnt_oks": 3,
                },
            },
        }
    )


def _mock_db_execute_scalar(scalar_value: Any) -> MagicMock:
    """Mock db.execute().scalar() возвращает заданное значение."""
    mock_db = MagicMock()
    mock_db.execute.return_value.scalar.return_value = scalar_value
    return mock_db


def test_upsert_quarter_geom_skips_invalid_geometry_type() -> None:
    """Point / LineString / None geometry → False (skip без DB call)."""
    db = MagicMock()

    # Нет geometry вообще
    feature_no_geom = NSPDBulkFeature.model_validate(
        {"type": "Feature", "geometry": None, "properties": {"category": CAT_QUARTER_STATS}}
    )
    assert upsert_quarter_geom_from_feature(db, "66:41:0303161", feature_no_geom) is False
    db.execute.assert_not_called()

    # Point geometry
    feature_point = NSPDBulkFeature.model_validate(
        {
            "type": "Feature",
            "geometry": {"type": "Point", "coordinates": [6090000.0, 7590000.0]},
            "properties": {"category": CAT_QUARTER_STATS},
        }
    )
    assert upsert_quarter_geom_from_feature(db, "66:41:0303161", feature_point) is False
    db.execute.assert_not_called()

    # LineString geometry
    feature_line = NSPDBulkFeature.model_validate(
        {
            "type": "Feature",
            "geometry": {
                "type": "LineString",
                "coordinates": [[6090000.0, 7590000.0], [6090200.0, 7590200.0]],
            },
            "properties": {"category": CAT_QUARTER_STATS},
        }
    )
    assert upsert_quarter_geom_from_feature(db, "66:41:0303161", feature_line) is False
    db.execute.assert_not_called()


def test_upsert_quarter_geom_returns_false_when_sql_updates_nothing() -> None:
    """Если UPDATE не затронул строк (скипнул SQL sanity check) → scalar()=None → False.

    Симулирует случай: новая geom тоже broken (<100m) ИЛИ existing geom уже здорова.
    В обоих случаях RETURNING не вернёт строку, scalar() = None.
    """
    db = _mock_db_execute_scalar(None)  # RETURNING ничего не вернул
    feature = _make_quarter_stats_feature()  # здоровая feature

    result = upsert_quarter_geom_from_feature(db, "66:41:0303161", feature)

    assert result is False
    db.execute.assert_called_once()
    # Убедиться что был передан корректный quarter параметр
    call_params = db.execute.call_args[0][1]
    assert call_params["quarter"] == "66:41:0303161"
    assert "geom" in call_params
    assert call_params["min_w"] == 100.0
    assert call_params["max_w"] == 10000.0


def test_upsert_quarter_geom_returns_true_when_sql_updates_row() -> None:
    """Если UPDATE обновил строку (existing broken + new healthy) → scalar()=cad_number → True."""
    db = _mock_db_execute_scalar("66:41:0303161")  # RETURNING вернул cad_number
    feature = _make_quarter_stats_feature()  # здоровая feature 200m×200m

    result = upsert_quarter_geom_from_feature(db, "66:41:0303161", feature)

    assert result is True
    db.execute.assert_called_once()


def test_upsert_quarter_geom_multipolygon_accepted() -> None:
    """MultiPolygon тип тоже принимается (не только Polygon)."""
    db = _mock_db_execute_scalar("66:41:0303161")
    # MultiPolygon coords: список из одного полигона
    mp_coords = [
        [
            [
                [6090000.0, 7590000.0],
                [6090200.0, 7590000.0],
                [6090200.0, 7590200.0],
                [6090000.0, 7590200.0],
                [6090000.0, 7590000.0],
            ]
        ]
    ]
    feature = _make_quarter_stats_feature(geom_type="MultiPolygon", geom_coords=mp_coords)

    result = upsert_quarter_geom_from_feature(db, "66:41:0303161", feature)

    assert result is True
    db.execute.assert_called_once()


def test_upsert_quarter_geom_sql_uses_cast_not_double_colon() -> None:
    """SQL использует CAST(:x AS text) не :x::text (SQLAlchemy psycopg v3 trap)."""
    executed_sqls: list[str] = []

    db = MagicMock()
    db.execute.side_effect = lambda stmt, params=None: (
        executed_sqls.append(str(stmt)) or MagicMock(scalar=MagicMock(return_value=None))
    )

    feature = _make_quarter_stats_feature()
    upsert_quarter_geom_from_feature(db, "66:41:0303161", feature)

    assert len(executed_sqls) == 1
    sql = executed_sqls[0]
    assert "::jsonb" not in sql
    assert "::text" not in sql
    assert "CAST(:geom AS text)" in sql


def test_upsert_quarter_geom_custom_width_bounds() -> None:
    """min_width_m и max_width_m пробрасываются как параметры SQL (не hardcoded)."""
    db = _mock_db_execute_scalar(None)
    feature = _make_quarter_stats_feature()

    upsert_quarter_geom_from_feature(
        db, "66:41:0303161", feature, min_width_m=50.0, max_width_m=5000.0
    )

    call_params = db.execute.call_args[0][1]
    assert call_params["min_w"] == 50.0
    assert call_params["max_w"] == 5000.0


@pytest.mark.asyncio
async def test_harvest_quarter_geom_heal_failure_does_not_propagate() -> None:
    """Phase 4: если upsert_quarter_geom_from_feature бросает (malformed GeoJSON
    из NSPD → ST_GeomFromGeoJSON exception), harvest_quarter должен:
    1. Залогировать warning
    2. Rollback'нуть SAVEPOINT через begin_nested
    3. ПРОДОЛЖИТЬ — НЕ propagate exception
    4. Сохранить upsert_quarter_stats (успешный до geom failure)
    """
    from app.services.cadastre.bulk_harvest import harvest_quarter

    stats_feature = _make_quarter_stats_feature()
    snapshot = QuarterSnapshot(
        quarter="66:41:0303161",
        fetched_at="2026-05-15T10:00:00+00:00",
        features=[stats_feature],
        meta_counts={CAT_QUARTER_STATS: 1},
    )

    db = MagicMock()
    db.execute = MagicMock(
        return_value=MagicMock(
            mappings=MagicMock(
                return_value=MagicMock(first=MagicMock(return_value={"phase_state": None}))
            )
        )
    )
    db.commit = MagicMock()
    db.begin_nested = MagicMock()  # context manager mock — exit без error

    client = AsyncMock()
    client.search_by_quarter = AsyncMock(return_value=snapshot)

    with (
        patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert,
        patch("app.services.cadastre.bulk_harvest.upsert_quarter_stats") as mock_stats,
        patch("app.services.cadastre.bulk_harvest.upsert_quarter_geom_from_feature") as mock_heal,
    ):
        mock_upsert.return_value = {
            "parcels": 0,
            "buildings": 0,
            "constructions": 0,
            "oncs": 0,
            "enks": 0,
            "zouit": 0,
            "skipped": 0,
        }
        mock_heal.side_effect = Exception("ST_GeomFromGeoJSON: invalid polygon")

        # Это не должно raise
        result = await harvest_quarter(
            db=db,
            client=client,
            quarter="66:41:0303161",
            job_id=1,
            update_progress=lambda s: None,
        )

    # Stats был вызван (Phase 4 продолжил после geom failure)
    mock_stats.assert_called_once()
    mock_heal.assert_called_once()
    # Phase=done — harvest завершился штатно
    assert result.phase_state == {"phase": "done", "quarter": "66:41:0303161"}