diff --git a/backend/app/services/cadastre/bulk_harvest.py b/backend/app/services/cadastre/bulk_harvest.py index 0770bf1a..289d4469 100644 --- a/backend/app/services/cadastre/bulk_harvest.py +++ b/backend/app/services/cadastre/bulk_harvest.py @@ -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 diff --git a/backend/app/workers/tasks/scrape_cadastre.py b/backend/app/workers/tasks/scrape_cadastre.py index 0cf57026..4c1ec35d 100644 --- a/backend/app/workers/tasks/scrape_cadastre.py +++ b/backend/app/workers/tasks/scrape_cadastre.py @@ -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 (последний успешный обход 0: threshold = _dt.datetime.now(_dt.UTC) - _dt.timedelta(hours=skip_fresh_hours) diff --git a/backend/tests/services/test_cadastre_bulk.py b/backend/tests/services/test_cadastre_bulk.py index 8369f30d..1252ccc2 100644 --- a/backend/tests/services/test_cadastre_bulk.py +++ b/backend/tests/services/test_cadastre_bulk.py @@ -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 diff --git a/backend/tests/workers/test_scrape_cadastre.py b/backend/tests/workers/test_scrape_cadastre.py new file mode 100644 index 00000000..fbcd6e1f --- /dev/null +++ b/backend/tests/workers/test_scrape_cadastre.py @@ -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)