lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(nspd): harvest_quarter Celery task + beat + admin endpoint (#94 pt.4/4 part A)

Browse source 
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.
This commit is contained in:
lekss361 2026-05-12 18:33:09 +03:00
parent 6fce81e10c
commit 9b2a289e83
4 changed files with 866 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
backend/app/api/v1/admin_scrape.py
View file

@ -384,6 +384,50 @@ def trigger_noise_sync(
return {"task_id": result.id, "queued_at": "now"}
class HarvestQuarterRequest(BaseModel):
"""Параметры ручного запуска harvest одного квартала."""
quarter_cad: str = Field(
...,
min_length=5,
pattern=r"^\d+:\d+:\d+$",
description="3-сегментный кадастровый номер квартала, например '66:41:0204016'",
)
region_code: int = Field(default=66, ge=1, le=99)
include_zouit: bool = Field(default=True)
include_risks: bool = Field(default=False)
@router.post("/nspd/harvest-quarter")
def trigger_harvest_quarter(
payload: HarvestQuarterRequest,
x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
"""Manual trigger одного quarter harvest (для testing / конкретного квартала).
Запускает harvest_quarter Celery task и возвращает task_id.
Для получения результата — poll через Celery inspect или нождите строку
в nspd_quarter_dumps WHERE quarter_cad = payload.quarter_cad.
"""
_check_token(x_admin_token)
from app.workers.tasks.nspd_sync import harvest_quarter
task = harvest_quarter.apply_async(
args=[payload.quarter_cad, payload.region_code],
kwargs={
"include_zouit": payload.include_zouit,
"include_risks": payload.include_risks,
},
)
return {
"task_id": task.id,
"quarter_cad": payload.quarter_cad,
"region_code": payload.region_code,
"include_zouit": payload.include_zouit,
"include_risks": payload.include_risks,
}
@router.post("/pzz-sync")
def trigger_pzz_sync(
x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,

34
backend/app/workers/celery_app.py
View file

@ -53,6 +53,7 @@ celery_app = Celery(
"app.workers.tasks.scrape_objective",
"app.workers.tasks.objective_etl",
"app.workers.tasks.nspd_geo",
"app.workers.tasks.nspd_sync",
"app.workers.tasks.poi_sync",
"app.workers.tasks.noise_sync",
"app.workers.tasks.pzz_sync",
@ -230,6 +231,17 @@ def _build_beat_schedule() -> dict:
"options": {"queue": "celery"},
}
# NSPD quarter dump refresh — понедельник 04:00 МСК.
# Fanout'ит harvest_quarter.apply_async для кварталов старше 90 дней или
# отсутствующих в nspd_quarter_dumps. batch_size=50 — ~5 мин при 600ms rate.
# Запускается через час после poi-sync чтобы не создавать конкуренцию.
schedule["nspd-harvest-stale-quarters"] = {
"task": "tasks.nspd_sync.harvest_stale_quarters",
"schedule": _parse_cron("0 4 * * mon"),
"kwargs": {"region_code": 66, "max_age_days": 90, "batch_size": 50},
"options": {"queue": "celery"},
}
return schedule
@ -406,3 +418,25 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None:
except Exception as e:
logger.warning("worker_ready: failed to enqueue geo resume job=%s: %s", jid, e)
logger.info("worker_ready: resume scan finished (geo_jobs=%d)", len(geo_resume_jobs))
# Sanity check: nspd_quarter_dumps table must exist (migration 88).
# Logs critical error but does NOT crash the worker — table may be absent
# in dev/staging before migration is applied.
try:
from app.core.db import SessionLocal as _SessionLocal
_db = _SessionLocal()
try:
_db.execute(text("SELECT 1 FROM nspd_quarter_dumps LIMIT 0"))
logger.info("worker_ready: nspd_quarter_dumps table OK")
except Exception as _te:
logger.critical(
"worker_ready: nspd_quarter_dumps table missing or inaccessible — "
"apply migration 88_nspd_quarter_dumps.sql before using harvest_quarter. "
"Error: %s",
_te,
)
finally:
_db.close()
except Exception as _e:
logger.warning("worker_ready: nspd_quarter_dumps sanity check failed: %s", _e)

407
backend/app/workers/tasks/nspd_sync.py Normal file
View file

@ -0,0 +1,407 @@
"""Celery tasks для NSPD quarter harvest.
Архитектура (Sprint 1.1 item #3):
harvest_quarter — один квартал: search_by_quarter → UPSERT nspd_quarter_dumps.
harvest_stale_quarters — beat-triggered bulk: ищет устаревшие/отсутствующие дампы,
фанаут harvest_quarter.apply_async() на каждый.
Sanity check при startup:
worker_ready сигнал проверяет наличие таблицы nspd_quarter_dumps. Если отсутствует
(migration 88_nspd_quarter_dumps.sql не применена) — пишет critical log, не падает.
UPSERT pattern:
ON CONFLICT (quarter_cad) DO UPDATE — обновляет все колонки кроме region_code
(иммутабельный per-row, задаётся при первом insert).
Геометрия квартала: ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(...),3857),4326))
Bbox: ST_MakeEnvelope(xmin,ymin,xmax,ymax,3857).
"""
from __future__ import annotations
import json
import logging
import time
from typing import Any
from sqlalchemy import text
from app.core.db import SessionLocal
from app.services.scrapers.nspd_client import (
NSPDClient,
NSPDFeature,
NspdLiteWafError,
QuarterDump,
)
from app.workers.celery_app import celery_app
logger = logging.getLogger(__name__)
# ── Geometry helpers ──────────────────────────────────────────────────────────
def _build_features_json(dump: QuarterDump) -> list[dict[str, Any]]:
"""Собрать все NSPDFeature из QuarterDump в плоский list[dict] для JSONB.
Каждый элемент: {layer, feature_id, geometry, properties}.
Порядок: core layers → zouit layers → risk layers (для консистентности).
"""
def _feat_to_dict(layer: str, f: NSPDFeature) -> dict[str, Any]:
return {
"layer": layer,
"feature_id": f.feature_id,
"geometry": f.geometry,
"properties": f.properties,
}
out: list[dict[str, Any]] = []
# Core layers
for feat in dump.parcels:
out.append(_feat_to_dict("parcels", feat))
for feat in dump.buildings:
out.append(_feat_to_dict("buildings", feat))
for feat in dump.territorial_zones:
out.append(_feat_to_dict("territorial_zones", feat))
for feat in dump.red_lines:
out.append(_feat_to_dict("red_lines", feat))
for feat in dump.engineering_structures:
out.append(_feat_to_dict("engineering_structures", feat))
# ЗОУИТ groups — keys: okn, engineering, natural, protected, other
for short_name, features in dump.zouit.items():
layer_name = f"zouit_{short_name}"
for feat in features:
out.append(_feat_to_dict(layer_name, feat))
# Risk groups — keys: flooding_underground, flooding, swampification, ...
for short_name, features in dump.risks.items():
layer_name = f"risk_{short_name}"
for feat in features:
out.append(_feat_to_dict(layer_name, feat))
return out
def _build_zouit_count(dump: QuarterDump) -> int:
"""Сумма features по всем 5 ЗОУИТ группам."""
return sum(len(v) for v in dump.zouit.values())
def _build_risks_count(dump: QuarterDump) -> int:
"""Сумма features по всем risk группам."""
return sum(len(v) for v in dump.risks.values())
# ── UPSERT helper ─────────────────────────────────────────────────────────────
_UPSERT_SQL = text(
"""
INSERT INTO nspd_quarter_dumps (
quarter_cad, quarter_geom, bbox_3857,
parcels_count, buildings_count, territorial_zones_count,
red_lines_count, engineering_count, zouit_count, risks_count, total_features,
features_json, layers_fetched, fetched_at_utc, harvest_duration_ms,
harvest_error, region_code
) VALUES (
:quarter_cad,
CASE WHEN :geom_json IS NULL THEN NULL
ELSE ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom_json), 3857), 4326))
END,
CASE WHEN :bbox_xmin IS NULL THEN NULL
ELSE ST_MakeEnvelope(:bbox_xmin, :bbox_ymin, :bbox_xmax, :bbox_ymax, 3857)
END,
:parcels_count, :buildings_count, :territorial_zones_count,
:red_lines_count, :engineering_count, :zouit_count, :risks_count, :total_features,
CAST(:features_json AS jsonb),
CAST(:layers_fetched AS text[]),
:fetched_at_utc::timestamptz,
:harvest_duration_ms,
:harvest_error,
:region_code
)
ON CONFLICT (quarter_cad) DO UPDATE SET
quarter_geom = EXCLUDED.quarter_geom,
bbox_3857 = EXCLUDED.bbox_3857,
parcels_count = EXCLUDED.parcels_count,
buildings_count = EXCLUDED.buildings_count,
territorial_zones_count = EXCLUDED.territorial_zones_count,
red_lines_count = EXCLUDED.red_lines_count,
engineering_count = EXCLUDED.engineering_count,
zouit_count = EXCLUDED.zouit_count,
risks_count = EXCLUDED.risks_count,
total_features = EXCLUDED.total_features,
features_json = EXCLUDED.features_json,
layers_fetched = EXCLUDED.layers_fetched,
fetched_at_utc = EXCLUDED.fetched_at_utc,
harvest_duration_ms = EXCLUDED.harvest_duration_ms,
harvest_error = EXCLUDED.harvest_error
-- region_code исключён из UPDATE: иммутабельное свойство квартала
"""
)
def _upsert_dump(
quarter_cad: str,
region_code: int,
dump: QuarterDump | None,
features_json: list[dict[str, Any]] | None,
duration_ms: int | None,
harvest_error: str | None,
) -> None:
"""UPSERT строки в nspd_quarter_dumps.
dump=None при partial-failure: вставляем строку с harvest_error + нулевыми счётчиками.
"""
db = SessionLocal()
try:
if dump is not None:
geom_json: str | None = (
json.dumps(dump.quarter.geometry, ensure_ascii=False)
if dump.quarter and dump.quarter.geometry
else None
)
bbox = dump.bbox_3857
params: dict[str, Any] = {
"quarter_cad": quarter_cad,
"geom_json": geom_json,
"bbox_xmin": bbox[0] if bbox else None,
"bbox_ymin": bbox[1] if bbox else None,
"bbox_xmax": bbox[2] if bbox else None,
"bbox_ymax": bbox[3] if bbox else None,
"parcels_count": len(dump.parcels),
"buildings_count": len(dump.buildings),
"territorial_zones_count": len(dump.territorial_zones),
"red_lines_count": len(dump.red_lines),
"engineering_count": len(dump.engineering_structures),
"zouit_count": _build_zouit_count(dump),
"risks_count": _build_risks_count(dump),
"total_features": dump.total_features,
"features_json": json.dumps(features_json or [], ensure_ascii=False),
"layers_fetched": "{" + ",".join(dump.layers_fetched) + "}",
"fetched_at_utc": dump.fetched_at_utc,
"harvest_duration_ms": duration_ms,
"harvest_error": harvest_error,
"region_code": region_code,
}
else:
# Error-only row: нулевые счётчики, harvest_error filled
import datetime as _dt
params = {
"quarter_cad": quarter_cad,
"geom_json": None,
"bbox_xmin": None,
"bbox_ymin": None,
"bbox_xmax": None,
"bbox_ymax": None,
"parcels_count": 0,
"buildings_count": 0,
"territorial_zones_count": 0,
"red_lines_count": 0,
"engineering_count": 0,
"zouit_count": 0,
"risks_count": 0,
"total_features": 0,
"features_json": "[]",
"layers_fetched": "{}",
"fetched_at_utc": _dt.datetime.now(_dt.UTC).isoformat(),
"harvest_duration_ms": duration_ms,
"harvest_error": harvest_error,
"region_code": region_code,
}
db.execute(_UPSERT_SQL, params)
db.commit()
finally:
db.close()
# ── Main task ─────────────────────────────────────────────────────────────────
@celery_app.task(
bind=True,
name="tasks.nspd_sync.harvest_quarter",
max_retries=3,
soft_time_limit=120, # 2 мин — worst-case 22 HTTP × ~600ms = ~13s + margin
autoretry_for=(NspdLiteWafError,),
retry_backoff=True,
retry_backoff_max=120,
retry_jitter=True,
)
def harvest_quarter(
self: Any,
quarter_cad: str,
region_code: int = 66,
include_zouit: bool = True,
include_risks: bool = False,
) -> dict[str, Any]:
"""Single-quarter harvest. NSPDClient.search_by_quarter → UPSERT nspd_quarter_dumps.
Идемпотентен: повторный вызов обновляет строку (ON CONFLICT DO UPDATE).
WAF 403/429 → autoretry с exponential backoff (max 3 попытки).
Другие исключения → запись harvest_error в строку, return error dict (не raise).
"""
t0 = time.monotonic()
logger.info(
"harvest_quarter start: cad=%s region=%d include_zouit=%s include_risks=%s",
quarter_cad,
region_code,
include_zouit,
include_risks,
)
client = NSPDClient()
dump: QuarterDump | None = None
features_json: list[dict[str, Any]] | None = None
try:
dump = client.search_by_quarter(
quarter_cad,
include_zouit=include_zouit,
include_risks=include_risks,
)
features_json = _build_features_json(dump)
duration_ms = int((time.monotonic() - t0) * 1000)
_upsert_dump(quarter_cad, region_code, dump, features_json, duration_ms, None)
logger.info(
"harvest_quarter done: cad=%s region=%d duration=%dms total=%d",
quarter_cad,
region_code,
duration_ms,
dump.total_features,
)
return {
"quarter_cad": quarter_cad,
"region_code": region_code,
"total_features": dump.total_features,
"parcels_count": len(dump.parcels),
"buildings_count": len(dump.buildings),
"harvest_duration_ms": duration_ms,
"layers_fetched": list(dump.layers_fetched),
"bbox_3857": dump.bbox_3857,
}
except NspdLiteWafError:
# autoretry_for перехватит и сделает retry с backoff.
# Просто re-raise чтобы Celery знал о WAF.
duration_ms = int((time.monotonic() - t0) * 1000)
logger.warning(
"harvest_quarter WAF: cad=%s attempt=%d duration=%dms",
quarter_cad,
self.request.retries,
duration_ms,
)
raise
except Exception as exc:
duration_ms = int((time.monotonic() - t0) * 1000)
err_str = f"{type(exc).__name__}: {exc}"
logger.error(
"harvest_quarter error: cad=%s region=%d duration=%dms error=%s",
quarter_cad,
region_code,
duration_ms,
err_str,
)
# Пишем строку с harvest_error — не даём silent failure.
try:
_upsert_dump(quarter_cad, region_code, None, None, duration_ms, err_str[:500])
except Exception as db_exc:
logger.error(
"harvest_quarter: failed to write error row for cad=%s: %s",
quarter_cad,
db_exc,
)
return {
"quarter_cad": quarter_cad,
"region_code": region_code,
"error": err_str,
"harvest_duration_ms": duration_ms,
}
# ── Bulk beat-triggered task ──────────────────────────────────────────────────
@celery_app.task(name="tasks.nspd_sync.harvest_stale_quarters")
def harvest_stale_quarters(
region_code: int = 66,
max_age_days: int = 90,
batch_size: int = 50,
) -> dict[str, Any]:
"""Beat-triggered: найти устаревшие/отсутствующие дампы и сфанаутить harvest_quarter.
Стратегия:
Берём кадастровые кварталы из cad_quarters_geom WHERE region_code=:rc
которых нет в nspd_quarter_dumps (или есть, но старше max_age_days суток
или имеют harvest_error).
Limit batch_size — чтобы не сжечь весь rate-limit за один запуск beat.
Fanout: harvest_quarter.apply_async(args=[cad, region_code]).
Идемпотентен: следующий запуск подхватит остатки если batch_size < stale total.
"""
logger.info(
"harvest_stale_quarters: region=%d max_age_days=%d batch_size=%d",
region_code,
max_age_days,
batch_size,
)
db = SessionLocal()
stale_cads: list[str] = []
try:
rows = db.execute(
text(
"""
SELECT cad_number
FROM cad_quarters_geom
WHERE region_code = :rc
AND cad_number NOT IN (
SELECT quarter_cad
FROM nspd_quarter_dumps
WHERE region_code = :rc
AND fetched_at_utc > NOW() - INTERVAL '1 day' * :days
AND harvest_error IS NULL
)
ORDER BY cad_number
LIMIT :lim
"""
),
{"rc": region_code, "days": max_age_days, "lim": batch_size},
).all()
stale_cads = [r[0] for r in rows]
except Exception as e:
logger.error("harvest_stale_quarters: DB query failed: %s", e)
try:
db.rollback()
except Exception:
pass
return {"error": str(e), "enqueued": 0}
finally:
db.close()
enqueued = 0
for cad in stale_cads:
try:
harvest_quarter.apply_async(args=[cad, region_code])
enqueued += 1
except Exception as e:
logger.warning("harvest_stale_quarters: enqueue failed for cad=%s: %s", cad, e)
logger.info(
"harvest_stale_quarters done: region=%d enqueued=%d stale_found=%d",
region_code,
enqueued,
len(stale_cads),
)
return {
"enqueued": enqueued,
"stale_found": len(stale_cads),
"region_code": region_code,
"max_age_days": max_age_days,
}

381
backend/tests/test_nspd_sync.py Normal file
View file

@ -0,0 +1,381 @@
"""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.__enter__ = MagicMock(return_value=mock_db)
mock_db.__exit__ = MagicMock(return_value=False)
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.__enter__ = MagicMock(return_value=mock_db)
mock_db.__exit__ = MagicMock(return_value=False)
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"}