fix(cadastre): address bot blockers — pivot Fix H to raw_targets + tests

Bot review on PR #187 caught:
1. cadastre_jobs has no extra_config column → Fix H crashed on KeyError
2. Missing tests for Fix H + Fix I
3. _parse_nspd_date didn't validate date ranges

Changes:
- Pivot Fix H: read skip_fresh_hours from raw_targets JSONB (existing column),
  no schema migration needed
- _parse_nspd_date: validate via date.fromisoformat() — rejects 2023-13-45
- Add 3 tests for Fix H in backend/tests/workers/test_scrape_cadastre.py
- Add 2 tests for Fix I heartbeat callbacks in test_cadastre_bulk.py
- Add 1 test for parse_nspd_date range validation (was: only format)

49 tests passing.
This commit is contained in:
lekss361 2026-05-15 22:23:02 +03:00
parent 726b15ad55
commit f526a43ef9
4 changed files with 233 additions and 8 deletions

View file

@ -1217,19 +1217,30 @@ def _parse_nspd_date(value: Any) -> str | None:
- None / empty string
PostgreSQL DATE cast принимает ISO. Без conversion DD.MM.YYYY DatetimeFieldOverflow
crash в upsert_*.
crash в upsert_*. Также валидируем что результат actually-parseable date чтобы
invalid input не уехал в SQL.
"""
import datetime as _dt
if value is None or value == "":
return None
if not isinstance(value, str):
return None
s = value.strip()
candidate: str | None = None
# ISO format already — pass through (PG accept YYYY-MM-DD)
if len(s) >= 10 and s[4] == "-" and s[7] == "-":
return s[:10]
candidate = s[:10]
# DD.MM.YYYY → YYYY-MM-DD
if len(s) == 10 and s[2] == "." and s[5] == ".":
elif len(s) == 10 and s[2] == "." and s[5] == ".":
parts = s.split(".")
if len(parts) == 3 and all(p.isdigit() for p in parts):
return f"{parts[2]}-{parts[1]}-{parts[0]}"
return None
candidate = f"{parts[2]}-{parts[1]}-{parts[0]}"
if candidate is None:
return None
# Validate: actually parseable as date (catches '2023-13-45', '32.13.2023', etc)
try:
_dt.date.fromisoformat(candidate)
return candidate
except ValueError:
return None

View file

@ -359,10 +359,10 @@ def enqueue_cadastre_harvest(self: Any, job_id: int) -> dict[str, Any]:
# Fix H: per-quarter freshness skip — не переобрабатывать quarters с свежим
# cad_quarter_stats.fetched_at (последний успешный обход <FRESHNESS_HOURS).
# extra_config.skip_fresh_hours=N управляет порогом; 0 → отключает skip.
# raw_targets.skip_fresh_hours=N управляет порогом; 0 → отключает skip.
# Хранится в существующем `raw_targets` JSONB (no schema change).
# Это сэкономит ~30k NSPD calls при повторных запусках того же scope.
extra_config = job["extra_config"] or {}
skip_fresh_hours = extra_config.get("skip_fresh_hours", 1)
skip_fresh_hours = int(raw_targets.get("skip_fresh_hours", 1) or 0)
skipped_fresh: set[str] = set()
if skip_fresh_hours > 0:
threshold = _dt.datetime.now(_dt.UTC) - _dt.timedelta(hours=skip_fresh_hours)

View file

@ -194,6 +194,18 @@ def test_parse_nspd_date_none_and_invalid() -> None:
assert _parse_nspd_date(123) is None # not str
def test_parse_nspd_date_validates_month_day_ranges() -> None:
"""Invalid month/day (13.03 / 32.13 / etc) → None — не пропускать в SQL."""
from app.services.cadastre.bulk_harvest import _parse_nspd_date
# Невалидный месяц (13) в ISO формате
assert _parse_nspd_date("2023-13-45") is None
# Невалидный день в DD.MM.YYYY
assert _parse_nspd_date("32.13.2023") is None
# Невалидный month
assert _parse_nspd_date("31.02.2023") is None # Feb 31
# ── Grid geometry ────────────────────────────────────────────────────────────
@ -1122,3 +1134,73 @@ async def test_harvest_quarter_geom_heal_failure_does_not_propagate() -> None:
mock_heal.assert_called_once()
# Phase=done — harvest завершился штатно
assert result.phase_state == {"phase": "done", "quarter": "66:41:0303161"}
# ── Fix I: heartbeat в grid-walk loop ────────────────────────────────────────
@pytest.mark.asyncio
async def test_grid_walk_emits_heartbeat_callbacks() -> None:
"""_grid_walk_category должен вызывать update_progress каждые heartbeat_every cells.
225 cells (15×15) / 50 = 4 heartbeats minimum (idx=50, 100, 150, 200).
"""
from app.services.cadastre.bulk_harvest import _grid_walk_category
# Mock db: quarter_bbox_3857 returns valid bbox
db = MagicMock()
db.execute.return_value.mappings.return_value.first.return_value = {
"xmin": 6735845.0,
"ymin": 8329000.0,
"xmax": 6736595.0,
"ymax": 8329750.0,
}
# Mock client: wms_feature_info returns empty list (no upserts to worry about)
client = AsyncMock()
client.wms_feature_info = AsyncMock(return_value=[])
progress_states: list[dict[str, Any]] = []
upserted, requests = await _grid_walk_category(
db=db,
client=client,
quarter="66:41:0303161",
layer_id=36368,
update_progress=lambda s: progress_states.append(s),
heartbeat_every=50,
)
# 225 cells / 50 = 4 heartbeats at idx=50/100/150/200
assert len(progress_states) == 4
assert all("grid_cells_done" in s for s in progress_states)
assert progress_states[0]["grid_cells_done"] == 50
assert progress_states[-1]["grid_cells_done"] == 200
assert progress_states[0]["quarter"] == "66:41:0303161"
# 225 WMS calls (request counter)
assert requests == 225
@pytest.mark.asyncio
async def test_grid_walk_no_heartbeat_when_callback_none() -> None:
"""Если update_progress=None, не должны падать (callback optional)."""
from app.services.cadastre.bulk_harvest import _grid_walk_category
db = MagicMock()
db.execute.return_value.mappings.return_value.first.return_value = {
"xmin": 6735845.0,
"ymin": 8329000.0,
"xmax": 6736595.0,
"ymax": 8329750.0,
}
client = AsyncMock()
client.wms_feature_info = AsyncMock(return_value=[])
# Не передаём update_progress — должно не падать
upserted, requests = await _grid_walk_category(
db=db,
client=client,
quarter="66:41:0303161",
layer_id=36368,
)
assert requests == 225

View file

@ -0,0 +1,132 @@
"""Тесты для backend/app/workers/tasks/scrape_cadastre.py.
Покрывает Fix H per-quarter freshness skip в enqueue_cadastre_harvest.
"""
from __future__ import annotations
import datetime as _dt
from typing import Any
from unittest.mock import MagicMock, patch
def _make_mock_job(quarters: list[str], raw_targets_extra: dict[str, Any] | None = None) -> dict:
"""Mock cadastre_jobs row."""
raw_targets = {"quarters": quarters}
if raw_targets_extra:
raw_targets.update(raw_targets_extra)
return {
"job_id": 99,
"status": "pending",
"raw_targets": raw_targets,
}
def _make_mock_db(job: dict, fresh_quarters: list[str] | None = None) -> MagicMock:
"""Mock DB session.
1. db.execute("SELECT * FROM cadastre_jobs ...").mappings().first() job
2. db.execute("SELECT cad_number FROM cad_quarter_stats ...").scalars().all() fresh
3. db.execute(UPDATE) no-op
"""
db = MagicMock()
fresh_quarters = fresh_quarters or []
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
sql = str(stmt).upper()
result = MagicMock()
if "FROM CADASTRE_JOBS" in sql and "WHERE JOB_ID" in sql and "SELECT *" in str(stmt):
result.mappings.return_value.first.return_value = job
elif "FROM CAD_QUARTER_STATS" in sql:
result.scalars.return_value.all.return_value = fresh_quarters
else:
# UPDATE / other queries
result.scalars.return_value.all.return_value = []
result.mappings.return_value.first.return_value = None
return result
db.execute = execute_router
return db
def test_enqueue_skips_fresh_quarters_with_default_threshold() -> None:
"""Fix H: фреш кварталы (cad_quarter_stats.fetched_at < 1h) пропускаются."""
from app.workers.tasks import scrape_cadastre
quarters = ["66:41:0001001", "66:41:0001002", "66:41:0001003"]
fresh = ["66:41:0001001", "66:41:0001002"] # 2 свежих, 1 stale
job = _make_mock_job(quarters)
db = _make_mock_db(job, fresh_quarters=fresh)
with (
patch.object(scrape_cadastre, "SessionLocal", return_value=db),
patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async") as mock_apply,
):
result = scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
# Только 1 stale quarter должен попасть в apply_async
assert mock_apply.call_count == 1
assert mock_apply.call_args[1]["args"] == ["66:41:0001003", 99]
assert result["enqueued"] == 1
assert result["skipped_fresh"] == 2
def test_enqueue_no_skip_when_skip_fresh_hours_zero() -> None:
"""skip_fresh_hours=0 → отключает freshness check, все quarters enqueued."""
from app.workers.tasks import scrape_cadastre
quarters = ["66:41:0001001", "66:41:0001002"]
job = _make_mock_job(quarters, raw_targets_extra={"skip_fresh_hours": 0})
db = _make_mock_db(job, fresh_quarters=["66:41:0001001"]) # один свежий
with (
patch.object(scrape_cadastre, "SessionLocal", return_value=db),
patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async") as mock_apply,
):
result = scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
# Все 2 quarters должны быть enqueued (skip отключен)
assert mock_apply.call_count == 2
assert result["enqueued"] == 2
assert result["skipped_fresh"] == 0
def test_enqueue_freshness_threshold_passed_in_sql() -> None:
"""skip_fresh_hours=2 → threshold = NOW() - 2h в SQL query."""
from app.workers.tasks import scrape_cadastre
quarters = ["66:41:0001001"]
captured_params: list[dict[str, Any]] = []
job = _make_mock_job(quarters, raw_targets_extra={"skip_fresh_hours": 2})
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
sql = str(stmt).upper()
result = MagicMock()
if "FROM CADASTRE_JOBS" in sql and "SELECT *" in str(stmt):
result.mappings.return_value.first.return_value = job
elif "FROM CAD_QUARTER_STATS" in sql:
if params:
captured_params.append(params)
result.scalars.return_value.all.return_value = []
else:
result.scalars.return_value.all.return_value = []
return result
db = MagicMock()
db.execute = execute_router
with (
patch.object(scrape_cadastre, "SessionLocal", return_value=db),
patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async"),
):
scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
assert len(captured_params) == 1
threshold: _dt.datetime = captured_params[0]["t"]
now = _dt.datetime.now(_dt.UTC)
# Threshold должен быть приблизительно 2h назад (±5s tolerance)
delta = now - threshold
assert _dt.timedelta(hours=1, minutes=59) < delta < _dt.timedelta(hours=2, minutes=1)