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:
parent
726b15ad55
commit
f526a43ef9
4 changed files with 233 additions and 8 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
132
backend/tests/workers/test_scrape_cadastre.py
Normal file
132
backend/tests/workers/test_scrape_cadastre.py
Normal 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)
|
||||
Loading…
Add table
Reference in a new issue