feat(tradein): nightly task to deactivate stale avito listings (#759) #862
5 changed files with 437 additions and 0 deletions
|
|
@ -182,6 +182,13 @@ class Settings(BaseSettings):
|
||||||
# Сколько раз сменить IP при блоке прежде чем сдаться (на одну страницу).
|
# Сколько раз сменить IP при блоке прежде чем сдаться (на одну страницу).
|
||||||
avito_proxy_max_rotations: int = 2
|
avito_proxy_max_rotations: int = 2
|
||||||
|
|
||||||
|
# ── #759: deactivate stale avito listings ───────────────────────────────
|
||||||
|
# Avito-объявления, не виденные scraper'ом более avito_stale_ttl_days дней,
|
||||||
|
# помечаются is_active=false ночной задачей deactivate_stale_avito_listings.
|
||||||
|
# TTL=10 — баланс между "снятое объявление" (обычно 3-7 дней молчания) и
|
||||||
|
# допуском на перерыв в работе scraper'а. ENV: AVITO_STALE_TTL_DAYS.
|
||||||
|
avito_stale_ttl_days: int = 10
|
||||||
|
|
||||||
# ── Yandex SERP cookies (#801/T4) ───────────────────────────────────────
|
# ── Yandex SERP cookies (#801/T4) ───────────────────────────────────────
|
||||||
# Путь к JSON-файлу с cookies браузера (формат: [{name, value, ...}, ...]).
|
# Путь к JSON-файлу с cookies браузера (формат: [{name, value, ...}, ...]).
|
||||||
# Если задан и файл существует — cookies передаются в curl_cffi-сессию при
|
# Если задан и файл существует — cookies передаются в curl_cffi-сессию при
|
||||||
|
|
|
||||||
|
|
@ -19,6 +19,10 @@ 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)
|
||||||
|
- deactivate_stale_avito → deactivate_stale_avito_listings
|
||||||
|
(tasks/deactivate_stale_avito.py, #759; marks avito listings
|
||||||
|
with last_seen_at > TTL days as is_active=false — no DELETE,
|
||||||
|
no external calls, enabled by default; window after avito sweep)
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
@ -513,6 +517,41 @@ async def trigger_refresh_search_matview_run(
|
||||||
return run_id
|
return run_id
|
||||||
|
|
||||||
|
|
||||||
|
async def trigger_deactivate_stale_avito_run(
|
||||||
|
db: Session, schedule_row: dict[str, Any]
|
||||||
|
) -> int | None:
|
||||||
|
"""Создать scrape_runs + launch deactivate_stale_avito_listings в executor (sync DB-only task).
|
||||||
|
|
||||||
|
Помечает avito-объявления, не виденные >TTL дней, как is_active=false (#759).
|
||||||
|
Задача синхронная (чистый internal UPDATE, БЕЗ внешних HTTP-вызовов) — гоняем в
|
||||||
|
run_in_executor по образцу trigger_listing_source_snapshot_run. SAFE to enable —
|
||||||
|
schedule seed enabled=true (090), pure internal DB op.
|
||||||
|
Окно 06:00-07:00 UTC — после avito_city_sweep (02:00-05:00 UTC).
|
||||||
|
|
||||||
|
Returns run_id (или None если skip — есть running run).
|
||||||
|
"""
|
||||||
|
run_id = _claim_run(db, schedule_row)
|
||||||
|
if run_id is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
async def _run() -> None:
|
||||||
|
run_db = SessionLocal()
|
||||||
|
try:
|
||||||
|
from app.tasks.deactivate_stale_avito import deactivate_stale_avito_listings
|
||||||
|
|
||||||
|
loop = asyncio.get_event_loop()
|
||||||
|
await loop.run_in_executor(None, deactivate_stale_avito_listings, run_db, run_id)
|
||||||
|
except Exception:
|
||||||
|
logger.exception("scheduler: deactivate_stale_avito 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 deactivate_stale_avito run_id=%d", run_id)
|
||||||
|
return run_id
|
||||||
|
|
||||||
|
|
||||||
async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||||||
"""Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).
|
"""Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).
|
||||||
|
|
||||||
|
|
@ -829,6 +868,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 == "deactivate_stale_avito":
|
||||||
|
await trigger_deactivate_stale_avito_run(db, sch)
|
||||||
else:
|
else:
|
||||||
logger.warning("scheduler: unknown source=%s, skip", source)
|
logger.warning("scheduler: unknown source=%s, skip", source)
|
||||||
finally:
|
finally:
|
||||||
|
|
|
||||||
70
tradein-mvp/backend/app/tasks/deactivate_stale_avito.py
Normal file
70
tradein-mvp/backend/app/tasks/deactivate_stale_avito.py
Normal file
|
|
@ -0,0 +1,70 @@
|
||||||
|
"""Deactivate stale avito listings that have not been seen in >TTL days (#759).
|
||||||
|
|
||||||
|
Нечасто обновляемые объявления на Avito (last_seen_at < NOW() - TTL) помечаются
|
||||||
|
is_active=false, чтобы coverage/estimator denominators не раздувались мёртвыми
|
||||||
|
объявлениями. Строки НЕ удаляются — история нужна для бэктеста (#667).
|
||||||
|
|
||||||
|
Только source='avito'. Cian/Yandex не затрагиваются.
|
||||||
|
|
||||||
|
Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app
|
||||||
|
scheduler'ом через trigger_deactivate_stale_avito_run() (scheduler.py),
|
||||||
|
по образцу snapshot_listing_sources / recompute_asking_to_sold_ratios
|
||||||
|
(sync task в run_in_executor).
|
||||||
|
|
||||||
|
TTL берётся из settings.avito_stale_ttl_days (env AVITO_STALE_TTL_DAYS, default 10).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from sqlalchemy import text
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from app.core.config import settings
|
||||||
|
from app.services import scrape_runs as runs_mod
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# UPDATE: помечаем is_active=false для avito-объявлений, не виденных >TTL дней.
|
||||||
|
# CAST(:ttl_days || ' days' AS interval) — psycopg v3 safe (никаких :param::type).
|
||||||
|
# Только source='avito' и is_active=true — чтобы не трогать уже неактивные строки
|
||||||
|
# и не затрагивать cian/yandex.
|
||||||
|
_DEACTIVATE_SQL = text(
|
||||||
|
"""
|
||||||
|
UPDATE listings
|
||||||
|
SET is_active = false
|
||||||
|
WHERE source = 'avito'
|
||||||
|
AND is_active = true
|
||||||
|
AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]:
|
||||||
|
"""Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад.
|
||||||
|
|
||||||
|
Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
|
||||||
|
Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed).
|
||||||
|
|
||||||
|
Returns {"deactivated": N} — количество обновлённых строк.
|
||||||
|
"""
|
||||||
|
counters: dict[str, int] = {"deactivated": 0}
|
||||||
|
try:
|
||||||
|
result = db.execute(_DEACTIVATE_SQL, {"ttl_days": settings.avito_stale_ttl_days})
|
||||||
|
counters["deactivated"] = result.rowcount or 0
|
||||||
|
|
||||||
|
db.commit()
|
||||||
|
runs_mod.mark_done(db, run_id, counters)
|
||||||
|
logger.info(
|
||||||
|
"deactivate_stale_avito run_id=%d done: deactivated=%d (ttl_days=%d)",
|
||||||
|
run_id,
|
||||||
|
counters["deactivated"],
|
||||||
|
settings.avito_stale_ttl_days,
|
||||||
|
)
|
||||||
|
return counters
|
||||||
|
except Exception as exc:
|
||||||
|
logger.exception("deactivate_stale_avito run_id=%d failed", run_id)
|
||||||
|
db.rollback()
|
||||||
|
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
|
||||||
|
raise
|
||||||
|
|
@ -0,0 +1,65 @@
|
||||||
|
-- 090_scrape_schedules_seed_deactivate_stale_avito.sql
|
||||||
|
-- Issue #759 — seed scrape_schedules row for nightly deactivation of stale avito listings.
|
||||||
|
--
|
||||||
|
-- Context:
|
||||||
|
-- Avito listings that haven't been seen by the scraper for >AVITO_STALE_TTL_DAYS days
|
||||||
|
-- (default 10) remain is_active=true indefinitely. This inflates the active-listing
|
||||||
|
-- denominator used by the coverage and estimator services, degrading accuracy.
|
||||||
|
--
|
||||||
|
-- Fix:
|
||||||
|
-- 1. app/tasks/deactivate_stale_avito.py — deactivate_stale_avito_listings()
|
||||||
|
-- issues a single UPDATE listings SET is_active=false WHERE source='avito'
|
||||||
|
-- AND is_active=true AND last_seen_at < NOW()-INTERVAL(:ttl_days days).
|
||||||
|
-- Rows are NOT deleted — history is required for backtest (#667).
|
||||||
|
-- 2. scheduler.py — added trigger_deactivate_stale_avito_run() + dispatch branch
|
||||||
|
-- `elif source == "deactivate_stale_avito"` in scheduler_loop().
|
||||||
|
-- 3. app/core/config.py — AVITO_STALE_TTL_DAYS setting (default 10).
|
||||||
|
-- 4. This migration — seeds the scrape_schedules row.
|
||||||
|
--
|
||||||
|
-- Schedule window 06:00-07:00 UTC:
|
||||||
|
-- After the avito_city_sweep window (02:00-05:00 UTC per #814). Running after the
|
||||||
|
-- sweep ensures listings genuinely seen in the nightly pass are NOT deactivated.
|
||||||
|
--
|
||||||
|
-- next_run_at bootstrapped to tomorrow 06:00 UTC so the scheduler does not fire
|
||||||
|
-- immediately on deploy (same pattern as 079/082/088).
|
||||||
|
--
|
||||||
|
-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
|
||||||
|
--
|
||||||
|
-- Dependencies:
|
||||||
|
-- 052_scrape_schedules.sql (table + UNIQUE(source)).
|
||||||
|
-- listings table with last_seen_at column (002_core_tables.sql).
|
||||||
|
-- scheduler.py change (trigger_deactivate_stale_avito_run 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.
|
||||||
|
|
||||||
|
BEGIN;
|
||||||
|
|
||||||
|
INSERT INTO scrape_schedules (
|
||||||
|
source,
|
||||||
|
enabled,
|
||||||
|
window_start_hour,
|
||||||
|
window_end_hour,
|
||||||
|
next_run_at,
|
||||||
|
default_params
|
||||||
|
)
|
||||||
|
VALUES
|
||||||
|
(
|
||||||
|
'deactivate_stale_avito',
|
||||||
|
true, -- SAFE: pure internal DB UPDATE, no ext calls
|
||||||
|
6,
|
||||||
|
7,
|
||||||
|
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 6)) AT TIME ZONE 'UTC',
|
||||||
|
'{}'::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), '
|
||||||
|
'deactivate_stale_avito (#759).';
|
||||||
|
|
||||||
|
COMMIT;
|
||||||
254
tradein-mvp/backend/tests/test_deactivate_stale_avito.py
Normal file
254
tradein-mvp/backend/tests/test_deactivate_stale_avito.py
Normal file
|
|
@ -0,0 +1,254 @@
|
||||||
|
"""Tests for the stale avito listing deactivation task (#759).
|
||||||
|
|
||||||
|
Static assertions (SQL shape, psycopg-v3 cast discipline, scheduler wiring) mirror
|
||||||
|
the pattern established in test_listing_source_snapshot.py. One behavioural test drives
|
||||||
|
the counter logic via a fake db without a live database.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import inspect
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
# Stub DATABASE_URL before any app import (same guard as test_scheduler.py /
|
||||||
|
# test_listing_source_snapshot.py — these tests are static/fake-db, no live DB).
|
||||||
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||||
|
|
||||||
|
from app.services import scheduler
|
||||||
|
from app.tasks import deactivate_stale_avito as task_mod
|
||||||
|
|
||||||
|
_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
|
||||||
|
_MIGRATION_090 = _SQL_DIR / "090_scrape_schedules_seed_deactivate_stale_avito.sql"
|
||||||
|
|
||||||
|
_DEACTIVATE_SQL = str(task_mod._DEACTIVATE_SQL.text)
|
||||||
|
_TASK_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings)
|
||||||
|
|
||||||
|
|
||||||
|
# ── SQL shape ─────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_sql_targets_only_avito_source() -> None:
|
||||||
|
"""UPDATE must filter source='avito' — cian/yandex rows must not be touched."""
|
||||||
|
assert "source = 'avito'" in _DEACTIVATE_SQL
|
||||||
|
|
||||||
|
|
||||||
|
def test_sql_only_deactivates_not_deletes() -> None:
|
||||||
|
"""Rows must be marked is_active=false, never DELETEd (history needed for #667)."""
|
||||||
|
assert "SET is_active = false" in _DEACTIVATE_SQL
|
||||||
|
assert "DELETE" not in _DEACTIVATE_SQL.upper()
|
||||||
|
|
||||||
|
|
||||||
|
def test_sql_guards_already_inactive_rows() -> None:
|
||||||
|
"""Filter AND is_active=true avoids redundant updates on rows already deactivated."""
|
||||||
|
assert "is_active = true" in _DEACTIVATE_SQL
|
||||||
|
|
||||||
|
|
||||||
|
def test_sql_uses_last_seen_at_threshold() -> None:
|
||||||
|
"""Threshold is last_seen_at < NOW() - interval, not any other column."""
|
||||||
|
assert "last_seen_at" in _DEACTIVATE_SQL
|
||||||
|
assert "NOW()" in _DEACTIVATE_SQL
|
||||||
|
|
||||||
|
|
||||||
|
def test_sql_uses_cast_interval_not_double_colon() -> None:
|
||||||
|
"""psycopg v3: bind param via CAST(:ttl_days || ' days' AS interval), never ::interval."""
|
||||||
|
assert "CAST(:ttl_days || ' days' AS interval)" in _DEACTIVATE_SQL
|
||||||
|
# No bind-param (:name) followed by ::type anywhere.
|
||||||
|
assert not re.search(r":\w+::", _DEACTIVATE_SQL)
|
||||||
|
|
||||||
|
|
||||||
|
def test_sql_ttl_param_named_ttl_days() -> None:
|
||||||
|
"""Bind param must be :ttl_days (wired from settings.avito_stale_ttl_days)."""
|
||||||
|
assert ":ttl_days" in _DEACTIVATE_SQL
|
||||||
|
|
||||||
|
|
||||||
|
# ── Settings wiring ───────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_settings_has_avito_stale_ttl_days() -> None:
|
||||||
|
from app.core.config import settings
|
||||||
|
|
||||||
|
assert hasattr(settings, "avito_stale_ttl_days")
|
||||||
|
assert settings.avito_stale_ttl_days == 10 # default
|
||||||
|
|
||||||
|
|
||||||
|
def test_task_reads_ttl_from_settings() -> None:
|
||||||
|
"""Task must use settings.avito_stale_ttl_days, not a hardcoded literal."""
|
||||||
|
assert "settings.avito_stale_ttl_days" in _TASK_SRC
|
||||||
|
|
||||||
|
|
||||||
|
# ── Task function contract ────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_task_returns_deactivated_counter() -> None:
|
||||||
|
assert '"deactivated"' in _TASK_SRC or "'deactivated'" in _TASK_SRC
|
||||||
|
|
||||||
|
|
||||||
|
def test_task_finalises_run() -> None:
|
||||||
|
assert "mark_done" in _TASK_SRC
|
||||||
|
assert "mark_failed" in _TASK_SRC
|
||||||
|
|
||||||
|
|
||||||
|
def test_task_commits_and_rollbacks_on_error() -> None:
|
||||||
|
assert "db.commit()" in _TASK_SRC
|
||||||
|
assert "db.rollback()" in _TASK_SRC
|
||||||
|
|
||||||
|
|
||||||
|
# ── Scheduler wiring ──────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_trigger_fn_exists_and_is_async() -> None:
|
||||||
|
fn = getattr(scheduler, "trigger_deactivate_stale_avito_run", None)
|
||||||
|
assert fn is not None, "trigger_deactivate_stale_avito_run missing from scheduler"
|
||||||
|
assert inspect.iscoroutinefunction(fn)
|
||||||
|
|
||||||
|
|
||||||
|
def test_dispatch_branch_wired() -> None:
|
||||||
|
"""scheduler_loop dispatches source='deactivate_stale_avito' to the trigger."""
|
||||||
|
loop_src = inspect.getsource(scheduler.scheduler_loop)
|
||||||
|
assert 'source == "deactivate_stale_avito"' in loop_src
|
||||||
|
assert "trigger_deactivate_stale_avito_run(db, sch)" in loop_src
|
||||||
|
|
||||||
|
|
||||||
|
def test_trigger_runs_sync_task_in_executor() -> None:
|
||||||
|
"""Sync DB-only task → run_in_executor, not a bare await."""
|
||||||
|
trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_avito_run)
|
||||||
|
# run_in_executor call may span two lines; check both parts separately.
|
||||||
|
assert "run_in_executor(" in trig_src
|
||||||
|
assert "deactivate_stale_avito_listings" in trig_src
|
||||||
|
assert "_claim_run(db, schedule_row)" in trig_src
|
||||||
|
|
||||||
|
|
||||||
|
# ── Migration 090 ─────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_migration_090_exists() -> None:
|
||||||
|
assert _MIGRATION_090.is_file(), f"missing migration: {_MIGRATION_090}"
|
||||||
|
|
||||||
|
|
||||||
|
def test_migration_090_seeds_correct_source_and_window() -> None:
|
||||||
|
sql = _MIGRATION_090.read_text("utf-8")
|
||||||
|
assert "'deactivate_stale_avito'" in sql
|
||||||
|
# Window 06:00-07:00 UTC — after avito sweep (02:00-05:00).
|
||||||
|
assert re.search(r"\b6\b", sql), "window_start_hour 6 missing"
|
||||||
|
assert re.search(r"\b7\b", sql), "window_end_hour 7 missing"
|
||||||
|
|
||||||
|
|
||||||
|
def test_migration_090_is_idempotent() -> None:
|
||||||
|
sql = _MIGRATION_090.read_text("utf-8")
|
||||||
|
assert "ON CONFLICT (source) DO NOTHING" in sql
|
||||||
|
|
||||||
|
|
||||||
|
def test_migration_090_enabled_true() -> None:
|
||||||
|
sql = _MIGRATION_090.read_text("utf-8")
|
||||||
|
insert_block = sql.split("INSERT INTO scrape_schedules")[1]
|
||||||
|
# enabled=true (SAFE — pure internal DB UPDATE).
|
||||||
|
assert "true," in insert_block
|
||||||
|
|
||||||
|
|
||||||
|
def test_migration_090_is_transactional() -> None:
|
||||||
|
sql = _MIGRATION_090.read_text("utf-8")
|
||||||
|
assert "BEGIN;" in sql
|
||||||
|
assert "COMMIT;" in sql
|
||||||
|
|
||||||
|
|
||||||
|
def test_migration_090_no_psycopg_trap() -> None:
|
||||||
|
"""Migration is plain DDL/DML seed (no bind params), guard :x::type trap anyway."""
|
||||||
|
sql = _MIGRATION_090.read_text("utf-8")
|
||||||
|
assert not re.search(r":\w+::", sql)
|
||||||
|
|
||||||
|
|
||||||
|
# ── Behavioural test: counter logic via fake db ───────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
class _FakeResult:
|
||||||
|
def __init__(self, rowcount: int) -> None:
|
||||||
|
self.rowcount = rowcount
|
||||||
|
|
||||||
|
|
||||||
|
class _FakeDB:
|
||||||
|
"""Minimal stand-in for SQLAlchemy Session."""
|
||||||
|
|
||||||
|
def __init__(self, rowcount: int) -> None:
|
||||||
|
self._rowcount = rowcount
|
||||||
|
self.executed: list[Any] = []
|
||||||
|
self.committed = False
|
||||||
|
self.rolled_back = False
|
||||||
|
|
||||||
|
def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
|
||||||
|
self.executed.append((stmt, params))
|
||||||
|
return _FakeResult(self._rowcount)
|
||||||
|
|
||||||
|
def commit(self) -> None:
|
||||||
|
self.committed = True
|
||||||
|
|
||||||
|
def rollback(self) -> None:
|
||||||
|
self.rolled_back = True
|
||||||
|
|
||||||
|
|
||||||
|
def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
"""deactivate_stale_avito_listings maps execute() rowcount → counters + marks done."""
|
||||||
|
marked: dict[str, Any] = {}
|
||||||
|
monkeypatch.setattr(
|
||||||
|
task_mod.runs_mod,
|
||||||
|
"mark_done",
|
||||||
|
lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
|
||||||
|
|
||||||
|
db = _FakeDB(rowcount=137)
|
||||||
|
out = task_mod.deactivate_stale_avito_listings(db, run_id=42) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
assert out == {"deactivated": 137}
|
||||||
|
assert db.committed is True
|
||||||
|
assert len(db.executed) == 1
|
||||||
|
# Bind params must contain ttl_days wired from settings.
|
||||||
|
_stmt, params = db.executed[0]
|
||||||
|
assert params is not None
|
||||||
|
assert "ttl_days" in params
|
||||||
|
assert params["ttl_days"] == 10 # default from settings
|
||||||
|
# run_id finalised via mark_done.
|
||||||
|
assert marked["run_id"] == 42
|
||||||
|
assert marked["counters"] == {"deactivated": 137}
|
||||||
|
|
||||||
|
|
||||||
|
def test_counter_logic_zero_rows(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
"""rowcount=0 → deactivated=0, still commits and marks done."""
|
||||||
|
marked: dict[str, Any] = {}
|
||||||
|
monkeypatch.setattr(
|
||||||
|
task_mod.runs_mod,
|
||||||
|
"mark_done",
|
||||||
|
lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
|
||||||
|
|
||||||
|
db = _FakeDB(rowcount=0)
|
||||||
|
out = task_mod.deactivate_stale_avito_listings(db, run_id=1) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
assert out == {"deactivated": 0}
|
||||||
|
assert db.committed is True
|
||||||
|
assert marked["counters"] == {"deactivated": 0}
|
||||||
|
|
||||||
|
|
||||||
|
def test_failure_path_rollback_and_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
"""On execute error: rollback + mark_failed + re-raise (no silent swallow)."""
|
||||||
|
failed: dict[str, Any] = {}
|
||||||
|
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
task_mod.runs_mod,
|
||||||
|
"mark_failed",
|
||||||
|
lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
|
||||||
|
)
|
||||||
|
|
||||||
|
class _BoomDB(_FakeDB):
|
||||||
|
def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
|
||||||
|
raise RuntimeError("db exploded")
|
||||||
|
|
||||||
|
db = _BoomDB(rowcount=0)
|
||||||
|
with pytest.raises(RuntimeError, match="db exploded"):
|
||||||
|
task_mod.deactivate_stale_avito_listings(db, run_id=7) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
assert db.rolled_back is True
|
||||||
|
assert failed["run_id"] == 7
|
||||||
Loading…
Add table
Reference in a new issue