Light1YT
50555821fb
Some checks failed
CI / changes (push) Successful in 7s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 7s
CI / frontend-tests (pull_request) Has been skipped
CI / backend-tests (push) Successful in 7m42s
CI / backend-tests (pull_request) Successful in 7m40s
Deploy / changes (push) Has been cancelled
Deploy / build-backend (push) Has been cancelled
Deploy / build-worker (push) Has been cancelled
Deploy / build-frontend (push) Has been cancelled
Deploy / deploy (push) Has been cancelled
#41: per-category routing-radius decay через reusable ORS /matrix client (foot-walk время + per-category пороги + decay-curves), opt-in ?decay=routing, graceful straight-line fallback (analyze не 500 при ORS down) + length-guard durations==destinations. #44: +5 OSM-категорий (transformer/gas/water/sewerage/heat) via Overpass nwr; nearest_{water_main,substation,gas,heat}_m в analyze.utilities. #255bk: ST_AsGeoJSON(CAST(geom AS geometry)) в _get_cad_zouit_overlaps → geom_geojson. Closes #41 Closes #44
474 lines
20 KiB
Python
474 lines
20 KiB
Python
"""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,
|
||
_get_cad_zouit_overlaps,
|
||
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
|
||
|
||
|
||
# ── _get_cad_zouit_overlaps — geom_geojson field (#255) ──────────────────────
|
||
|
||
|
||
def _make_zouit_row(
|
||
type_zone: str,
|
||
category_name: str,
|
||
name: str,
|
||
reg_numb_border: str | None,
|
||
zouit_id: int,
|
||
geom_geojson: str | None,
|
||
) -> tuple[Any, ...]:
|
||
"""Строка cad_zouit в порядке SELECT _get_cad_zouit_overlaps.
|
||
|
||
(type_zone, category_name, name_by_doc, reg_numb_border, id, geom_geojson)
|
||
"""
|
||
return (type_zone, category_name, name, reg_numb_border, zouit_id, geom_geojson)
|
||
|
||
|
||
def test_cad_zouit_overlaps_includes_geom_geojson() -> None:
|
||
"""#255: каждый overlap содержит geom_geojson (string из ST_AsGeoJSON)."""
|
||
geojson_str = (
|
||
'{"type":"Polygon","coordinates":'
|
||
"[[[60.6,56.8],[60.7,56.8],[60.7,56.9],[60.6,56.8]]]}"
|
||
)
|
||
rows = [
|
||
_make_zouit_row(
|
||
"Охранная зона тепловой сети",
|
||
"ЗОУИТ general",
|
||
"Охранная зона тепловой сети с кад. № 66:41:0204016:57",
|
||
"66:41:0204016:57",
|
||
123,
|
||
geojson_str,
|
||
)
|
||
]
|
||
db = MagicMock()
|
||
execute_result = MagicMock()
|
||
execute_result.fetchall.return_value = rows
|
||
db.execute.return_value = execute_result
|
||
|
||
result = _get_cad_zouit_overlaps(db, "POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.8))")
|
||
|
||
assert len(result) == 1
|
||
overlap = result[0]
|
||
assert "geom_geojson" in overlap, "overlap должен нести geom_geojson для ZouitLayer"
|
||
assert overlap["geom_geojson"] == geojson_str
|
||
assert overlap["source"] == "cad_zouit"
|
||
assert overlap["type_zone"] == "Охранная зона тепловой сети"
|
||
assert overlap["raw_props"]["zouit_id"] == 123
|
||
|
||
|
||
def test_cad_zouit_overlaps_geom_geojson_nullable() -> None:
|
||
"""geom_geojson=None (битая/NULL geometry) не валит маппинг — поле = None."""
|
||
rows = [
|
||
_make_zouit_row("СЗЗ", "иные", "Санитарно-защитная зона", None, 7, None),
|
||
]
|
||
db = MagicMock()
|
||
execute_result = MagicMock()
|
||
execute_result.fetchall.return_value = rows
|
||
db.execute.return_value = execute_result
|
||
|
||
result = _get_cad_zouit_overlaps(db, "POLYGON ((...))")
|
||
|
||
assert len(result) == 1
|
||
assert result[0]["geom_geojson"] is None
|
||
|
||
|
||
def test_cad_zouit_overlaps_query_failure_returns_empty() -> None:
|
||
"""Сбой запроса (например, отсутствие cad_zouit) → graceful [] (не пробрасываем)."""
|
||
db = MagicMock()
|
||
db.execute.side_effect = RuntimeError("relation cad_zouit does not exist")
|
||
|
||
result = _get_cad_zouit_overlaps(db, "POLYGON ((...))")
|
||
|
||
assert result == []
|
||
|
||
|
||
# ── 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"] == []
|