diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 33a8f77d..431fffe4 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -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, diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index a7d64213..afa18809 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -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 @@ -290,6 +302,9 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: ) ) _db.commit() + except Exception: + _db.rollback() + raise finally: _db.close() except Exception as _e: @@ -406,3 +421,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) diff --git a/backend/app/workers/tasks/nspd_sync.py b/backend/app/workers/tasks/nspd_sync.py new file mode 100644 index 00000000..af6567f1 --- /dev/null +++ b/backend/app/workers/tasks/nspd_sync.py @@ -0,0 +1,408 @@ +"""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 datetime as _dt +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": list(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 + 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() + except Exception: + db.rollback() + raise + 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 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, + } diff --git a/backend/tests/test_nspd_sync.py b/backend/tests/test_nspd_sync.py new file mode 100644 index 00000000..8b077cf6 --- /dev/null +++ b/backend/tests/test_nspd_sync.py @@ -0,0 +1,377 @@ +"""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_.""" + 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"}