feat(tradein): schedule yandex address backfill task (#855) #866
4 changed files with 427 additions and 0 deletions
|
|
@ -19,6 +19,11 @@ Sources:
|
||||||
(tasks/asking_to_sold_ratio.py, #648 Stage 4; pure internal DB
|
(tasks/asking_to_sold_ratio.py, #648 Stage 4; pure internal DB
|
||||||
re-derivation of the asking→sold ratios — no external calls,
|
re-derivation of the asking→sold ratios — no external calls,
|
||||||
enabled by default; window after rosreestr_dkp_import)
|
enabled by default; window after rosreestr_dkp_import)
|
||||||
|
- yandex_address_backfill → run_yandex_address_backfill
|
||||||
|
(tasks/yandex_address_backfill.py, #855; EKB-pilot: fetches
|
||||||
|
yandex detail pages, extracts full address from <title>,
|
||||||
|
updates listings.address for street-only addresses;
|
||||||
|
nightly 02:00-03:00 UTC, before refresh_search_matview)
|
||||||
- deactivate_stale_avito → deactivate_stale_avito_listings
|
- deactivate_stale_avito → deactivate_stale_avito_listings
|
||||||
(tasks/deactivate_stale_avito.py, #759; marks avito listings
|
(tasks/deactivate_stale_avito.py, #759; marks avito listings
|
||||||
with last_seen_at > TTL days as is_active=false — no DELETE,
|
with last_seen_at > TTL days as is_active=false — no DELETE,
|
||||||
|
|
@ -412,6 +417,44 @@ async def _execute_cian_backfill(
|
||||||
raise
|
raise
|
||||||
|
|
||||||
|
|
||||||
|
async def trigger_yandex_address_backfill_run(
|
||||||
|
db: Session, schedule_row: dict[str, Any]
|
||||||
|
) -> int | None:
|
||||||
|
"""Создать scrape_runs + launch run_yandex_address_backfill в asyncio.create_task.
|
||||||
|
|
||||||
|
EKB-only pilot (#855): backfills listings.address for yandex listings that have
|
||||||
|
street-only addresses (no house number). Fetches detail pages via curl_cffi chrome120,
|
||||||
|
extracts full address from <title>, updates listings WHERE address differs.
|
||||||
|
|
||||||
|
Mirrors trigger_cian_backfill_run: claim run → create_task → mark_done/failed.
|
||||||
|
executor logic lives in tasks/yandex_address_backfill.py which delegates to
|
||||||
|
services/yandex_address_backfill.py (backfill_yandex_addresses).
|
||||||
|
|
||||||
|
Returns run_id или None (skip — already running).
|
||||||
|
"""
|
||||||
|
run_id = _claim_run(db, schedule_row)
|
||||||
|
if run_id is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
params = schedule_row.get("default_params") or {}
|
||||||
|
|
||||||
|
async def _run() -> None:
|
||||||
|
run_db = SessionLocal()
|
||||||
|
try:
|
||||||
|
from app.tasks.yandex_address_backfill import run_yandex_address_backfill
|
||||||
|
|
||||||
|
await run_yandex_address_backfill(run_db, run_id=run_id, params=params)
|
||||||
|
except Exception:
|
||||||
|
logger.exception("scheduler: run_yandex_address_backfill crashed run_id=%d", run_id)
|
||||||
|
finally:
|
||||||
|
run_db.close()
|
||||||
|
|
||||||
|
task = asyncio.create_task(_run())
|
||||||
|
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||||||
|
logger.info("scheduler: triggered yandex_address_backfill run_id=%d", run_id)
|
||||||
|
return run_id
|
||||||
|
|
||||||
|
|
||||||
async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||||||
"""Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task)."""
|
"""Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task)."""
|
||||||
run_id = _claim_run(db, schedule_row)
|
run_id = _claim_run(db, schedule_row)
|
||||||
|
|
@ -868,6 +911,8 @@ async def scheduler_loop() -> None:
|
||||||
await trigger_asking_to_sold_ratio_run(db, sch)
|
await trigger_asking_to_sold_ratio_run(db, sch)
|
||||||
elif source == "refresh_search_matview":
|
elif source == "refresh_search_matview":
|
||||||
await trigger_refresh_search_matview_run(db, sch)
|
await trigger_refresh_search_matview_run(db, sch)
|
||||||
|
elif source == "yandex_address_backfill":
|
||||||
|
await trigger_yandex_address_backfill_run(db, sch)
|
||||||
elif source == "deactivate_stale_avito":
|
elif source == "deactivate_stale_avito":
|
||||||
await trigger_deactivate_stale_avito_run(db, sch)
|
await trigger_deactivate_stale_avito_run(db, sch)
|
||||||
else:
|
else:
|
||||||
|
|
|
||||||
116
tradein-mvp/backend/app/tasks/yandex_address_backfill.py
Normal file
116
tradein-mvp/backend/app/tasks/yandex_address_backfill.py
Normal file
|
|
@ -0,0 +1,116 @@
|
||||||
|
"""Scheduled task: backfill listings.address for EKB yandex listings with street-only addresses.
|
||||||
|
|
||||||
|
Pilot scope: Екатеринбург only (region_code=66, city hardcoded).
|
||||||
|
Yandex SERP cards include only the street name (e.g. «улица Горького»); the offer detail
|
||||||
|
page <title> contains the full address including house number:
|
||||||
|
«Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321 на Яндекс.Недвижимости»
|
||||||
|
|
||||||
|
This task selects active listings that match the predicate (source='yandex', region_code=66,
|
||||||
|
address IS NOT NULL, address has no house number, source_url IS NOT NULL), fetches each
|
||||||
|
detail page via curl_cffi(chrome120), extracts the full address from the <title>, and
|
||||||
|
UPDATE listings.address ONLY when the extracted value differs from the current one.
|
||||||
|
|
||||||
|
Wired into the in-app scheduler as source='yandex_address_backfill'. Triggered nightly via
|
||||||
|
the scrape_schedules seed (migration 091). Manual trigger also available via POST
|
||||||
|
/api/v1/admin/scrape/yandex-address-backfill.
|
||||||
|
|
||||||
|
Delegates actual HTTP + DB work to app.services.yandex_address_backfill.backfill_yandex_addresses
|
||||||
|
(the proven service module that already implements RE_TITLE_ADDRESS, _extract_address_from_title,
|
||||||
|
and _eligible_listings).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from app.services import scrape_runs as runs_mod
|
||||||
|
from app.services.yandex_address_backfill import (
|
||||||
|
YandexAddressBackfillResult,
|
||||||
|
backfill_yandex_addresses,
|
||||||
|
)
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"YandexAddressBackfillTaskResult",
|
||||||
|
"run_yandex_address_backfill",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class YandexAddressBackfillTaskResult:
|
||||||
|
"""Result counters from a scheduled backfill run."""
|
||||||
|
|
||||||
|
checked: int = 0
|
||||||
|
saved: int = 0
|
||||||
|
skipped: int = 0
|
||||||
|
errors: int = 0
|
||||||
|
duration_sec: float = field(default=0.0)
|
||||||
|
|
||||||
|
|
||||||
|
async def run_yandex_address_backfill(
|
||||||
|
db: Session,
|
||||||
|
*,
|
||||||
|
run_id: int,
|
||||||
|
params: dict,
|
||||||
|
) -> YandexAddressBackfillTaskResult:
|
||||||
|
"""Execute Yandex address backfill with run lifecycle management.
|
||||||
|
|
||||||
|
Wraps backfill_yandex_addresses(), emitting heartbeat before the batch call
|
||||||
|
and marking the scrape_run as done/failed on completion.
|
||||||
|
|
||||||
|
Params (from default_params jsonb in scrape_schedules):
|
||||||
|
batch_size: int — max listings to process (default 200).
|
||||||
|
request_delay_sec: float — seconds between requests (default 3.0).
|
||||||
|
"""
|
||||||
|
batch_size = int(params.get("batch_size", 200))
|
||||||
|
request_delay_sec = float(params.get("request_delay_sec", 3.0))
|
||||||
|
|
||||||
|
counters: dict[str, int] = {
|
||||||
|
"checked": 0,
|
||||||
|
"saved": 0,
|
||||||
|
"skipped": 0,
|
||||||
|
"errors": 0,
|
||||||
|
}
|
||||||
|
|
||||||
|
try:
|
||||||
|
runs_mod.update_heartbeat(db, run_id, counters)
|
||||||
|
|
||||||
|
result: YandexAddressBackfillResult = await backfill_yandex_addresses(
|
||||||
|
db,
|
||||||
|
limit=batch_size,
|
||||||
|
request_delay_sec=request_delay_sec,
|
||||||
|
)
|
||||||
|
|
||||||
|
counters = {
|
||||||
|
"checked": result.checked,
|
||||||
|
"saved": result.saved,
|
||||||
|
"skipped": result.skipped,
|
||||||
|
"errors": result.errors,
|
||||||
|
"duration_sec": int(result.duration_sec),
|
||||||
|
}
|
||||||
|
runs_mod.mark_done(db, run_id, counters)
|
||||||
|
logger.info(
|
||||||
|
"scheduler: yandex_address_backfill run_id=%d done — "
|
||||||
|
"checked=%d saved=%d skipped=%d errors=%d %.1fs",
|
||||||
|
run_id,
|
||||||
|
result.checked,
|
||||||
|
result.saved,
|
||||||
|
result.skipped,
|
||||||
|
result.errors,
|
||||||
|
result.duration_sec,
|
||||||
|
)
|
||||||
|
return YandexAddressBackfillTaskResult(
|
||||||
|
checked=result.checked,
|
||||||
|
saved=result.saved,
|
||||||
|
skipped=result.skipped,
|
||||||
|
errors=result.errors,
|
||||||
|
duration_sec=result.duration_sec,
|
||||||
|
)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.exception("scheduler: yandex_address_backfill run_id=%d failed", run_id)
|
||||||
|
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
|
||||||
|
raise
|
||||||
|
|
@ -0,0 +1,71 @@
|
||||||
|
-- 091_scrape_schedules_seed_yandex_address_backfill.sql
|
||||||
|
-- Issue #855 — seed scrape_schedules row for nightly yandex address backfill.
|
||||||
|
--
|
||||||
|
-- Context:
|
||||||
|
-- Active EKB yandex listings often have street-only addresses (e.g. «улица Горького»)
|
||||||
|
-- because Yandex SERP cards omit the house number. The detail page <title> contains
|
||||||
|
-- the full address including house number:
|
||||||
|
-- «Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321».
|
||||||
|
-- Without house numbers, geocoding falls back to street-centroid coordinates and
|
||||||
|
-- valuation analogs have lower match quality.
|
||||||
|
--
|
||||||
|
-- Fix:
|
||||||
|
-- 1. services/yandex_address_backfill.py — backfill_yandex_addresses() service function
|
||||||
|
-- (curl_cffi chrome120, RE_TITLE_ADDRESS regex over <title>, UPDATE only-if-changed).
|
||||||
|
-- 2. tasks/yandex_address_backfill.py — run_yandex_address_backfill() task wrapper
|
||||||
|
-- (run lifecycle: heartbeat + mark_done/mark_failed).
|
||||||
|
-- 3. scheduler.py — trigger_yandex_address_backfill_run() + dispatch branch
|
||||||
|
-- `elif source == "yandex_address_backfill"` in scheduler_loop().
|
||||||
|
-- 4. This migration — seeds the scrape_schedules row (enabled=true, 02:00-03:00 UTC).
|
||||||
|
--
|
||||||
|
-- Schedule window 02:00-03:00 UTC:
|
||||||
|
-- Nightly, before refresh_search_matview (03:00-04:00 UTC). Runs after the yandex
|
||||||
|
-- city sweep (dormant/manual window 01:00-02:00 UTC) so backfill sees the freshest
|
||||||
|
-- is_active=TRUE rows. Expected duration: ~30 min for 200 listings at 3s delay.
|
||||||
|
--
|
||||||
|
-- next_run_at bootstrapped to tomorrow 02:00 UTC so the scheduler does not fire
|
||||||
|
-- immediately on deploy (same pattern as 078/079/082/088).
|
||||||
|
--
|
||||||
|
-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
|
||||||
|
--
|
||||||
|
-- Dependencies:
|
||||||
|
-- 052_scrape_schedules.sql (table + UNIQUE(source)).
|
||||||
|
-- scheduler.py change (trigger_yandex_address_backfill_run must be deployed).
|
||||||
|
-- services/yandex_address_backfill.py (backfill_yandex_addresses must be deployed).
|
||||||
|
--
|
||||||
|
-- Deploy order:
|
||||||
|
-- Apply this migration after deploying the scheduler.py + task changes so the
|
||||||
|
-- scheduler can dispatch the new source name correctly on first fire.
|
||||||
|
--
|
||||||
|
-- EKB pilot scope: only listings with region_code=66 are processed. Expanding to
|
||||||
|
-- other cities requires parametrizing the city/region_code in backfill_yandex_addresses.
|
||||||
|
|
||||||
|
BEGIN;
|
||||||
|
|
||||||
|
INSERT INTO scrape_schedules (
|
||||||
|
source,
|
||||||
|
enabled,
|
||||||
|
window_start_hour,
|
||||||
|
window_end_hour,
|
||||||
|
next_run_at,
|
||||||
|
default_params
|
||||||
|
)
|
||||||
|
VALUES
|
||||||
|
(
|
||||||
|
'yandex_address_backfill',
|
||||||
|
true, -- EKB pilot; HTTP calls (curl_cffi chrome120)
|
||||||
|
2, -- window 02:00-03:00 UTC
|
||||||
|
3,
|
||||||
|
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC',
|
||||||
|
'{"batch_size": 200, "request_delay_sec": 3.0}'::jsonb
|
||||||
|
)
|
||||||
|
ON CONFLICT (source) DO NOTHING;
|
||||||
|
|
||||||
|
COMMENT ON TABLE scrape_schedules IS
|
||||||
|
'In-app scheduler config (заменяет cron-script setup). '
|
||||||
|
'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), '
|
||||||
|
'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), '
|
||||||
|
'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), '
|
||||||
|
'yandex_address_backfill (#855, EKB pilot).';
|
||||||
|
|
||||||
|
COMMIT;
|
||||||
195
tradein-mvp/backend/tests/tasks/test_yandex_address_backfill.py
Normal file
195
tradein-mvp/backend/tests/tasks/test_yandex_address_backfill.py
Normal file
|
|
@ -0,0 +1,195 @@
|
||||||
|
"""Tests for yandex_address_backfill — pure regex extraction, no network/DB."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||||||
|
|
||||||
|
# DATABASE_URL required by config before any app import.
|
||||||
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||||
|
|
||||||
|
# WeasyPrint stub — not installed in CI without GTK.
|
||||||
|
_wp_mock = MagicMock()
|
||||||
|
sys.modules.setdefault("weasyprint", _wp_mock)
|
||||||
|
|
||||||
|
import pytest # noqa: E402
|
||||||
|
|
||||||
|
from app.services.yandex_address_backfill import ( # noqa: E402
|
||||||
|
YandexAddressBackfillResult,
|
||||||
|
_extract_address_from_title,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Regex used to assert that an extracted address contains a house number.
|
||||||
|
_RE_HAS_HOUSE_NUMBER = re.compile(r",\s*\d+")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Pure regex / extraction tests (no network, no DB)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_extract_address_standard_title() -> None:
|
||||||
|
"""Standard title: «Продажа квартиры — Екатеринбург, улица Горького, 36 — id 7654321»."""
|
||||||
|
html = (
|
||||||
|
"<title>Продажа квартиры — Екатеринбург, улица Горького, 36 "
|
||||||
|
"— id 7654321 на Яндекс.Недвижимости</title>"
|
||||||
|
)
|
||||||
|
addr = _extract_address_from_title(html)
|
||||||
|
assert addr is not None, "Should extract address from standard title"
|
||||||
|
assert _RE_HAS_HOUSE_NUMBER.search(
|
||||||
|
addr
|
||||||
|
), f"Extracted address should contain house number: {addr!r}"
|
||||||
|
assert "Екатеринбург" in addr
|
||||||
|
assert "Горького" in addr
|
||||||
|
assert "36" in addr
|
||||||
|
|
||||||
|
|
||||||
|
def test_extract_address_with_zhk_prefix() -> None:
|
||||||
|
"""Title with optional ЖК prefix before city: «— ЖК Каменный Ручей, Екатеринбург, …»."""
|
||||||
|
html = (
|
||||||
|
"<title>Продажа квартиры — ЖК Каменный Ручей, "
|
||||||
|
"Екатеринбург, проспект Космонавтов, 110А — id 99001 </title>"
|
||||||
|
)
|
||||||
|
addr = _extract_address_from_title(html)
|
||||||
|
assert addr is not None, "Should extract address when ЖК prefix is present"
|
||||||
|
assert _RE_HAS_HOUSE_NUMBER.search(
|
||||||
|
addr
|
||||||
|
), f"Extracted address should contain house number: {addr!r}"
|
||||||
|
assert "Екатеринбург" in addr
|
||||||
|
|
||||||
|
|
||||||
|
def test_extract_address_nbsp_replaced() -> None:
|
||||||
|
"""\\xa0 (non-breaking space) in title should be normalised before regex match."""
|
||||||
|
html = (
|
||||||
|
"<title>Продажа\xa0квартиры\xa0—\xa0Екатеринбург,\xa0улица\xa0Ленина,\xa012"
|
||||||
|
"\xa0—\xa0id\xa0123456</title>"
|
||||||
|
)
|
||||||
|
addr = _extract_address_from_title(html)
|
||||||
|
assert addr is not None, "\\xa0 should be replaced before regex"
|
||||||
|
assert _RE_HAS_HOUSE_NUMBER.search(
|
||||||
|
addr
|
||||||
|
), f"Extracted address should contain house number: {addr!r}"
|
||||||
|
|
||||||
|
|
||||||
|
def test_extract_address_no_title_returns_none() -> None:
|
||||||
|
"""HTML without a matching <title> returns None."""
|
||||||
|
html = "<html><body>Объявление снято</body></html>"
|
||||||
|
addr = _extract_address_from_title(html)
|
||||||
|
assert addr is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_extract_address_title_without_id_returns_none() -> None:
|
||||||
|
"""Title that matches city but has no '— id NNN' sentinel returns None."""
|
||||||
|
html = "<title>Екатеринбург, улица Горького, 36</title>"
|
||||||
|
addr = _extract_address_from_title(html)
|
||||||
|
assert addr is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_extract_address_street_only_still_extracted() -> None:
|
||||||
|
"""Title with street-only (no house number) is extracted but fails the house-number check.
|
||||||
|
|
||||||
|
The extractor itself does not gate on house number presence — the caller
|
||||||
|
(backfill_yandex_addresses) checks _RE_HAS_HOUSE_NUMBER separately and skips.
|
||||||
|
"""
|
||||||
|
html = "<title>Продажа квартиры — Екатеринбург, улица Горького — id 7654321</title>"
|
||||||
|
addr = _extract_address_from_title(html)
|
||||||
|
# May or may not match depending on whether «улица Горького» satisfies the regex —
|
||||||
|
# the key assertion is that IF it does, it does NOT contain a house number.
|
||||||
|
if addr is not None:
|
||||||
|
assert not _RE_HAS_HOUSE_NUMBER.search(
|
||||||
|
addr
|
||||||
|
), "Street-only extracted address should NOT contain a house number"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Task-level smoke tests (no real DB, no real HTTP)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_run_yandex_address_backfill_no_eligible_listings() -> None:
|
||||||
|
"""No eligible listings → run marked done with zeroed counters, no HTTP calls."""
|
||||||
|
from app.tasks.yandex_address_backfill import run_yandex_address_backfill
|
||||||
|
|
||||||
|
db = MagicMock()
|
||||||
|
|
||||||
|
empty_result = YandexAddressBackfillResult(
|
||||||
|
checked=0, saved=0, skipped=0, errors=0, duration_sec=0.01
|
||||||
|
)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch(
|
||||||
|
"app.tasks.yandex_address_backfill.backfill_yandex_addresses",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value=empty_result,
|
||||||
|
) as mock_backfill,
|
||||||
|
patch("app.tasks.yandex_address_backfill.runs_mod") as mock_runs,
|
||||||
|
):
|
||||||
|
result = await run_yandex_address_backfill(db, run_id=42, params={})
|
||||||
|
|
||||||
|
mock_backfill.assert_awaited_once()
|
||||||
|
mock_runs.update_heartbeat.assert_called_once_with(
|
||||||
|
db, 42, {"checked": 0, "saved": 0, "skipped": 0, "errors": 0}
|
||||||
|
)
|
||||||
|
mock_runs.mark_done.assert_called_once()
|
||||||
|
mock_runs.mark_failed.assert_not_called()
|
||||||
|
|
||||||
|
assert result.checked == 0
|
||||||
|
assert result.saved == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_run_yandex_address_backfill_saves_enriched_count() -> None:
|
||||||
|
"""5 checked, 3 saved → result reflects correct counters."""
|
||||||
|
from app.tasks.yandex_address_backfill import run_yandex_address_backfill
|
||||||
|
|
||||||
|
db = MagicMock()
|
||||||
|
|
||||||
|
backfill_result = YandexAddressBackfillResult(
|
||||||
|
checked=5, saved=3, skipped=1, errors=1, duration_sec=15.0
|
||||||
|
)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch(
|
||||||
|
"app.tasks.yandex_address_backfill.backfill_yandex_addresses",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value=backfill_result,
|
||||||
|
),
|
||||||
|
patch("app.tasks.yandex_address_backfill.runs_mod") as mock_runs,
|
||||||
|
):
|
||||||
|
result = await run_yandex_address_backfill(
|
||||||
|
db, run_id=99, params={"batch_size": 100, "request_delay_sec": 2.0}
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_runs.mark_done.assert_called_once()
|
||||||
|
done_counters = mock_runs.mark_done.call_args[0][2] # positional: (db, run_id, counters)
|
||||||
|
assert done_counters["checked"] == 5
|
||||||
|
assert done_counters["saved"] == 3
|
||||||
|
assert done_counters["errors"] == 1
|
||||||
|
|
||||||
|
assert result.saved == 3
|
||||||
|
assert result.errors == 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_run_yandex_address_backfill_marks_failed_on_exception() -> None:
|
||||||
|
"""backfill_yandex_addresses raises → mark_failed called, exception re-raised."""
|
||||||
|
from app.tasks.yandex_address_backfill import run_yandex_address_backfill
|
||||||
|
|
||||||
|
db = MagicMock()
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch(
|
||||||
|
"app.tasks.yandex_address_backfill.backfill_yandex_addresses",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
side_effect=RuntimeError("network timeout"),
|
||||||
|
),
|
||||||
|
patch("app.tasks.yandex_address_backfill.runs_mod") as mock_runs,
|
||||||
|
):
|
||||||
|
with pytest.raises(RuntimeError, match="network timeout"):
|
||||||
|
await run_yandex_address_backfill(db, run_id=7, params={})
|
||||||
|
|
||||||
|
mock_runs.mark_failed.assert_called_once()
|
||||||
|
mock_runs.mark_done.assert_not_called()
|
||||||
Loading…
Add table
Reference in a new issue