Light1YT
1883271908
Some checks failed
CI / changes (push) Successful in 7s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 6s
CI / frontend-tests (pull_request) Has been skipped
CI / backend-tests (push) Successful in 6m43s
CI / backend-tests (pull_request) Successful in 6m41s
Deploy / build-backend (push) Blocked by required conditions
Deploy / deploy (push) Blocked by required conditions
Deploy / build-worker (push) Blocked by required conditions
Deploy / build-frontend (push) Blocked by required conditions
Deploy / changes (push) Has been cancelled
#252: NspdBulkServerError на HTTP 5xx + WMS <ServiceException>-body → quarter-level autoretry; _grid_walk_category флагует layer_failed только если ВСЕ cells 5xx (none ok), network-errors исключены — пустой/флапающий слой не ложно-скипается. harvest_meta.layer_X_failed. #259: pzz-sync-monthly beat отключён (PKK6 deprecated, pzz_zones_ekb=0, заменён zone_regulation_cache+NSPD dumps). logger.exception→warning (стоп GlitchTip BACKEND-1B), raise сохранён для admin-endpoint. #200: geom_unavailable флаг (миграция 151) + backfill_parcel_geom task — grid-walk layer 36368 per quarter лечит 930/964 NULL-geom участков (NSPD search вернул центроиды), остаток marks unavailable (не ретраить вечно). SAVEPOINT per-cell, idempotent. Closes #252 Closes #259 Closes #200
1638 lines
61 KiB
Python
1638 lines
61 KiB
Python
"""Тесты для 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
|
||
|
||
|
||
def test_parse_nspd_date_handles_dd_mm_yyyy() -> None:
|
||
"""NSPD Сооружения/ОНС/ЕНК отдают даты в DD.MM.YYYY → конвертируем в ISO."""
|
||
from app.services.cadastre.bulk_harvest import _parse_nspd_date
|
||
|
||
assert _parse_nspd_date("13.03.2023") == "2023-03-13"
|
||
assert _parse_nspd_date("24.02.2022") == "2022-02-24"
|
||
|
||
|
||
def test_parse_nspd_date_passes_through_iso() -> None:
|
||
"""ISO формат (YYYY-MM-DD) уже валидный — pass через."""
|
||
from app.services.cadastre.bulk_harvest import _parse_nspd_date
|
||
|
||
assert _parse_nspd_date("2023-03-13") == "2023-03-13"
|
||
assert _parse_nspd_date("2010-08-12 00:00:00") == "2010-08-12"
|
||
|
||
|
||
def test_parse_nspd_date_none_and_invalid() -> None:
|
||
"""None / empty string / garbage → None."""
|
||
from app.services.cadastre.bulk_harvest import _parse_nspd_date
|
||
|
||
assert _parse_nspd_date(None) is None
|
||
assert _parse_nspd_date("") is None
|
||
assert _parse_nspd_date("nonsense") is None
|
||
assert _parse_nspd_date(123) is None # not str
|
||
|
||
|
||
def test_parse_nspd_date_validates_month_day_ranges() -> None:
|
||
"""Invalid month/day (13.03 / 32.13 / etc) → None — не пропускать в SQL."""
|
||
from app.services.cadastre.bulk_harvest import _parse_nspd_date
|
||
|
||
# Невалидный месяц (13) в ISO формате
|
||
assert _parse_nspd_date("2023-13-45") is None
|
||
# Невалидный день в DD.MM.YYYY
|
||
assert _parse_nspd_date("32.13.2023") is None
|
||
# Невалидный month
|
||
assert _parse_nspd_date("31.02.2023") is None # Feb 31
|
||
|
||
|
||
# ── 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
|
||
|
||
|
||
def _db_with_real_savepoints() -> tuple[MagicMock, list[Any]]:
|
||
"""Mock Session где begin_nested() — настоящий context manager (для SAVEPOINT-теста).
|
||
|
||
Возвращает (db, savepoint_calls) — savepoint_calls фиксирует каждый __enter__,
|
||
т.е. сколько раз был открыт per-feature SAVEPOINT.
|
||
"""
|
||
savepoint_calls: list[Any] = []
|
||
db = MagicMock()
|
||
|
||
def _begin_nested() -> Any:
|
||
cm = MagicMock()
|
||
cm.__enter__ = MagicMock(side_effect=lambda: savepoint_calls.append(1))
|
||
cm.__exit__ = MagicMock(return_value=False)
|
||
return cm
|
||
|
||
db.begin_nested = MagicMock(side_effect=_begin_nested)
|
||
return db, savepoint_calls
|
||
|
||
|
||
def test_upsert_features_wraps_each_row_in_savepoint() -> None:
|
||
"""backend.md SAVEPOINT rule: upsert_features открывает begin_nested() per feature."""
|
||
parcel = _make_parcel_feature("66:41:0303161:1")
|
||
building = _make_building_feature("66:41:0303161:1:1")
|
||
db, savepoint_calls = _db_with_real_savepoints()
|
||
|
||
with (
|
||
patch("app.services.cadastre.bulk_harvest.upsert_parcel"),
|
||
patch("app.services.cadastre.bulk_harvest.upsert_building"),
|
||
):
|
||
upsert_features(db, [parcel, building], source="search")
|
||
|
||
# По одному SAVEPOINT на каждый из 2 features
|
||
assert db.begin_nested.call_count == 2
|
||
assert len(savepoint_calls) == 2
|
||
|
||
|
||
def test_upsert_features_one_bad_feature_does_not_abort_others() -> None:
|
||
"""Один битый feature (raise в upsert_*) НЕ роняет весь snapshot — остальные сохраняются,
|
||
counts остаются точными (инкремент только при успехе).
|
||
|
||
Раньше: без per-row begin_nested один битый feature отравлял всю tx квартала →
|
||
все features rollback'ились (а счётчики уже инкрементированы = inconsistent).
|
||
"""
|
||
parcel_ok = _make_parcel_feature("66:41:0303161:1")
|
||
building_bad = _make_building_feature("66:41:0303161:1:1")
|
||
parcel_ok2 = _make_parcel_feature("66:41:0303161:2")
|
||
db, _savepoint_calls = _db_with_real_savepoints()
|
||
|
||
with (
|
||
patch("app.services.cadastre.bulk_harvest.upsert_parcel") as mock_parcel,
|
||
patch("app.services.cadastre.bulk_harvest.upsert_building") as mock_building,
|
||
):
|
||
# Здание падает (например type mismatch / NOT NULL), парцелы — ок
|
||
mock_building.side_effect = Exception("geom type mismatch / NOT NULL violation")
|
||
counts = upsert_features(db, [parcel_ok, building_bad, parcel_ok2], source="search")
|
||
|
||
# Оба парцела обработаны несмотря на падение здания между ними
|
||
assert mock_parcel.call_count == 2
|
||
mock_building.assert_called_once()
|
||
# Счётчики точные: 2 успешных парцела, 0 зданий (битое не считается)
|
||
assert counts["parcels"] == 2
|
||
assert counts["buildings"] == 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
|
||
|
||
|
||
def test_upsert_parcel_sql_uses_st_multi_for_multipolygon_schema() -> None:
|
||
"""cad_parcels.geom = MULTIPOLYGON (migration 93) — SQL должен ST_Multi() оборачивать
|
||
transform, чтобы Polygon coerce'ился в MultiPolygon (NSPD возвращает оба варианта).
|
||
"""
|
||
from app.services.cadastre.bulk_harvest import upsert_parcel
|
||
|
||
db = MagicMock()
|
||
executed_sqls: list[str] = []
|
||
db.execute = lambda stmt, params=None: executed_sqls.append(str(stmt)) or MagicMock()
|
||
|
||
upsert_parcel(db, _make_parcel_feature(), source="search")
|
||
|
||
assert len(executed_sqls) == 1
|
||
sql = executed_sqls[0]
|
||
assert "ST_Multi(" in sql, "SQL должен содержать ST_Multi для MultiPolygon schema"
|
||
|
||
|
||
# ── harvest_quarter integration (async, mocked) ───────────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_harvest_quarter_does_not_early_exit_on_shared_phase_done() -> None:
|
||
"""harvest_quarter НЕ должен бейлиться если другой worker записал phase=done
|
||
в shared `cadastre_jobs.phase_state`. Раньше тут был early-exit, который
|
||
вызывал 95% workers пропускать работу (race condition между parallel Celery
|
||
tasks одного job).
|
||
|
||
Idempotency обеспечивается ON CONFLICT DO UPDATE в upsert_*, поэтому даже
|
||
если task re-enqueued после shared phase=done, work должна выполниться.
|
||
"""
|
||
from app.services.cadastre.bulk_harvest import harvest_quarter
|
||
|
||
snapshot = QuarterSnapshot(
|
||
quarter="66:41:0303161",
|
||
fetched_at="2026-05-15T10:00:00+00:00",
|
||
features=[],
|
||
meta_counts={},
|
||
)
|
||
|
||
db = MagicMock()
|
||
# Симулируем shared phase_state с phase=done от ДРУГОГО quarter.
|
||
# xmin=None → quarter_bbox_3857 (grid-walk geometry helper) вернёт None →
|
||
# grid-walk + territorial_zones фазы корректно пропускаются (тест про
|
||
# snapshot/idempotency, не про geometry). Тот же dict возвращается на ВСЕ
|
||
# db.execute().mappings().first() в этом тесте.
|
||
db.execute = MagicMock(
|
||
return_value=MagicMock(
|
||
mappings=MagicMock(
|
||
return_value=MagicMock(
|
||
first=MagicMock(
|
||
return_value={
|
||
"phase_state": {"phase": "done", "quarter": "66:41:9999999"},
|
||
"xmin": None,
|
||
"ymin": None,
|
||
"xmax": None,
|
||
"ymax": None,
|
||
}
|
||
)
|
||
)
|
||
)
|
||
)
|
||
)
|
||
db.commit = MagicMock()
|
||
|
||
client = AsyncMock()
|
||
client.search_by_quarter = AsyncMock(return_value=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": 0,
|
||
"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 ДОЛЖЕН быть вызван несмотря на shared phase=done
|
||
client.search_by_quarter.assert_awaited_once_with("66:41:0303161")
|
||
assert result.snapshot_requests == 1
|
||
assert result.phase_state == {"phase": "done", "quarter": "66:41:0303161"}
|
||
|
||
|
||
@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 → начинаем с нуля
|
||
# xmin=None → quarter_bbox_3857 (grid-walk geometry helper) вернёт None →
|
||
# grid-walk + territorial_zones фазы пропускаются (тесты про snapshot /
|
||
# per-cat-probe, не про geometry). Тот же dict на ВСЕ
|
||
# db.execute().mappings().first() вызовы.
|
||
db.execute = MagicMock(
|
||
return_value=MagicMock(
|
||
mappings=MagicMock(
|
||
return_value=MagicMock(
|
||
first=MagicMock(
|
||
return_value={
|
||
"phase_state": None,
|
||
"xmin": None,
|
||
"ymin": None,
|
||
"xmax": None,
|
||
"ymax": 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
|
||
# Phase 1 = 1 NSPD search call → snapshot_requests=1
|
||
assert result.snapshot_requests == 1
|
||
assert result.grid_walk_requests == 0
|
||
assert result.total_requests == 1
|
||
|
||
# Проверить что последний progress = done
|
||
assert progress_states[-1]["phase"] == "done"
|
||
|
||
|
||
def test_harvest_result_total_requests_property() -> None:
|
||
"""HarvestResult.total_requests = snapshot_requests + grid_walk_requests."""
|
||
from app.services.cadastre.bulk_harvest import HarvestResult
|
||
|
||
r = HarvestResult(quarter="66:41:0303161", snapshot_requests=3, grid_walk_requests=225)
|
||
assert r.total_requests == 228
|
||
|
||
empty = HarvestResult(quarter="66:41:0303161")
|
||
assert empty.total_requests == 0
|
||
|
||
|
||
# ── 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, layer_failed = 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
|
||
assert layer_failed is False # все cells успешны
|
||
|
||
|
||
# ── ЗОУИТ 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()
|
||
# xmin=None → quarter_bbox_3857 (grid-walk geometry helper) вернёт None →
|
||
# grid-walk + territorial_zones фазы пропускаются (тесты про snapshot /
|
||
# per-cat-probe, не про geometry). Тот же dict на ВСЕ
|
||
# db.execute().mappings().first() вызовы.
|
||
db.execute = MagicMock(
|
||
return_value=MagicMock(
|
||
mappings=MagicMock(
|
||
return_value=MagicMock(
|
||
first=MagicMock(
|
||
return_value={
|
||
"phase_state": None,
|
||
"xmin": None,
|
||
"ymin": None,
|
||
"xmax": None,
|
||
"ymax": 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()
|
||
# xmin=None → quarter_bbox_3857 (grid-walk geometry helper) вернёт None →
|
||
# grid-walk + territorial_zones фазы пропускаются (тесты про snapshot /
|
||
# per-cat-probe, не про geometry). Тот же dict на ВСЕ
|
||
# db.execute().mappings().first() вызовы.
|
||
db.execute = MagicMock(
|
||
return_value=MagicMock(
|
||
mappings=MagicMock(
|
||
return_value=MagicMock(
|
||
first=MagicMock(
|
||
return_value={
|
||
"phase_state": None,
|
||
"xmin": None,
|
||
"ymin": None,
|
||
"xmax": None,
|
||
"ymax": 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()
|
||
# xmin=None → quarter_bbox_3857 (grid-walk geometry helper) вернёт None →
|
||
# grid-walk + territorial_zones фазы пропускаются (тесты про snapshot /
|
||
# per-cat-probe, не про geometry). Тот же dict на ВСЕ
|
||
# db.execute().mappings().first() вызовы.
|
||
db.execute = MagicMock(
|
||
return_value=MagicMock(
|
||
mappings=MagicMock(
|
||
return_value=MagicMock(
|
||
first=MagicMock(
|
||
return_value={
|
||
"phase_state": None,
|
||
"xmin": None,
|
||
"ymin": None,
|
||
"xmax": None,
|
||
"ymax": 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_parcel_skips_geom_for_point_geometry() -> None:
|
||
"""NSPD изредка возвращает Point для ЗУ — cad_parcels.geom strict Polygon,
|
||
INSERT fail'нул бы. Фильтр на Python level → geom=NULL, raw_props сохранён.
|
||
"""
|
||
from app.services.cadastre.bulk_harvest import upsert_parcel
|
||
|
||
point_parcel = NSPDBulkFeature.model_validate(
|
||
{
|
||
"id": 99,
|
||
"type": "Feature",
|
||
"geometry": {"type": "Point", "coordinates": [6753892.91, 7747664.31]},
|
||
"properties": {
|
||
"category": 36368,
|
||
"options": {
|
||
"cad_num": "66:41:0104002:18",
|
||
"quarter_cad_number": "66:41:0104002",
|
||
"declared_area": 200.0,
|
||
"readable_address": "г Екатеринбург, ул Тестовая",
|
||
},
|
||
},
|
||
}
|
||
)
|
||
|
||
db = MagicMock()
|
||
upsert_parcel(db, point_parcel, source="search")
|
||
|
||
# db.execute должен быть вызван (parcel НЕ skipped, только geom=NULL)
|
||
db.execute.assert_called_once()
|
||
call_params = db.execute.call_args[0][1]
|
||
assert call_params["cad_num"] == "66:41:0104002:18"
|
||
# geom параметр должен быть None
|
||
assert call_params["geom"] is None
|
||
|
||
|
||
def test_upsert_parcel_keeps_polygon_geometry() -> None:
|
||
"""Polygon geometry для ЗУ — обычный happy path."""
|
||
from app.services.cadastre.bulk_harvest import upsert_parcel
|
||
|
||
parcel = _make_parcel_feature() # Polygon
|
||
db = MagicMock()
|
||
upsert_parcel(db, parcel, source="search")
|
||
|
||
db.execute.assert_called_once()
|
||
call_params = db.execute.call_args[0][1]
|
||
assert call_params["geom"] is not None
|
||
assert '"type": "Polygon"' in call_params["geom"]
|
||
|
||
|
||
def test_upsert_zouit_skips_geom_for_linestring() -> None:
|
||
"""ЗОУИТ schema MultiPolygon — LineString не валиден, geom=NULL."""
|
||
from app.services.cadastre.bulk_harvest import upsert_zouit
|
||
|
||
line_zouit = NSPDBulkFeature.model_validate(
|
||
{
|
||
"id": 1,
|
||
"type": "Feature",
|
||
"geometry": {
|
||
"type": "LineString",
|
||
"coordinates": [[6090000, 7590000], [6090100, 7590100]],
|
||
},
|
||
"properties": {
|
||
"category": 36940,
|
||
"options": {
|
||
"reg_numb_border": "66-0.1-2.1-99",
|
||
"type_zone": "Linear",
|
||
"registration_date": "2020-01-01",
|
||
},
|
||
},
|
||
}
|
||
)
|
||
|
||
db = MagicMock()
|
||
upsert_zouit(db, line_zouit, source="search")
|
||
|
||
db.execute.assert_called_once()
|
||
call_params = db.execute.call_args[0][1]
|
||
assert call_params["geom"] is None
|
||
|
||
|
||
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()
|
||
# xmin=None → quarter_bbox_3857 (grid-walk geometry helper) вернёт None →
|
||
# grid-walk + territorial_zones фазы пропускаются (тесты про snapshot /
|
||
# per-cat-probe, не про geometry). Тот же dict на ВСЕ
|
||
# db.execute().mappings().first() вызовы.
|
||
db.execute = MagicMock(
|
||
return_value=MagicMock(
|
||
mappings=MagicMock(
|
||
return_value=MagicMock(
|
||
first=MagicMock(
|
||
return_value={
|
||
"phase_state": None,
|
||
"xmin": None,
|
||
"ymin": None,
|
||
"xmax": None,
|
||
"ymax": 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"}
|
||
|
||
|
||
# ── Fix I: heartbeat в grid-walk loop ────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grid_walk_emits_heartbeat_callbacks() -> None:
|
||
"""_grid_walk_category должен вызывать update_progress каждые heartbeat_every cells.
|
||
|
||
225 cells (15×15) / 50 = 4 heartbeats minimum (idx=50, 100, 150, 200).
|
||
"""
|
||
from app.services.cadastre.bulk_harvest import _grid_walk_category
|
||
|
||
# Mock db: quarter_bbox_3857 returns valid bbox
|
||
db = MagicMock()
|
||
db.execute.return_value.mappings.return_value.first.return_value = {
|
||
"xmin": 6735845.0,
|
||
"ymin": 8329000.0,
|
||
"xmax": 6736595.0,
|
||
"ymax": 8329750.0,
|
||
}
|
||
|
||
# Mock client: wms_feature_info returns empty list (no upserts to worry about)
|
||
client = AsyncMock()
|
||
client.wms_feature_info = AsyncMock(return_value=[])
|
||
|
||
progress_states: list[dict[str, Any]] = []
|
||
|
||
_upserted, requests, _failed = await _grid_walk_category(
|
||
db=db,
|
||
client=client,
|
||
quarter="66:41:0303161",
|
||
layer_id=36368,
|
||
update_progress=lambda s: progress_states.append(s),
|
||
heartbeat_every=50,
|
||
)
|
||
|
||
# 225 cells / 50 = 4 heartbeats at idx=50/100/150/200
|
||
assert len(progress_states) == 4
|
||
assert all("grid_cells_done" in s for s in progress_states)
|
||
assert progress_states[0]["grid_cells_done"] == 50
|
||
assert progress_states[-1]["grid_cells_done"] == 200
|
||
assert progress_states[0]["quarter"] == "66:41:0303161"
|
||
# 225 WMS calls (request counter)
|
||
assert requests == 225
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grid_walk_no_heartbeat_when_callback_none() -> None:
|
||
"""Если update_progress=None, не должны падать (callback optional)."""
|
||
from app.services.cadastre.bulk_harvest import _grid_walk_category
|
||
|
||
db = MagicMock()
|
||
db.execute.return_value.mappings.return_value.first.return_value = {
|
||
"xmin": 6735845.0,
|
||
"ymin": 8329000.0,
|
||
"xmax": 6736595.0,
|
||
"ymax": 8329750.0,
|
||
}
|
||
client = AsyncMock()
|
||
client.wms_feature_info = AsyncMock(return_value=[])
|
||
|
||
# Не передаём update_progress — должно не падать
|
||
_upserted, requests, _failed = await _grid_walk_category(
|
||
db=db,
|
||
client=client,
|
||
quarter="66:41:0303161",
|
||
layer_id=36368,
|
||
)
|
||
assert requests == 225
|
||
|
||
|
||
# ── Issue #252: per-layer skip при 5xx/ServiceException ──────────────────────
|
||
|
||
|
||
def _mock_db_grid_bbox() -> MagicMock:
|
||
"""Mock db с валидным bbox для quarter_bbox_3857 + begin_nested CM."""
|
||
db = MagicMock()
|
||
db.execute.return_value.mappings.return_value.first.return_value = {
|
||
"xmin": 6735845.0,
|
||
"ymin": 8329000.0,
|
||
"xmax": 6736595.0,
|
||
"ymax": 8329750.0,
|
||
}
|
||
db.begin_nested = MagicMock(
|
||
return_value=MagicMock(__enter__=MagicMock(return_value=None), __exit__=MagicMock())
|
||
)
|
||
db.commit = MagicMock()
|
||
return db
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grid_walk_marks_layer_failed_when_all_cells_500() -> None:
|
||
"""Issue #252: ВСЕ cells слоя упали на NspdBulkServerError → layer_failed=True."""
|
||
from app.scrapers.nspd_bulk_client import NspdBulkServerError
|
||
from app.services.cadastre.bulk_harvest import _grid_walk_category
|
||
|
||
db = _mock_db_grid_bbox()
|
||
client = AsyncMock()
|
||
client.wms_feature_info = AsyncMock(side_effect=NspdBulkServerError("HTTP 500"))
|
||
|
||
upserted, requests, layer_failed = await _grid_walk_category(
|
||
db=db,
|
||
client=client,
|
||
quarter="66:41:0303161",
|
||
layer_id=36368,
|
||
grid_size=3, # 9 cells, все упадут
|
||
)
|
||
|
||
assert upserted == 0
|
||
assert requests == 9
|
||
assert layer_failed is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grid_walk_layer_not_failed_when_some_cells_ok() -> None:
|
||
"""Issue #252: если хоть один cell прошёл — layer_failed=False (слой жив, просто пуст)."""
|
||
from app.scrapers.nspd_bulk_client import NspdBulkServerError
|
||
from app.services.cadastre.bulk_harvest import _grid_walk_category
|
||
|
||
db = _mock_db_grid_bbox()
|
||
client = AsyncMock()
|
||
# Первый cell ок (пустой ответ), остальные 500
|
||
client.wms_feature_info = AsyncMock(
|
||
side_effect=[[], NspdBulkServerError("500"), NspdBulkServerError("500")]
|
||
)
|
||
|
||
_upserted, requests, layer_failed = await _grid_walk_category(
|
||
db=db,
|
||
client=client,
|
||
quarter="66:41:0303161",
|
||
layer_id=36368,
|
||
grid_size=1, # уйдёт по side_effect — но grid_size=1 → 1 cell
|
||
)
|
||
|
||
# grid_size=1 → 1 cell (ок) → layer жив
|
||
assert layer_failed is False
|
||
assert requests == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grid_walk_network_error_does_not_trigger_layer_failed() -> None:
|
||
"""Issue #252: сетевая (не server-side) ошибка cell НЕ учитывается в fail-rate."""
|
||
from app.scrapers.nspd_bulk_client import NspdBulkError
|
||
from app.services.cadastre.bulk_harvest import _grid_walk_category
|
||
|
||
db = _mock_db_grid_bbox()
|
||
client = AsyncMock()
|
||
# NspdBulkError (не Server) — например network. layer_failed требует server_errors>0.
|
||
client.wms_feature_info = AsyncMock(side_effect=NspdBulkError("Network error"))
|
||
|
||
_upserted, requests, layer_failed = await _grid_walk_category(
|
||
db=db,
|
||
client=client,
|
||
quarter="66:41:0303161",
|
||
layer_id=36368,
|
||
grid_size=3,
|
||
)
|
||
|
||
# server_errors==0 (это не 5xx) → layer_failed=False, квартал не получит skip-флаг
|
||
assert layer_failed is False
|
||
assert requests == 9
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_harvest_quarter_records_failed_layer_in_phase_state() -> None:
|
||
"""Issue #252: сбойный слой grid-walk → harvest_meta.layer_X_failed в done-state."""
|
||
from app.services.cadastre.bulk_harvest import CAT_PARCEL, 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={CAT_PARCEL: 170}, # overflow → grid-walk слоя 36368
|
||
)
|
||
|
||
db = _mock_db_grid_bbox()
|
||
client = AsyncMock()
|
||
client.search_by_quarter = AsyncMock(return_value=snapshot)
|
||
client.get_territorial_zones_in_bbox = AsyncMock(return_value=[])
|
||
|
||
progress_states: list[dict[str, Any]] = []
|
||
|
||
with (
|
||
patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert,
|
||
patch("app.services.cadastre.bulk_harvest._grid_walk_category") as mock_walk,
|
||
patch("app.services.cadastre.bulk_harvest.quarter_bbox_3857") as mock_bbox,
|
||
):
|
||
mock_bbox.return_value = (6735845.0, 8329000.0, 6736595.0, 8329750.0)
|
||
mock_upsert.return_value = {
|
||
"parcels": 1,
|
||
"buildings": 0,
|
||
"constructions": 0,
|
||
"oncs": 0,
|
||
"enks": 0,
|
||
"zouit": 0,
|
||
"skipped": 0,
|
||
}
|
||
# grid-walk слоя CAT_PARCEL вернул layer_failed=True
|
||
mock_walk.return_value = (0, 9, True)
|
||
|
||
result = await harvest_quarter(
|
||
db=db,
|
||
client=client,
|
||
quarter="66:41:0303161",
|
||
job_id=1,
|
||
update_progress=lambda s: progress_states.append(s),
|
||
)
|
||
|
||
# failed_layers содержит CAT_PARCEL
|
||
assert result.failed_layers == [CAT_PARCEL]
|
||
# Финальный done-state несёт harvest_meta-флаг
|
||
assert result.phase_state is not None
|
||
assert result.phase_state["harvest_meta"] == {f"layer_{CAT_PARCEL}_failed": True}
|
||
# Квартал НЕ свалился — дошёл до done
|
||
assert progress_states[-1]["phase"] == "done"
|
||
|
||
|
||
# ── Issue #200: geom backfill для участков с geom IS NULL ─────────────────────
|
||
|
||
|
||
def test_mark_geom_unavailable_parametrized_no_fstring() -> None:
|
||
"""_mark_geom_unavailable использует ANY(CAST(:cads AS text[])) — не f-string SQL."""
|
||
from app.services.cadastre.bulk_harvest import _mark_geom_unavailable
|
||
|
||
executed: list[tuple[str, Any]] = []
|
||
|
||
db = MagicMock()
|
||
|
||
def capture(stmt: Any, params: Any = None) -> MagicMock:
|
||
executed.append((str(stmt), params))
|
||
return MagicMock(rowcount=2)
|
||
|
||
db.execute = capture
|
||
|
||
n = _mark_geom_unavailable(db, ["66:41:0303161:1", "66:41:0303161:2"])
|
||
|
||
assert n == 2
|
||
assert len(executed) == 1
|
||
sql, params = executed[0]
|
||
assert "ANY(CAST(:cads AS text[]))" in sql
|
||
assert "geom_unavailable = TRUE" in sql
|
||
assert params["cads"] == ["66:41:0303161:1", "66:41:0303161:2"]
|
||
|
||
|
||
def test_mark_geom_unavailable_empty_noop() -> None:
|
||
"""Пустой список → 0, без db.execute."""
|
||
from app.services.cadastre.bulk_harvest import _mark_geom_unavailable
|
||
|
||
db = MagicMock()
|
||
assert _mark_geom_unavailable(db, []) == 0
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_parcel_geom_heals_and_marks_unavailable() -> None:
|
||
"""Issue #200: участок с восстановленным geom → healed; оставшийся NULL → unavailable."""
|
||
from app.services.cadastre.bulk_harvest import backfill_parcel_geom
|
||
|
||
# Очередь: 2 участка одного квартала.
|
||
null_rows = ["66:41:0303161:1", "66:41:0303161:2"]
|
||
# После grid-walk участок :2 всё ещё NULL → должен быть помечен unavailable.
|
||
still_null = ["66:41:0303161:2"]
|
||
|
||
db = MagicMock()
|
||
captured: list[dict[str, Any]] = []
|
||
|
||
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
|
||
sql = str(stmt)
|
||
result = MagicMock()
|
||
if "geom IS NULL AND geom_unavailable = FALSE" in sql and "LIMIT" in sql:
|
||
result.scalars.return_value.all.return_value = null_rows
|
||
elif "AND geom IS NULL" in sql and "ANY(CAST(:cads AS text[]))" in sql and "SELECT" in sql:
|
||
result.scalars.return_value.all.return_value = still_null
|
||
elif "UPDATE cad_parcels SET geom_unavailable = TRUE" in sql:
|
||
captured.append({"sql": sql, "params": params or {}})
|
||
result.rowcount = len(still_null)
|
||
else:
|
||
result.scalars.return_value.all.return_value = []
|
||
result.rowcount = 0
|
||
return result
|
||
|
||
db.execute = execute_router
|
||
db.commit = MagicMock()
|
||
|
||
client = AsyncMock()
|
||
|
||
with patch("app.services.cadastre.bulk_harvest._grid_walk_category") as mock_walk:
|
||
# grid-walk слоя 36368 «нашёл» 1 полигон, без layer_failed
|
||
mock_walk.return_value = (1, 9, False)
|
||
result = await backfill_parcel_geom(db=db, client=client, limit=500)
|
||
|
||
assert result.parcels_targeted == 2
|
||
assert result.quarters_scanned == 1
|
||
assert result.parcels_healed == 1 # :1 получил geom
|
||
assert result.parcels_marked_unavailable == 1 # :2 помечен
|
||
# grid-walk вызван для квартала с layer_id=36368
|
||
mock_walk.assert_called_once()
|
||
assert mock_walk.call_args.kwargs["quarter"] == "66:41:0303161"
|
||
assert mock_walk.call_args.kwargs["layer_id"] == 36368
|
||
# UPDATE geom_unavailable выполнен ровно для оставшегося NULL
|
||
assert len(captured) == 1
|
||
assert captured[0]["params"]["cads"] == still_null
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_parcel_geom_empty_queue_noop() -> None:
|
||
"""Issue #200: нет участков с geom IS NULL → пустой результат, grid-walk не вызывается."""
|
||
from app.services.cadastre.bulk_harvest import backfill_parcel_geom
|
||
|
||
db = MagicMock()
|
||
db.execute.return_value.scalars.return_value.all.return_value = []
|
||
client = AsyncMock()
|
||
|
||
with patch("app.services.cadastre.bulk_harvest._grid_walk_category") as mock_walk:
|
||
result = await backfill_parcel_geom(db=db, client=client, limit=500)
|
||
|
||
assert result.parcels_targeted == 0
|
||
assert result.quarters_scanned == 0
|
||
mock_walk.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_parcel_geom_marks_orphan_cads_unavailable() -> None:
|
||
"""Issue #200: cad без валидного 3-сегментного квартала → сразу unavailable (нет bbox)."""
|
||
from app.services.cadastre.bulk_harvest import backfill_parcel_geom
|
||
|
||
orphan_rows = ["66:41", "garbage"] # <3 сегментов → grid-walk невозможен
|
||
|
||
db = MagicMock()
|
||
marked: list[Any] = []
|
||
|
||
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
|
||
sql = str(stmt)
|
||
result = MagicMock()
|
||
if "geom IS NULL AND geom_unavailable = FALSE" in sql and "LIMIT" in sql:
|
||
result.scalars.return_value.all.return_value = orphan_rows
|
||
elif "UPDATE cad_parcels SET geom_unavailable = TRUE" in sql:
|
||
marked.append(params)
|
||
result.rowcount = len(params["cads"])
|
||
else:
|
||
result.scalars.return_value.all.return_value = []
|
||
result.rowcount = 0
|
||
return result
|
||
|
||
db.execute = execute_router
|
||
db.commit = MagicMock()
|
||
client = AsyncMock()
|
||
|
||
with patch("app.services.cadastre.bulk_harvest._grid_walk_category") as mock_walk:
|
||
result = await backfill_parcel_geom(db=db, client=client, limit=500)
|
||
|
||
# Оба orphan помечены unavailable, grid-walk не вызывался
|
||
mock_walk.assert_not_called()
|
||
assert result.parcels_marked_unavailable == 2
|
||
assert marked and sorted(marked[0]["cads"]) == ["66:41", "garbage"]
|