gendesign/backend/tests/test_nspd_sync.py
lekss361 f3b0ce5687
feat(nspd): harvest_quarter Celery task + beat + admin endpoint (#94 pt.4 part A) (#111)
* feat(nspd): harvest_quarter Celery task + beat + admin endpoint (#94 pt.4/4 part A)

Sprint 1.1 item #3 из #94 part 2. Замыкает harvest cycle: Celery task
оборачивает NSPDClient.search_by_quarter (PR #109) + UPSERT в
nspd_quarter_dumps (PR #110). После — analyze_parcel будет читать dumps
вместо on-demand NSPD HTTP в request-цикле (Sprint 1.1 #4, следующий PR).

Backend:
- New `backend/app/workers/tasks/nspd_sync.py`:
  - `harvest_quarter(quarter_cad, region_code=66, include_zouit=True,
    include_risks=False)` — single-quarter harvest. autoretry_for=
    (NspdLiteWafError,) с retry_backoff=True, max_retries=3,
    soft_time_limit=120s. Generic exception → upsert error row с
    harvest_error=str(exc), counts=0 (НЕ re-raise).
  - `harvest_stale_quarters(region_code, max_age_days=90, batch_size=50)` —
    beat fanout: SELECT cad_quarters_geom EXCEPT fresh dumps → apply_async per cad.
  - UPSERT использует ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON,3857),4326))
    для quarter_geom (schema MultiPolygon strict per database-expert contract).
- `celery_app.py`:
  - Add `nspd_sync` в `include`
  - Beat entry `nspd-harvest-stale-quarters` Mon 04:00 МСК, batch_size=50,
    max_age_days=90
  - `worker_ready` sanity check `SELECT 1 FROM nspd_quarter_dumps LIMIT 0`
    (non-fatal critical log per Bug_Worker_Ready_EarlyReturn_Fixed pattern)
- `admin_scrape.py`:
  - `HarvestQuarterRequest` Pydantic с pattern `^\d+:\d+:\d+$` (quarter only)
  - `POST /api/v1/admin/scrape/nspd/harvest-quarter` manual trigger,
    requires X-Admin-Token, returns task_id.

Tests (`test_nspd_sync.py`): 14 mock-based (no real NSPD/DB):
- happy path: search_by_quarter mock → verify SQL params/structure
- empty quarter → row with quarter_geom=NULL, total=0, no harvest_error
- WAF error → autoretry kicks in (raise re-propagated)
- Generic error → upsert error row
- Fanout: 50 stale → 50 apply_async calls
- Layer names preservation (all 5 zouit + 11 risks if enabled)
- Geometry SRID handling

Code review (code-reviewer pre-push): APPROVE, 0 blocking, 5 minor deferred:
- inline datetime import in error branch (style)
- layers_fetched array literal via string concat (current layer names safe)
- harvest_stale_quarters error rows always re-tried (intended, undocumented)
- test call_args.kwargs fragile (works for current callsite)
- beat collision with refresh-ekb-districts-medians every ~2 years (negligible)

Part of #94. Next: Sprint 1.1 #4 — analyze_parcel reads dump cache.

* fix(nspd-sync): address PR #111 auto-review M1-M5+M7 findings

M1 (runtime bug): remove WHERE region_code from outer cad_quarters_geom SELECT
in harvest_stale_quarters — column does not exist in prod schema
(cad_number, geom, raw_props, fetched_at, source). Would crash on first beat
tick. Inner nspd_quarter_dumps subquery keeps region_code filter unchanged.

M2 (correctness): _upsert_dump теперь делает db.rollback() в except перед
db.close() — prevents PendingRollbackError на следующем session use, если
PostGIS отбракует geometry или JSON cast failure.

M3 (low): worker_ready breadcrumb INSERT тоже теперь имеет rollback —
consistency с остальными session usages в celery_app.py.

M4 (style): layers_fetched — заменил manual string concat
'{' + ','.join(...) + '}' на native Python list. psycopg v3 сам сериализует
list → text[]. Защита от future layer names с запятыми/фигурными скобками.

M5 (style): import datetime as _dt поднят из else branch в top-level imports.

M7 (test cleanup): удалил dead __enter__/__exit__ mock setup в
test_harvest_stale_quarters_fanout / _empty. harvest_stale_quarters не
использует session как context manager — moc'и no-op.

14/14 tests pass. ruff/format clean.

Per auto-review on 9b2a289.

---------

Co-authored-by: lekss361 <claudestars@proton.me>
2026-05-12 18:45:57 +03:00

377 lines
14 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Unit tests для nspd_sync Celery tasks (#94 Sprint 1.1 item #3).
Без сетевых запросов и реальной БД — mock-based. Проверяет:
- harvest_quarter happy path (correct upsert params, return dict)
- harvest_quarter empty quarter (NSPD не вернул квартал — empty row, no crash)
- harvest_quarter WAF retry (NspdLiteWafError → task re-raises → Celery autoretry)
- harvest_quarter generic exception (harvest_error row written)
- harvest_stale_quarters fanout (apply_async N times)
- _build_features_json layers completeness (all 5 ЗОУИТ присутствуют)
"""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from app.services.scrapers.nspd_client import (
NSPDFeature,
NspdLiteWafError,
QuarterDump,
)
from app.workers.tasks.nspd_sync import (
_build_features_json,
_build_risks_count,
_build_zouit_count,
harvest_quarter,
harvest_stale_quarters,
)
# ── Fixtures ──────────────────────────────────────────────────────────────────
def _make_feat(fid: str = "x") -> NSPDFeature:
return NSPDFeature.from_raw({"id": fid, "geometry": None, "properties": {}})
def _make_feat_with_geom(fid: str, geom: dict[str, Any]) -> NSPDFeature:
return NSPDFeature.from_raw({"id": fid, "geometry": geom, "properties": {"k": "v"}})
def _make_dump(
quarter_cad: str = "66:41:0204016",
include_zouit: bool = True,
include_risks: bool = False,
quarter_feat: NSPDFeature | None = None,
bbox: tuple[float, float, float, float] | None = (100.0, 200.0, 300.0, 400.0),
) -> QuarterDump:
"""Фабрика QuarterDump для тестов."""
zouit: dict[str, list[NSPDFeature]] = {}
if include_zouit:
zouit = {
"okn": [_make_feat("ok1"), _make_feat("ok2")],
"engineering": [_make_feat("eng1")],
"natural": [],
"protected": [_make_feat("prot1")],
"other": [],
}
risks: dict[str, list[NSPDFeature]] = {}
if include_risks:
risks = {"flooding": [_make_feat("fl1")], "landslide": []}
return QuarterDump(
quarter_cad=quarter_cad,
quarter=quarter_feat,
parcels=[_make_feat("p1"), _make_feat("p2")],
buildings=[_make_feat("b1")],
territorial_zones=[_make_feat("tz1")],
red_lines=[],
engineering_structures=[_make_feat("e1")],
zouit=zouit,
risks=risks,
layers_fetched=(
"search",
"parcels",
"buildings",
"territorial_zones",
"red_lines",
"engineering_structures",
),
bbox_3857=bbox,
fetched_at_utc="2026-05-12T03:00:00+00:00",
)
# ── _build_features_json tests ────────────────────────────────────────────────
def test_features_json_layers_all_present_with_zouit() -> None:
"""Если include_zouit=True — все 5 ЗОУИТ-групп присутствуют в features_json."""
dump = _make_dump(include_zouit=True)
result = _build_features_json(dump)
layers = {item["layer"] for item in result}
# Core layers
assert "parcels" in layers
assert "buildings" in layers
assert "territorial_zones" in layers
assert "engineering_structures" in layers
# ЗОУИТ — все 5 ключей (даже с пустыми списками они не попадут — только
# populated items)
# okn: 2, engineering: 1, natural: 0, protected: 1, other: 0
assert "zouit_okn" in layers
assert "zouit_engineering" in layers
assert "zouit_protected" in layers
# natural и other — пустые, не должны быть в layers
assert "zouit_natural" not in layers
assert "zouit_other" not in layers
def test_features_json_no_zouit_no_risk_layers() -> None:
"""Без include_zouit — zouit_* и risk_* ключей нет в features_json."""
dump = _make_dump(include_zouit=False, include_risks=False)
result = _build_features_json(dump)
layers = {item["layer"] for item in result}
for layer_name in layers:
assert not layer_name.startswith("zouit_")
assert not layer_name.startswith("risk_")
def test_features_json_each_feature_has_required_keys() -> None:
"""Каждый элемент features_json содержит layer, feature_id, geometry, properties."""
dump = _make_dump(include_zouit=True)
result = _build_features_json(dump)
for item in result:
assert "layer" in item
assert "feature_id" in item
assert "geometry" in item
assert "properties" in item
def test_build_zouit_count_sums_correctly() -> None:
"""_build_zouit_count суммирует все ЗОУИТ-группы."""
dump = _make_dump(include_zouit=True)
# okn:2 + engineering:1 + natural:0 + protected:1 + other:0 = 4
assert _build_zouit_count(dump) == 4
def test_build_risks_count_empty() -> None:
"""_build_risks_count при пустом risks dict = 0."""
dump = _make_dump(include_risks=False)
assert _build_risks_count(dump) == 0
def test_build_risks_count_with_risks() -> None:
"""_build_risks_count суммирует risk-группы."""
dump = _make_dump(include_zouit=False, include_risks=True)
# flooding:1 + landslide:0 = 1
assert _build_risks_count(dump) == 1
# ── harvest_quarter tests ─────────────────────────────────────────────────────
@patch("app.workers.tasks.nspd_sync._upsert_dump")
@patch("app.workers.tasks.nspd_sync.NSPDClient")
def test_harvest_quarter_success(
mock_client_cls: MagicMock,
mock_upsert: MagicMock,
) -> None:
"""Happy path: search_by_quarter → return populated QuarterDump → upsert called."""
dump = _make_dump(include_zouit=True)
mock_client = MagicMock()
mock_client.search_by_quarter.return_value = dump
mock_client_cls.return_value = mock_client
result = harvest_quarter(
"66:41:0204016",
region_code=66,
include_zouit=True,
include_risks=False,
)
assert result["quarter_cad"] == "66:41:0204016"
assert result["region_code"] == 66
assert result["total_features"] == dump.total_features
assert "harvest_duration_ms" in result
assert "error" not in result
# Verify _upsert_dump called once with correct positional args
# Signature: (quarter_cad, region_code, dump, features_json, duration_ms, harvest_error)
assert mock_upsert.call_count == 1
upsert_call = mock_upsert.call_args
assert upsert_call[0][0] == "66:41:0204016" # quarter_cad
assert upsert_call[0][1] == 66 # region_code
assert upsert_call[0][2] is dump # dump object
assert upsert_call[0][5] is None # harvest_error=None on success
@patch("app.workers.tasks.nspd_sync._upsert_dump")
@patch("app.workers.tasks.nspd_sync.NSPDClient")
def test_harvest_quarter_empty_quarter(
mock_client_cls: MagicMock,
mock_upsert: MagicMock,
) -> None:
"""Quarter не найден в NSPD (quarter=None, bbox=None) — empty harvest, no crash."""
empty_dump = QuarterDump(
quarter_cad="66:41:9999999",
quarter=None,
parcels=[],
buildings=[],
territorial_zones=[],
red_lines=[],
engineering_structures=[],
zouit={"okn": [], "engineering": [], "natural": [], "protected": [], "other": []},
risks={},
layers_fetched=("search",),
bbox_3857=None,
fetched_at_utc="2026-05-12T03:00:00+00:00",
)
mock_client = MagicMock()
mock_client.search_by_quarter.return_value = empty_dump
mock_client_cls.return_value = mock_client
result = harvest_quarter("66:41:9999999", region_code=66)
# Should succeed with zero counts
assert result["quarter_cad"] == "66:41:9999999"
assert result["total_features"] == 0
assert "error" not in result
# Upsert called with the empty dump (not None)
# Signature: (quarter_cad, region_code, dump, features_json, duration_ms, harvest_error)
assert mock_upsert.call_count == 1
upsert_args = mock_upsert.call_args[0]
assert upsert_args[2] is empty_dump # dump passed
assert upsert_args[5] is None # harvest_error=None (valid empty harvest)
# features_json should be empty list
features_json_arg = upsert_args[3]
assert features_json_arg == []
@patch("app.workers.tasks.nspd_sync._upsert_dump")
@patch("app.workers.tasks.nspd_sync.NSPDClient")
def test_harvest_quarter_waf_retry(
mock_client_cls: MagicMock,
mock_upsert: MagicMock,
) -> None:
"""NspdLiteWafError → task re-raises (Celery autoretry_for будет retry).
autoretry_for=(NspdLiteWafError,) настроен в декораторе задачи.
Тест проверяет что re-raise происходит (task не глотает WAF).
"""
mock_client = MagicMock()
mock_client.search_by_quarter.side_effect = NspdLiteWafError("HTTP 429 WAF")
mock_client_cls.return_value = mock_client
with pytest.raises(NspdLiteWafError):
harvest_quarter("66:41:0204016", region_code=66)
# _upsert_dump НЕ должен быть вызван при WAF (не пишем error row, retry сделает попытку)
mock_upsert.assert_not_called()
@patch("app.workers.tasks.nspd_sync._upsert_dump")
@patch("app.workers.tasks.nspd_sync.NSPDClient")
def test_harvest_quarter_writes_error_row(
mock_client_cls: MagicMock,
mock_upsert: MagicMock,
) -> None:
"""Generic exception → error row inserted с harvest_error set; task НЕ raise."""
mock_client = MagicMock()
mock_client.search_by_quarter.side_effect = ConnectionError("timeout")
mock_client_cls.return_value = mock_client
result = harvest_quarter("66:41:0204016", region_code=66)
# Task returns error dict instead of raising
assert "error" in result
assert "ConnectionError" in result["error"]
assert result["quarter_cad"] == "66:41:0204016"
# _upsert_dump called with dump=None (error-only row)
# Signature: (quarter_cad, region_code, dump, features_json, duration_ms, harvest_error)
assert mock_upsert.call_count == 1
upsert_args = mock_upsert.call_args[0]
assert upsert_args[2] is None # dump=None
assert upsert_args[3] is None # features_json=None
assert upsert_args[5] is not None # harvest_error set
assert "ConnectionError" in upsert_args[5]
# ── harvest_stale_quarters tests ───────────────────────────────────────────────
@patch("app.workers.tasks.nspd_sync.harvest_quarter")
@patch("app.workers.tasks.nspd_sync.SessionLocal")
def test_harvest_stale_quarters_fanout(
mock_session_cls: MagicMock,
mock_task: MagicMock,
) -> None:
"""Stale cads найдены → harvest_quarter.apply_async вызван N раз."""
stale_cads = ["66:41:0000001", "66:41:0000002", "66:41:0000003"]
# Mock DB session
mock_db = MagicMock()
mock_session_cls.return_value = mock_db
mock_db.execute.return_value.all.return_value = [(c,) for c in stale_cads]
# harvest_quarter.apply_async — проверяем что вызван для каждого cad
mock_task.apply_async = MagicMock()
result = harvest_stale_quarters(region_code=66, max_age_days=90, batch_size=50)
assert result["enqueued"] == 3
assert result["stale_found"] == 3
assert result["region_code"] == 66
assert mock_task.apply_async.call_count == 3
# apply_async(args=[cad, region_code]) — args passed as keyword 'args'
called_cads = [c.kwargs["args"][0] for c in mock_task.apply_async.call_args_list]
assert called_cads == stale_cads
@patch("app.workers.tasks.nspd_sync.harvest_quarter")
@patch("app.workers.tasks.nspd_sync.SessionLocal")
def test_harvest_stale_quarters_empty(
mock_session_cls: MagicMock,
mock_task: MagicMock,
) -> None:
"""Нет stale кварталов → 0 enqueued, apply_async не вызван."""
mock_db = MagicMock()
mock_session_cls.return_value = mock_db
mock_db.execute.return_value.all.return_value = []
mock_task.apply_async = MagicMock()
result = harvest_stale_quarters(region_code=66)
assert result["enqueued"] == 0
assert result["stale_found"] == 0
mock_task.apply_async.assert_not_called()
# ── features_json layer names tests ──────────────────────────────────────────
def test_features_json_risk_layer_names() -> None:
"""Risk layers кодируются как risk_<short_name>."""
dump = _make_dump(include_zouit=False, include_risks=True)
result = _build_features_json(dump)
layers = {item["layer"] for item in result}
# flooding:1 item
assert "risk_flooding" in layers
# landslide:0 items — не должно быть ключа
assert "risk_landslide" not in layers
def test_features_json_geometry_preserved() -> None:
"""Geometry (EPSG:3857) сохраняется без трансформации в features_json."""
geom = {
"type": "Polygon",
"coordinates": [
[[100.0, 200.0], [300.0, 200.0], [300.0, 400.0], [100.0, 400.0], [100.0, 200.0]]
],
}
dump = QuarterDump(
quarter_cad="66:41:0204016",
quarter=None,
parcels=[_make_feat_with_geom("p1", geom)],
buildings=[],
territorial_zones=[],
red_lines=[],
engineering_structures=[],
zouit={},
risks={},
layers_fetched=("search", "parcels"),
bbox_3857=(100.0, 200.0, 300.0, 400.0),
fetched_at_utc="2026-05-12T00:00:00+00:00",
)
result = _build_features_json(dump)
assert len(result) == 1
assert result[0]["geometry"] == geom # not transformed
assert result[0]["layer"] == "parcels"
assert result[0]["feature_id"] == "p1"
assert result[0]["properties"] == {"k": "v"}