fdb54834e5
- data/sql/98_*: rename from 89 to avoid collision with existing 89_drop_dead_brin - _get_red_lines: ST_Length(ST_Intersection(planar)::geography) instead of ST_Area(ST_Intersection(::geography, ::geography)) — fixes PostGIS 3.4 tolerance error and returns correct non-zero length for LINESTRING intersections - Rename intersection_area_sqm → intersection_length_m across schema, TS types, frontend component, and tests; add test_get_red_lines_db_exception_returns_empty Addresses review-bot feedback on PR #220.
404 lines
15 KiB
Python
404 lines
15 KiB
Python
"""Тесты для quarter_dump_lookup.py — risk zones + generic layer extraction + TIER 4.
|
||
|
||
Покрывает:
|
||
- _extract_features_by_layer: filter by prefix
|
||
- _get_risk_zones / extract through get_quarter_dump_data: intersect, no-intersect,
|
||
no-risks-in-dump
|
||
- _get_opportunity_parcels: auction/scheme/free/future/oopt layers, distance sort, empty
|
||
- _get_red_lines: intersecting, nearby-only, empty, early-exit
|
||
- EMPTY_DUMP_RESULT / make_empty_result: new keys присутствуют
|
||
Не требует реального PostgreSQL — мокает db.execute().
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
|
||
from app.services.site_finder.quarter_dump_lookup import (
|
||
EMPTY_DUMP_RESULT,
|
||
_extract_features_by_layer,
|
||
_get_opportunity_parcels,
|
||
_get_red_lines,
|
||
_get_risk_zones,
|
||
derive_quarter_cad,
|
||
make_empty_result,
|
||
)
|
||
|
||
# ── Fixtures & helpers ────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_feature(layer: str, geom: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
"""Минимальный feature dict в формате features_json."""
|
||
return {
|
||
"layer": layer,
|
||
"feature_id": "test-id",
|
||
"geometry": geom or {"type": "Polygon", "coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]]},
|
||
"properties": {"name": f"test-{layer}", "type_zone": f"zone-{layer}"},
|
||
}
|
||
|
||
|
||
# ── _extract_features_by_layer ────────────────────────────────────────────────
|
||
|
||
|
||
def test_extract_features_by_layer_prefix_match() -> None:
|
||
"""Возвращает только features с нужным prefix."""
|
||
features = [
|
||
_make_feature("risk_flooding"),
|
||
_make_feature("risk_burns"),
|
||
_make_feature("zouit_engineering"),
|
||
_make_feature("parcels"),
|
||
_make_feature("risk_landslide"),
|
||
]
|
||
result = _extract_features_by_layer(features, "risk_")
|
||
assert len(result) == 3
|
||
layers = {f["layer"] for f in result}
|
||
assert layers == {"risk_flooding", "risk_burns", "risk_landslide"}
|
||
|
||
|
||
def test_extract_features_by_layer_zouit_prefix() -> None:
|
||
"""Работает для zouit_ prefix — generic re-use."""
|
||
features = [
|
||
_make_feature("zouit_engineering"),
|
||
_make_feature("zouit_okn"),
|
||
_make_feature("risk_flooding"),
|
||
]
|
||
result = _extract_features_by_layer(features, "zouit_")
|
||
assert len(result) == 2
|
||
assert all(f["layer"].startswith("zouit_") for f in result)
|
||
|
||
|
||
def test_extract_features_by_layer_empty_list() -> None:
|
||
result = _extract_features_by_layer([], "risk_")
|
||
assert result == []
|
||
|
||
|
||
def test_extract_features_by_layer_no_match() -> None:
|
||
features = [_make_feature("parcels"), _make_feature("buildings")]
|
||
result = _extract_features_by_layer(features, "risk_")
|
||
assert result == []
|
||
|
||
|
||
def test_extract_features_by_layer_non_string_layer() -> None:
|
||
"""Gracefully skips features with non-string layer."""
|
||
features: list[dict[str, Any]] = [
|
||
{"layer": None, "feature_id": "x", "geometry": None, "properties": {}},
|
||
_make_feature("risk_flooding"),
|
||
]
|
||
result = _extract_features_by_layer(features, "risk_")
|
||
assert len(result) == 1
|
||
assert result[0]["layer"] == "risk_flooding"
|
||
|
||
|
||
# ── _get_risk_zones ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_mock_db_with_rows(rows: list[tuple[Any, ...]]) -> MagicMock:
|
||
"""Мок Session где execute().fetchall() возвращает rows."""
|
||
db = MagicMock()
|
||
mock_result = MagicMock()
|
||
mock_result.fetchall.return_value = rows
|
||
db.execute.return_value = mock_result
|
||
return db
|
||
|
||
|
||
def test_get_risk_zones_intersects() -> None:
|
||
"""Три risk features → три записи в результате."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (layer, props, geom_wkt, intersection_area_sqm)
|
||
(
|
||
"risk_flooding",
|
||
{"type_zone": "Затопление 1%"},
|
||
"POLYGON((0 0,1 0,1 1,0 0))",
|
||
1234.5,
|
||
),
|
||
(
|
||
"risk_burns",
|
||
{"name": "Гарь"},
|
||
"POLYGON((0 0,2 0,2 2,0 0))",
|
||
500.0,
|
||
),
|
||
(
|
||
"risk_landslide",
|
||
{},
|
||
"POLYGON((0 0,3 0,3 3,0 0))",
|
||
999.9,
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 3
|
||
|
||
flooding = next(r for r in result if r["layer"] == "risk_flooding")
|
||
assert flooding["subtype"] == "Затопление 1%"
|
||
assert flooding["intersection_area_sqm"] == 1234.5
|
||
assert flooding["geom_wkt"] == "POLYGON((0 0,1 0,1 1,0 0))"
|
||
|
||
burns = next(r for r in result if r["layer"] == "risk_burns")
|
||
assert burns["subtype"] == "Гарь"
|
||
|
||
landslide = next(r for r in result if r["layer"] == "risk_landslide")
|
||
# No properties name/type_zone → falls back to _RISK_SUBTYPE_LABELS
|
||
assert landslide["subtype"] == "Обвально-осыпные процессы"
|
||
|
||
|
||
def test_get_risk_zones_no_intersect() -> None:
|
||
"""Пустой fetchall → пустой список."""
|
||
db = _make_mock_db_with_rows([])
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_risk_zones_no_risks_in_dump() -> None:
|
||
"""risks_count == 0 → early exit, db.execute не вызывается."""
|
||
db = MagicMock()
|
||
layer_counts = {"risks_count": 0}
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))", layer_counts)
|
||
assert result == []
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
def test_get_risk_zones_db_exception_returns_empty() -> None:
|
||
"""DB exception → пустой список (не propagate)."""
|
||
db = MagicMock()
|
||
db.execute.side_effect = Exception("DB connection lost")
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_risk_zones_null_area() -> None:
|
||
"""NULL intersection_area_sqm → intersection_area_sqm=None в результате."""
|
||
rows = [("risk_erosion_water", {}, "POINT(0 0)", None)]
|
||
db = _make_mock_db_with_rows(rows)
|
||
result = _get_risk_zones(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert len(result) == 1
|
||
assert result[0]["intersection_area_sqm"] is None
|
||
|
||
|
||
# ── make_empty_result includes nspd_risk_zones ────────────────────────────────
|
||
|
||
|
||
def test_make_empty_result_has_risk_zones_key() -> None:
|
||
"""make_empty_result возвращает nspd_risk_zones: []."""
|
||
result = make_empty_result()
|
||
assert "nspd_risk_zones" in result
|
||
assert result["nspd_risk_zones"] == []
|
||
|
||
|
||
# ── _get_opportunity_parcels ─────────────────────────────────────────────────
|
||
|
||
|
||
def test_get_opportunity_parcels_finds_auction() -> None:
|
||
"""3 features (1 auction + 2 other types) — все возвращаются, auction в нужном layer."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (layer, props, geom_wkt, distance_m)
|
||
(
|
||
"opportunity_auction_parcels",
|
||
{"cad_num": "66:41:0204016:101"},
|
||
"POLYGON((0 0,1 0,1 1,0 0))",
|
||
50.0,
|
||
),
|
||
(
|
||
"opportunity_scheme_parcels",
|
||
{"cadastral_number": "66:41:0204016:102"},
|
||
"POLYGON((1 1,2 1,2 2,1 1))",
|
||
150.0,
|
||
),
|
||
(
|
||
"opportunity_oopt",
|
||
{},
|
||
"POLYGON((2 2,3 2,3 3,2 2))",
|
||
300.0,
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_opportunity_parcels(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 3
|
||
|
||
auction = next(r for r in result if r["layer"] == "auction_parcels")
|
||
assert auction["cad_num"] == "66:41:0204016:101"
|
||
assert auction["distance_m"] == 50.0
|
||
|
||
scheme = next(r for r in result if r["layer"] == "scheme_parcels")
|
||
assert scheme["cad_num"] == "66:41:0204016:102"
|
||
|
||
oopt = next(r for r in result if r["layer"] == "oopt")
|
||
assert oopt["cad_num"] is None
|
||
assert oopt["distance_m"] == 300.0
|
||
|
||
|
||
def test_get_opportunity_parcels_distance_sort() -> None:
|
||
"""Результаты отсортированы по distance_m ASC (SQL ORDER BY передаётся DB)."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# Порядок: ближайший первым (SQL уже сортирует — тест проверяет что порядок сохраняется)
|
||
(
|
||
"opportunity_free_parcels",
|
||
{"cad_num": "66:41:0204016:10"},
|
||
"POINT(0 0)",
|
||
10.0,
|
||
),
|
||
(
|
||
"opportunity_auction_parcels",
|
||
{"cad_num": "66:41:0204016:20"},
|
||
"POINT(1 1)",
|
||
200.0,
|
||
),
|
||
(
|
||
"opportunity_future_parcels",
|
||
{},
|
||
"POINT(2 2)",
|
||
450.0,
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_opportunity_parcels(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 3
|
||
# Порядок из DB (mock возвращает в порядке rows) — ближайший первый
|
||
assert result[0]["distance_m"] == 10.0
|
||
assert result[1]["distance_m"] == 200.0
|
||
assert result[2]["distance_m"] == 450.0
|
||
|
||
|
||
def test_get_opportunity_parcels_empty() -> None:
|
||
"""Нет opportunity features — пустой список."""
|
||
db = _make_mock_db_with_rows([])
|
||
result = _get_opportunity_parcels(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_opportunity_parcels_early_exit() -> None:
|
||
"""opportunity_count == 0 → early exit, db.execute не вызывается."""
|
||
db = MagicMock()
|
||
layer_counts = {"opportunity_count": 0}
|
||
result = _get_opportunity_parcels(
|
||
db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))", layer_counts
|
||
)
|
||
assert result == []
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
def test_get_opportunity_parcels_db_exception_returns_empty() -> None:
|
||
"""DB exception → пустой список (не propagate)."""
|
||
db = MagicMock()
|
||
db.execute.side_effect = Exception("connection lost")
|
||
result = _get_opportunity_parcels(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
# ── _get_red_lines ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_get_red_lines_intersects() -> None:
|
||
"""Red line intersecting parcel → intersection_length_m filled, distance_m=None."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (geom_wkt, does_intersect, intersection_length_m, distance_m)
|
||
(
|
||
"LINESTRING(0 0, 1 1)",
|
||
True,
|
||
45.3, # длина пересечения в метрах
|
||
0.0, # intersecting → distance_m becomes None in result
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 1
|
||
r = result[0]
|
||
assert r["geom_wkt"] == "LINESTRING(0 0, 1 1)"
|
||
assert r["intersection_length_m"] == 45.3
|
||
assert r["distance_m"] is None # null when intersecting
|
||
|
||
|
||
def test_get_red_lines_nearby_only() -> None:
|
||
"""Red line nearby only (не intersect) → distance_m filled, intersection_length_m=None."""
|
||
rows: list[tuple[Any, ...]] = [
|
||
# (geom_wkt, does_intersect, intersection_length_m, distance_m)
|
||
(
|
||
"LINESTRING(10 10, 20 20)",
|
||
False,
|
||
0.0, # нет пересечения
|
||
85.5,
|
||
),
|
||
]
|
||
db = _make_mock_db_with_rows(rows)
|
||
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
|
||
assert len(result) == 1
|
||
r = result[0]
|
||
assert r["intersection_length_m"] is None
|
||
assert r["distance_m"] == 85.5
|
||
|
||
|
||
def test_get_red_lines_db_exception_returns_empty() -> None:
|
||
"""DB exception → пустой список (не propagate)."""
|
||
db = MagicMock()
|
||
db.execute.side_effect = Exception("DB connection lost")
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_red_lines_empty() -> None:
|
||
"""Нет red lines features — пустой список."""
|
||
db = _make_mock_db_with_rows([])
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))")
|
||
assert result == []
|
||
|
||
|
||
def test_get_red_lines_early_exit() -> None:
|
||
"""red_lines_count == 0 → early exit, db.execute не вызывается."""
|
||
db = MagicMock()
|
||
layer_counts = {"red_lines_count": 0}
|
||
result = _get_red_lines(db, "66:41:0204016", "POLYGON((0 0,1 0,1 1,0 0))", layer_counts)
|
||
assert result == []
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
# ── EMPTY_DUMP_RESULT / make_empty_result includes new TIER 4 keys ────────────
|
||
|
||
|
||
def test_empty_dump_result_has_opportunity_key() -> None:
|
||
"""EMPTY_DUMP_RESULT содержит nspd_opportunity_parcels: []."""
|
||
assert "nspd_opportunity_parcels" in EMPTY_DUMP_RESULT
|
||
assert EMPTY_DUMP_RESULT["nspd_opportunity_parcels"] == []
|
||
|
||
|
||
def test_empty_dump_result_has_red_lines_key() -> None:
|
||
"""EMPTY_DUMP_RESULT содержит nspd_red_lines: []."""
|
||
assert "nspd_red_lines" in EMPTY_DUMP_RESULT
|
||
assert EMPTY_DUMP_RESULT["nspd_red_lines"] == []
|
||
|
||
|
||
def test_make_empty_result_has_tier4_keys() -> None:
|
||
"""make_empty_result() возвращает nspd_opportunity_parcels и nspd_red_lines."""
|
||
result = make_empty_result()
|
||
assert "nspd_opportunity_parcels" in result
|
||
assert result["nspd_opportunity_parcels"] == []
|
||
assert "nspd_red_lines" in result
|
||
assert result["nspd_red_lines"] == []
|
||
|
||
|
||
# ── derive_quarter_cad (smoke tests) ──────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"cad,expected",
|
||
[
|
||
("66:41:0204016", "66:41:0204016"),
|
||
("66:41:0204016:10", "66:41:0204016"),
|
||
("66:41:0204016:10:1", "66:41:0204016"),
|
||
("invalid", None),
|
||
("66:41", None),
|
||
],
|
||
)
|
||
def test_derive_quarter_cad(cad: str, expected: str | None) -> None:
|
||
assert derive_quarter_cad(cad) == expected
|