304 lines
13 KiB
Python
304 lines
13 KiB
Python
"""Тесты harvest_gknspecial_zones (#1063/#1064/#1065/#1066, #1067 GG-форсайт).
|
||
|
||
Чистые тесты без живого геопортала / БД:
|
||
• classify: full_name → корректный (layer_kind, source_layer_id) по каждому префиксу.
|
||
• reg_numb_border regex: извлечение из full_name, мусор → None.
|
||
• geom_data_id: 'portal_geo_gknspecial_zone.12345' → 12345; без точки/нечисло → None.
|
||
• активность: zone_state_v != 'Действующий' → skip; поле отсутствует → не skip.
|
||
• upsert idempotency: mock-клиент + mock-SessionLocal, два прогона → строки не дублируются.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from app.workers.tasks.gknspecial_harvest import (
|
||
_UNMATCHED_LAYER_KIND,
|
||
_UNMATCHED_SOURCE_LAYER_ID,
|
||
_classify_full_name,
|
||
_extract_geom_data_id,
|
||
_extract_reg_numb,
|
||
harvest_gknspecial_zones,
|
||
)
|
||
|
||
# ── classify ─────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_classify_servitude() -> None:
|
||
cls = _classify_full_name("Зона публичного сервитута 66:41-6.100")
|
||
assert cls == ("servitude", 990001)
|
||
|
||
|
||
def test_classify_water_protection() -> None:
|
||
assert _classify_full_name("Водоохранная зона р. Исеть") == ("water_protection", 990002)
|
||
|
||
|
||
def test_classify_coastal_strip() -> None:
|
||
assert _classify_full_name("Прибрежная защитная полоса р.Исеть") == ("coastal_strip", 990003)
|
||
|
||
|
||
def test_classify_water_sanitary() -> None:
|
||
assert _classify_full_name("Зона санитарной охраны ВЗУ") == ("water_sanitary", 990004)
|
||
|
||
|
||
def test_classify_sanitary_protection() -> None:
|
||
assert _classify_full_name("Санитарно-защитная зона завода") == ("sanitary_protection", 990005)
|
||
|
||
|
||
def test_classify_okn_protection_zone() -> None:
|
||
assert _classify_full_name("Зона охраны объекта культурного наследия") == (
|
||
"okn_protection_zone",
|
||
990006,
|
||
)
|
||
|
||
|
||
def test_classify_zrz() -> None:
|
||
assert _classify_full_name("Зона регулирования застройки") == ("zrz", 990007)
|
||
|
||
|
||
def test_classify_airfield() -> None:
|
||
assert _classify_full_name("Приаэродромная территория") == ("airfield", 990008)
|
||
|
||
|
||
def test_classify_airfield_with_suffix() -> None:
|
||
assert _classify_full_name("Приаэродромная территория аэропорта Кольцово") == (
|
||
"airfield",
|
||
990008,
|
||
)
|
||
|
||
|
||
def test_classify_unmatched_returns_none() -> None:
|
||
assert _classify_full_name("Неизвестная зона нового типа") is None
|
||
|
||
|
||
def test_classify_empty_returns_none() -> None:
|
||
assert _classify_full_name("") is None
|
||
|
||
|
||
# ── reg_numb_border regex ─────────────────────────────────────────────────────
|
||
|
||
|
||
def test_reg_numb_extracts_from_full_name() -> None:
|
||
assert _extract_reg_numb("Водоохранная зона 66:41-6.4125 р. Исеть") == "66:41-6.4125"
|
||
|
||
|
||
def test_reg_numb_extracts_with_multiple_digits() -> None:
|
||
assert _extract_reg_numb("Зона охраны 66:01-12.300") == "66:01-12.300"
|
||
|
||
|
||
def test_reg_numb_returns_none_for_garbage() -> None:
|
||
assert _extract_reg_numb("Нет номера границы здесь") is None
|
||
|
||
|
||
def test_reg_numb_returns_none_for_empty() -> None:
|
||
assert _extract_reg_numb("") is None
|
||
|
||
|
||
# ── geom_data_id ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_geom_data_id_normal() -> None:
|
||
assert _extract_geom_data_id("portal_geo_gknspecial_zone.12345") == 12345
|
||
|
||
|
||
def test_geom_data_id_large_number() -> None:
|
||
assert _extract_geom_data_id("portal_geo_gknspecial_zone.9999999") == 9999999
|
||
|
||
|
||
def test_geom_data_id_no_dot() -> None:
|
||
assert _extract_geom_data_id("portal_geo_gknspecial_zone") is None
|
||
|
||
|
||
def test_geom_data_id_non_numeric() -> None:
|
||
assert _extract_geom_data_id("portal_geo_gknspecial_zone.abc") is None
|
||
|
||
|
||
def test_geom_data_id_none_input() -> None:
|
||
assert _extract_geom_data_id(None) is None
|
||
|
||
|
||
def test_geom_data_id_empty_string() -> None:
|
||
assert _extract_geom_data_id("") is None
|
||
|
||
|
||
# ── zone_state_v фильтр активности ───────────────────────────────────────────
|
||
|
||
|
||
def _make_feature(
|
||
feature_id: str = "portal_geo_gknspecial_zone.1",
|
||
full_name: str = "Водоохранная зона",
|
||
zone_state_v: str | None = "Действующий",
|
||
geom: dict[str, Any] | None = None,
|
||
) -> MagicMock:
|
||
"""Фейковый EKBFeature для тестов."""
|
||
from app.services.scrapers.ekb_geoportal_client import EKBFeature
|
||
|
||
props: dict[str, Any] = {"full_name": full_name}
|
||
if zone_state_v is not None:
|
||
props["zone_state_v"] = zone_state_v
|
||
if geom is None:
|
||
geom = {"type": "Polygon", "coordinates": [[[60.5, 56.8], [60.6, 56.8], [60.5, 56.8]]]}
|
||
return EKBFeature(feature_id=feature_id, geometry=geom, properties=props)
|
||
|
||
|
||
def _run_task_with_feats(feats_per_prefix: list[list[Any]]) -> dict[str, int]:
|
||
"""Запустить harvest_gknspecial_zones с мок-клиентом и мок-БД.
|
||
|
||
feats_per_prefix: список из 8 элементов (по числу _PREFIX_MAP), каждый — список EKBFeature.
|
||
"""
|
||
mock_db = MagicMock()
|
||
mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
|
||
mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
|
||
|
||
# Счётчик вызовов features_by_cql → разные фичи для разных префиксов
|
||
call_idx: list[int] = [0]
|
||
|
||
def fake_features_by_cql(layer: str, cql_filter: str) -> list[Any]:
|
||
idx = call_idx[0]
|
||
call_idx[0] += 1
|
||
if idx < len(feats_per_prefix):
|
||
return feats_per_prefix[idx]
|
||
return []
|
||
|
||
with (
|
||
patch("app.workers.tasks.gknspecial_harvest.EKBGeoportalClient") as mock_client_cls,
|
||
patch("app.workers.tasks.gknspecial_harvest.SessionLocal", return_value=mock_db),
|
||
):
|
||
mock_client = MagicMock()
|
||
mock_client.features_by_cql.side_effect = fake_features_by_cql
|
||
mock_client_cls.return_value = mock_client
|
||
return harvest_gknspecial_zones()
|
||
|
||
|
||
def test_skip_archived_zone_state() -> None:
|
||
"""zone_state_v='Архивный' → фича не пишется."""
|
||
feat = _make_feature(zone_state_v="Архивный")
|
||
# 8 префиксов: первый с архивной фичей, остальные пустые
|
||
feats_per_prefix = [[feat]] + [[] for _ in range(7)]
|
||
result = _run_task_with_feats(feats_per_prefix)
|
||
assert result["features"] == 0
|
||
|
||
|
||
def test_include_feature_without_zone_state() -> None:
|
||
"""Нет поля zone_state_v → фича не скипается (неизвестное = активное)."""
|
||
feat = _make_feature(zone_state_v=None)
|
||
feats_per_prefix = [[feat]] + [[] for _ in range(7)]
|
||
result = _run_task_with_feats(feats_per_prefix)
|
||
assert result["features"] == 1
|
||
|
||
|
||
def test_include_active_zone_state() -> None:
|
||
"""zone_state_v='Действующий' → фича пишется."""
|
||
feat = _make_feature(zone_state_v="Действующий")
|
||
feats_per_prefix = [[feat]] + [[] for _ in range(7)]
|
||
result = _run_task_with_feats(feats_per_prefix)
|
||
assert result["features"] == 1
|
||
|
||
|
||
# ── upsert idempotency ───────────────────────────────────────────────────────
|
||
|
||
|
||
def _run_single_feat_task(feat: Any, execute_calls: list[Any]) -> None:
|
||
"""Вспомогательная — один прогон harvest_gknspecial_zones с одной фичей в первом префиксе."""
|
||
feats_per_prefix = [[feat]] + [[] for _ in range(7)]
|
||
call_idx: list[int] = [0]
|
||
|
||
def fake_by_cql(layer: str, cql_filter: str) -> list[Any]:
|
||
idx = call_idx[0]
|
||
call_idx[0] += 1
|
||
return feats_per_prefix[idx] if idx < len(feats_per_prefix) else []
|
||
|
||
mock_db = MagicMock()
|
||
mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
|
||
mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
|
||
mock_db.execute.side_effect = lambda *a, **kw: execute_calls.append(a)
|
||
|
||
with (
|
||
patch("app.workers.tasks.gknspecial_harvest.EKBGeoportalClient") as mock_client_cls,
|
||
patch("app.workers.tasks.gknspecial_harvest.SessionLocal", return_value=mock_db),
|
||
):
|
||
mock_client = MagicMock()
|
||
mock_client.features_by_cql.side_effect = fake_by_cql
|
||
mock_client_cls.return_value = mock_client
|
||
harvest_gknspecial_zones()
|
||
|
||
|
||
def test_upsert_idempotency_no_duplicates() -> None:
|
||
"""Два прогона с одной и той же фичей → db.execute дважды, но ON CONFLICT — не дублируем."""
|
||
feat = _make_feature(
|
||
feature_id="portal_geo_gknspecial_zone.777",
|
||
full_name="Водоохранная зона 66:41-6.100",
|
||
zone_state_v="Действующий",
|
||
)
|
||
execute_calls: list[Any] = []
|
||
_run_single_feat_task(feat, execute_calls)
|
||
_run_single_feat_task(feat, execute_calls)
|
||
|
||
# Оба прогона должны были вызвать db.execute по одному разу (за каждый run по одной фиче).
|
||
# ON CONFLICT в SQL — гарантирует что строка не дублируется на уровне БД.
|
||
# Здесь проверяем: execute вызывался ровно дважды (1 фича × 2 прогона).
|
||
assert len(execute_calls) == 2
|
||
|
||
|
||
def test_returns_types_and_features_counts() -> None:
|
||
"""Возврат {'types': N, 'features': M}."""
|
||
feat1 = _make_feature(feature_id="portal_geo_gknspecial_zone.1", full_name="Водоохранная зона")
|
||
feat2 = _make_feature(feature_id="portal_geo_gknspecial_zone.2", full_name="Водоохранная зона")
|
||
# 2 фичи в первом префиксе, остальные пустые
|
||
feats_per_prefix = [[feat1, feat2]] + [[] for _ in range(7)]
|
||
result = _run_task_with_feats(feats_per_prefix)
|
||
# types = 8 (все 8 префиксов запрошены успешно)
|
||
assert result["types"] == 8
|
||
assert result["features"] == 2
|
||
|
||
|
||
def test_unmatched_feature_written_to_sentinel() -> None:
|
||
"""Фича с full_name не из _PREFIX_MAP → пишется в unmatched (source_layer_id=990099)."""
|
||
feat = _make_feature(
|
||
feature_id="portal_geo_gknspecial_zone.999",
|
||
full_name="Неизвестная зона нового типа",
|
||
zone_state_v=None,
|
||
)
|
||
feats_per_prefix = [[feat]] + [[] for _ in range(7)]
|
||
|
||
written_params: list[dict[str, Any]] = []
|
||
|
||
def make_db() -> MagicMock:
|
||
mock_db = MagicMock()
|
||
mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
|
||
mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
|
||
|
||
def capture_execute(sql: Any, params: dict[str, Any] | None = None) -> None:
|
||
if params:
|
||
written_params.append(params)
|
||
|
||
mock_db.execute.side_effect = capture_execute
|
||
return mock_db
|
||
|
||
call_idx: list[int] = [0]
|
||
|
||
def fake_by_cql(layer: str, cql_filter: str) -> list[Any]:
|
||
idx = call_idx[0]
|
||
call_idx[0] += 1
|
||
return feats_per_prefix[idx] if idx < len(feats_per_prefix) else []
|
||
|
||
with (
|
||
patch("app.workers.tasks.gknspecial_harvest.EKBGeoportalClient") as mock_client_cls,
|
||
patch(
|
||
"app.workers.tasks.gknspecial_harvest.SessionLocal",
|
||
return_value=make_db(),
|
||
),
|
||
):
|
||
mock_client = MagicMock()
|
||
mock_client.features_by_cql.side_effect = fake_by_cql
|
||
mock_client_cls.return_value = mock_client
|
||
result = harvest_gknspecial_zones()
|
||
|
||
assert result["features"] == 1
|
||
assert len(written_params) == 1
|
||
assert written_params[0]["source_layer_id"] == _UNMATCHED_SOURCE_LAYER_ID
|
||
assert written_params[0]["layer_kind"] == _UNMATCHED_LAYER_KIND
|