"""Unit tests для quarter_dump_lookup helper (Sprint 1.1 item #4). Mock-based — без реальной БД и Celery. Проверяет: - derive_quarter_cad: 3/4/5-сегмент → правильный quarter - get_quarter_dump_data: отсутствующий дамп → triggers harvest, available=False - get_quarter_dump_data: устаревший дамп → stale=True, triggers harvest - get_quarter_dump_data: свежий дамп с territorial_zones → nspd_zoning populated - get_quarter_dump_data: свежий дамп с 2 zouit → nspd_zouit_overlaps length=2 - get_quarter_dump_data: parcel_wkt=None → metadata only, no spatial queries - make_empty_result: mutation-safety — каждый вызов возвращает независимый dict - early-exit: layer_counts == 0 → spatial query не вызывается """ from __future__ import annotations from datetime import UTC, datetime, timedelta from typing import Any from unittest.mock import MagicMock, patch from app.services.site_finder.quarter_dump_lookup import ( EMPTY_DUMP_RESULT, derive_quarter_cad, get_quarter_dump_data, make_empty_result, ) # ── derive_quarter_cad ──────────────────────────────────────────────────────── def test_derive_quarter_cad_3seg() -> None: """3-сегментный номер — квартал сам себе.""" assert derive_quarter_cad("66:41:0204016") == "66:41:0204016" def test_derive_quarter_cad_4seg() -> None: """4-сегментный участок → 3-сегментный квартал.""" assert derive_quarter_cad("66:41:0204016:10") == "66:41:0204016" def test_derive_quarter_cad_5seg() -> None: """5-сегментное здание → 3-сегментный квартал.""" assert derive_quarter_cad("66:41:0204016:10:1") == "66:41:0204016" def test_derive_quarter_cad_invalid() -> None: """Невалидный формат → None.""" assert derive_quarter_cad("66:41") is None assert derive_quarter_cad("abc:def:ghi") is None assert derive_quarter_cad("") is None def test_derive_quarter_cad_strips_whitespace() -> None: """Пробелы обрезаются.""" assert derive_quarter_cad(" 66:41:0204016 ") == "66:41:0204016" # ── get_quarter_dump_data — no row ──────────────────────────────────────────── def _make_db_mock(row: Any = None) -> MagicMock: """Создаёт mock Session.execute().first() с заданным результатом.""" db = MagicMock() execute_result = MagicMock() execute_result.first.return_value = row db.execute.return_value = execute_result return db def test_get_quarter_dump_no_row() -> None: """Дамп отсутствует в БД → harvest triggered, available=False.""" db = _make_db_mock(row=None) with patch( "app.services.site_finder.quarter_dump_lookup._trigger_harvest", return_value=True, ) as mock_trigger: result = get_quarter_dump_data(db, "66:41:0204016:10", parcel_wkt="POLYGON ((...))") mock_trigger.assert_called_once_with("66:41:0204016") assert result["nspd_dump"]["available"] is False assert result["nspd_dump"]["harvest_triggered"] is True assert result["nspd_dump"]["stale"] is False assert result["nspd_zoning"] is None assert result["nspd_zouit_overlaps"] == [] assert result["nspd_engineering_nearby"] == [] # ── get_quarter_dump_data — stale ───────────────────────────────────────────── def test_get_quarter_dump_stale() -> None: """Дамп старше 180 дней → stale=True, harvest triggered, available=False.""" old_date = datetime.now(UTC) - timedelta(days=200) # row: (quarter_cad, fetched_at_utc, total_features, harvest_error, # territorial_zones_count, zouit_count, engineering_count, # risks_count, opportunity_count, red_lines_count) db = _make_db_mock(row=("66:41:0204016", old_date, 50, None, 5, 2, 1, 0, 0, 0)) with patch( "app.services.site_finder.quarter_dump_lookup._trigger_harvest", return_value=True, ) as mock_trigger: result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt="POLYGON ((...))") mock_trigger.assert_called_once_with("66:41:0204016") assert result["nspd_dump"]["available"] is False assert result["nspd_dump"]["stale"] is True assert result["nspd_dump"]["harvest_triggered"] is True assert result["nspd_dump"]["total_features"] == 50 def test_get_quarter_dump_with_harvest_error() -> None: """Дамп с harvest_error → harvest_triggered=True, available=False.""" fresh_date = datetime.now(UTC) - timedelta(days=10) db = _make_db_mock(row=("66:41:0204016", fresh_date, 0, "WAF 429", 0, 0, 0, 0, 0, 0)) with patch( "app.services.site_finder.quarter_dump_lookup._trigger_harvest", return_value=True, ) as mock_trigger: result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt=None) mock_trigger.assert_called_once_with("66:41:0204016") assert result["nspd_dump"]["available"] is False assert result["nspd_dump"]["harvest_triggered"] is True # ── get_quarter_dump_data — no parcel_wkt ───────────────────────────────────── def test_get_quarter_dump_no_parcel_geometry() -> None: """parcel_wkt=None → metadata only, нет spatial queries.""" fresh_date = datetime.now(UTC) - timedelta(days=5) db = _make_db_mock(row=("66:41:0204016", fresh_date, 120, None, 10, 3, 2, 0, 0, 0)) result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt=None) # execute вызывался только один раз (для чтения строки дампа) assert db.execute.call_count == 1 assert result["nspd_dump"]["available"] is True assert result["nspd_dump"]["stale"] is False assert result["nspd_zoning"] is None assert result["nspd_zouit_overlaps"] == [] assert result["nspd_engineering_nearby"] == [] # ── get_quarter_dump_data — fresh dump with zoning ─────────────────────────── def _make_db_mock_with_spatial( dump_row: Any, zoning_row: Any = None, zouit_rows: list[Any] | None = None, engineering_rows: list[Any] | None = None, ) -> MagicMock: """Mock DB с разными результатами для разных запросов. Порядок вызовов db.execute жёстко привязан к реализации get_quarter_dump_data: 0. dump lookup → .first() 1. zoning → .first() (пропускается если territorial_zones_count == 0) 2. zouit → .fetchall() (пропускается если zouit_count == 0) 3. engineering → .fetchall() (пропускается если engineering_count == 0) Хрупко при реорганизации порядка запросов в production-коде. При изменении порядка в get_quarter_dump_data — обновить индексы call_count здесь. dump_row должен быть 10-элементным tuple: (quarter_cad, fetched_at_utc, total_features, harvest_error, territorial_zones_count, zouit_count, engineering_count, risks_count, opportunity_count, red_lines_count). """ db = MagicMock() call_count = 0 def side_effect(*args: Any, **kwargs: Any) -> MagicMock: nonlocal call_count result_mock = MagicMock() if call_count == 0: # dump lookup result_mock.first.return_value = dump_row elif call_count == 1: # zoning result_mock.first.return_value = zoning_row elif call_count == 2: # zouit result_mock.fetchall.return_value = zouit_rows or [] elif call_count == 3: # engineering result_mock.fetchall.return_value = engineering_rows or [] call_count += 1 return result_mock db.execute.side_effect = side_effect return db def test_get_quarter_dump_fresh_zoning() -> None: """Свежий дамп с territorial_zone → nspd_zoning populated.""" fresh_date = datetime.now(UTC) - timedelta(days=30) dump_row = ("66:41:0204016", fresh_date, 100, None, 5, 0, 0, 0, 0, 0) # zone_props JSONB в виде dict (SQLAlchemy вернёт dict) zone_props: dict[str, Any] = { "reg_numb_border": "Ж-3", "type_zone": "Зона жилой застройки средней этажности", } zoning_row_mock = MagicMock() zoning_row_mock.__getitem__ = lambda self, i: zone_props if i == 0 else None db = _make_db_mock_with_spatial( dump_row=dump_row, zoning_row=zoning_row_mock, zouit_rows=[], engineering_rows=[], ) wkt = "POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.9, 60.6 56.8))" result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt=wkt) assert result["nspd_dump"]["available"] is True assert result["nspd_zoning"] is not None assert result["nspd_zoning"]["zone_code"] == "Ж-3" assert result["nspd_zoning"]["zone_name"] == "Зона жилой застройки средней этажности" assert result["nspd_zoning"]["source"] == "nspd-quarter-dump" def test_get_quarter_dump_fresh_zoning_no_match() -> None: """Свежий дамп, centroid вне всех territorial_zones → nspd_zoning=None.""" fresh_date = datetime.now(UTC) - timedelta(days=30) dump_row = ("66:41:0204016", fresh_date, 100, None, 3, 0, 0, 0, 0, 0) db = _make_db_mock_with_spatial( dump_row=dump_row, zoning_row=None, zouit_rows=[], engineering_rows=[], ) result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt="POLYGON ((...))") assert result["nspd_zoning"] is None assert result["nspd_dump"]["available"] is True # ── get_quarter_dump_data — fresh dump with zouit overlaps ──────────────────── def test_get_quarter_dump_fresh_zouit_overlaps() -> None: """Свежий дамп с 2 zouit → nspd_zouit_overlaps length=2.""" fresh_date = datetime.now(UTC) - timedelta(days=10) # territorial_zones_count > 0 чтобы все 3 spatial запроса вызывались # (иначе call_count в mock смещается и zouit попадает не в тот слот) dump_row = ("66:41:0204016", fresh_date, 200, None, 1, 5, 0, 0, 0, 0) props1: dict[str, Any] = {"name": "ОКН Дом купца", "subcategory": "okn"} props2: dict[str, Any] = {"name": "ОЗ газопровода", "subcategory": "engineering"} def make_row(layer: str, props: dict[str, Any]) -> MagicMock: r = MagicMock() r.__getitem__ = lambda self, i: layer if i == 0 else props return r zouit_rows = [make_row("zouit_okn", props1), make_row("zouit_engineering", props2)] db = _make_db_mock_with_spatial( dump_row=dump_row, zoning_row=None, zouit_rows=zouit_rows, engineering_rows=[], ) result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt="POLYGON ((...))") assert len(result["nspd_zouit_overlaps"]) == 2 keys = {r["group_key"] for r in result["nspd_zouit_overlaps"]} assert "okn" in keys assert "engineering" in keys # ── get_quarter_dump_data — integration of nspd_dump in response ────────────── def test_get_quarter_dump_includes_nspd_fields() -> None: """Базовая проверка что все 4 nspd-ключа присутствуют в возвращаемом dict.""" fresh_date = datetime.now(UTC) - timedelta(days=1) dump_row = ("66:41:0204016", fresh_date, 55, None, 2, 1, 0, 0, 0, 0) db = _make_db_mock_with_spatial( dump_row=dump_row, zoning_row=None, zouit_rows=[], engineering_rows=[], ) result = get_quarter_dump_data(db, "66:41:0204016:5", parcel_wkt="POINT (60.6 56.8)") assert "nspd_zoning" in result assert "nspd_zouit_overlaps" in result assert "nspd_engineering_nearby" in result assert "nspd_dump" in result dump = result["nspd_dump"] assert "available" in dump assert "fetched_at_utc" in dump assert "stale" in dump assert "harvest_triggered" in dump assert "total_features" in dump # ── make_empty_result — mutation safety ───────────────────────────────────── def testmake_empty_result_returns_independent_copies() -> None: """Каждый вызов make_empty_result() возвращает независимый dict. Мутация nspd_dump в одном результате не должна влиять на другой и на модульный sentinel EMPTY_DUMP_RESULT. """ a = make_empty_result() b = make_empty_result() # Вложенный dict — независимый объект assert a["nspd_dump"] is not b["nspd_dump"] assert a["nspd_dump"] is not EMPTY_DUMP_RESULT["nspd_dump"] # Мутируем один — другой не трогается a["nspd_dump"]["harvest_triggered"] = True a["nspd_zouit_overlaps"].append({"group_key": "test"}) assert b["nspd_dump"]["harvest_triggered"] is False assert b["nspd_zouit_overlaps"] == [] assert EMPTY_DUMP_RESULT["nspd_dump"]["harvest_triggered"] is False assert EMPTY_DUMP_RESULT["nspd_zouit_overlaps"] == [] def testmake_empty_result_overrides() -> None: """make_empty_result принимает override-аргументы.""" r = make_empty_result(harvest_triggered=True, stale=True, total_features=42) assert r["nspd_dump"]["harvest_triggered"] is True assert r["nspd_dump"]["stale"] is True assert r["nspd_dump"]["total_features"] == 42 assert r["nspd_dump"]["available"] is False # ── early-exit when layer_counts == 0 ──────────────────────────────────────── def test_early_exit_all_counts_zero_no_spatial_queries() -> None: """Если все denormalized counts == 0 — spatial queries не вызываются.""" fresh_date = datetime.now(UTC) - timedelta(days=5) # territorial_zones_count=0, zouit_count=0, engineering_count=0 dump_row = ("66:41:0204016", fresh_date, 50, None, 0, 0, 0, 0, 0, 0) db = _make_db_mock_with_spatial(dump_row=dump_row) result = get_quarter_dump_data( db, "66:41:0204016", parcel_wkt="POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.8))" ) # 2 execute-вызова: чтение dump row (0) + cad_zouit fallback (#232) — при # zouit_count==0 _get_zouit_overlaps не early-exit'ит, а делает fallback на # cad_zouit. Остальные spatial helpers (zoning/engineering/risk/opportunity/ # red_lines) пропущены по своим нулевым счётчикам. assert db.execute.call_count == 2 assert result["nspd_zoning"] is None assert result["nspd_zouit_overlaps"] == [] assert result["nspd_engineering_nearby"] == [] assert result["nspd_dump"]["available"] is True def test_early_exit_partial_counts() -> None: """territorial_zones_count > 0, zouit_count == 0 → zoning query вызывается, zouit и engineering — нет.""" fresh_date = datetime.now(UTC) - timedelta(days=5) dump_row = ("66:41:0204016", fresh_date, 30, None, 3, 0, 0, 0, 0, 0) zone_props: dict[str, Any] = {"reg_numb_border": "Ж-1", "type_zone": "Зона жилая"} zoning_row_mock = MagicMock() zoning_row_mock.__getitem__ = lambda self, i: zone_props if i == 0 else None db = _make_db_mock_with_spatial( dump_row=dump_row, zoning_row=zoning_row_mock, zouit_rows=[], engineering_rows=[], ) result = get_quarter_dump_data( db, "66:41:0204016", parcel_wkt="POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.8))" ) # dump lookup (0) + zoning (1) + cad_zouit fallback (2) = 3 вызова; # engineering/risk/opportunity/red_lines — пропущены по нулевым счётчикам. # zouit_count==0 → _get_zouit_overlaps делает fallback на cad_zouit (#232), # который возвращает [] (mock fetchall пуст). assert db.execute.call_count == 3 assert result["nspd_zoning"] is not None assert result["nspd_zoning"]["zone_code"] == "Ж-1" assert result["nspd_zouit_overlaps"] == [] assert result["nspd_engineering_nearby"] == []