feat(scrapers): segment-aware generic deactivate_stale for yandex/cian #1558
5 changed files with 540 additions and 37 deletions
|
|
@ -23,10 +23,13 @@ Sources:
|
|||
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
|
||||
(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)
|
||||
- deactivate_stale_* → trigger_deactivate_stale_run → deactivate_stale_listings
|
||||
(tasks/deactivate_stale_avito.py, #759 + segment-aware
|
||||
generalisation; marks listings with last_seen_at > TTL days as
|
||||
is_active=false — no DELETE, no external calls. Source prefix
|
||||
dispatch covers deactivate_stale_avito (all segments, TTL=10) +
|
||||
deactivate_stale_yandex/cian (vtorichka only, TTL=30, segments
|
||||
from default_params). Window after the source sweep.)
|
||||
- sber_index_pull → run_sber_index_pull
|
||||
(tasks/sber_index_pull.py, #887; monthly pull of СберИндекс
|
||||
city-level secondary-housing price index — pure HTTP+DB, no
|
||||
|
|
@ -708,6 +711,59 @@ async def trigger_deactivate_stale_avito_run(
|
|||
return run_id
|
||||
|
||||
|
||||
async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||||
"""Универсальный триггер для всех deactivate_stale_* источников.
|
||||
|
||||
Читает listing_source, ttl_days, segments из schedule_row["default_params"] (jsonb).
|
||||
Для источника 'deactivate_stale_avito' с пустым default_params использует defaults:
|
||||
listing_source='avito', ttl_days=settings.avito_stale_ttl_days, segments=None.
|
||||
Для yandex/cian -- default_params обязан содержать listing_source + ttl_days + segments.
|
||||
|
||||
Запускает deactivate_stale_listings в executor (sync DB-only task).
|
||||
Возвращает run_id или None (skip -- already running).
|
||||
"""
|
||||
from app.core.config import settings as _settings
|
||||
|
||||
run_id = _claim_run(db, schedule_row)
|
||||
if run_id is None:
|
||||
return None
|
||||
|
||||
params = schedule_row.get("default_params") or {}
|
||||
listing_source: str = params.get("listing_source", "avito")
|
||||
ttl_days: int = params.get("ttl_days", _settings.avito_stale_ttl_days)
|
||||
segments: list[str] | None = params.get("segments") # list or None
|
||||
|
||||
async def _run() -> None:
|
||||
run_db = SessionLocal()
|
||||
try:
|
||||
from app.tasks.deactivate_stale_avito import deactivate_stale_listings
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
await loop.run_in_executor(
|
||||
None,
|
||||
lambda: deactivate_stale_listings(
|
||||
run_db,
|
||||
run_id,
|
||||
listing_source=listing_source,
|
||||
ttl_days=ttl_days,
|
||||
segments=segments,
|
||||
),
|
||||
)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"scheduler: deactivate_stale source=%s crashed run_id=%d",
|
||||
listing_source,
|
||||
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 source=%s run_id=%d", listing_source, run_id)
|
||||
return run_id
|
||||
|
||||
|
||||
async def trigger_sber_index_pull_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||||
"""Создать scrape_runs + launch run_sber_index_pull в asyncio.create_task.
|
||||
|
||||
|
|
@ -1268,8 +1324,8 @@ async def scheduler_loop() -> None:
|
|||
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":
|
||||
await trigger_deactivate_stale_avito_run(db, sch)
|
||||
elif source.startswith("deactivate_stale_"):
|
||||
await trigger_deactivate_stale_run(db, sch)
|
||||
elif source == "sber_index_pull":
|
||||
await trigger_sber_index_pull_run(db, sch)
|
||||
elif source == "rosreestr_quarter_poll":
|
||||
|
|
|
|||
|
|
@ -1,22 +1,27 @@
|
|||
"""Deactivate stale avito listings that have not been seen in >TTL days (#759).
|
||||
"""Deactivate stale listings that have not been seen in >TTL days.
|
||||
|
||||
Нечасто обновляемые объявления на Avito (last_seen_at < NOW() - TTL) помечаются
|
||||
is_active=false, чтобы coverage/estimator denominators не раздувались мёртвыми
|
||||
объявлениями. Строки НЕ удаляются — история нужна для бэктеста (#667).
|
||||
Первоначально создано для avito (#759). Теперь поддерживает любой источник
|
||||
(listing_source) с опциональной фильтрацией по listing_segment.
|
||||
|
||||
Только source='avito'. Cian/Yandex не затрагиваются.
|
||||
Ключевые решения:
|
||||
- Cian/Yandex не поддерживают full-coverage sweep -> паушальный TTL сломает живой
|
||||
инвентарь. DECISION: для yandex/cian деактивировать ТОЛЬКО listing_segment='vtorichka',
|
||||
TTL=30. novostroyki (9659 активных первичных строк) и NULL-сегмент не трогаем.
|
||||
- avito: все сегменты (segments=None), TTL=10 дней -- поведение без изменений.
|
||||
- Строки НЕ удаляются -- история нужна для бэктеста (#667).
|
||||
|
||||
Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app
|
||||
scheduler'ом через trigger_deactivate_stale_avito_run() (scheduler.py),
|
||||
Задача синхронная (DB-only, никаких внешних HTTP-вызовов) -- запускается in-app
|
||||
scheduler'ом через trigger_deactivate_stale_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).
|
||||
TTL для avito берётся из settings.avito_stale_ttl_days (env AVITO_STALE_TTL_DAYS, default 10).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
|
@ -26,45 +31,107 @@ 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 без фильтра по сегменту: все сегменты для данного source.
|
||||
# CAST(:ttl_days || ' days' AS interval) -- psycopg v3 safe (никаких :param::type).
|
||||
_DEACTIVATE_SQL_ALL_SEGMENTS = text(
|
||||
"""
|
||||
UPDATE listings
|
||||
SET is_active = false
|
||||
WHERE source = 'avito'
|
||||
WHERE source = :listing_source
|
||||
AND is_active = true
|
||||
AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||
"""
|
||||
)
|
||||
|
||||
# UPDATE с фильтром по сегменту (segments задан): затрагивает только указанные сегменты.
|
||||
# = ANY(CAST(:segments AS text[])) -- psycopg v3 адаптирует Python list -> text[].
|
||||
_DEACTIVATE_SQL_SEGMENTS = text(
|
||||
"""
|
||||
UPDATE listings
|
||||
SET is_active = false
|
||||
WHERE source = :listing_source
|
||||
AND is_active = true
|
||||
AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||
AND listing_segment = ANY(CAST(:segments AS text[]))
|
||||
"""
|
||||
)
|
||||
|
||||
def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]:
|
||||
"""Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад.
|
||||
# Алиас для обратной совместимости -- использовался в тестах через task_mod._DEACTIVATE_SQL
|
||||
_DEACTIVATE_SQL = _DEACTIVATE_SQL_ALL_SEGMENTS
|
||||
|
||||
|
||||
def deactivate_stale_listings(
|
||||
db: Session,
|
||||
run_id: int,
|
||||
*,
|
||||
listing_source: str,
|
||||
ttl_days: int,
|
||||
segments: list[str] | None = None,
|
||||
) -> dict[str, int]:
|
||||
"""Пометить is_active=false объявления с last_seen_at > ttl_days дней назад.
|
||||
|
||||
Параметры:
|
||||
listing_source: значение поля source в таблице listings ('avito', 'yandex', 'cian').
|
||||
ttl_days: количество дней TTL; объявления старше этого порога деактивируются.
|
||||
segments: если задан -- деактивировать только объявления с указанными
|
||||
listing_segment значениями. None -> все сегменты (поведение avito по умолчанию).
|
||||
|
||||
Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
|
||||
Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed).
|
||||
|
||||
Returns {"deactivated": N} — количество обновлённых строк.
|
||||
Returns {"deactivated": N} -- количество обновлённых строк.
|
||||
"""
|
||||
counters: dict[str, int] = {"deactivated": 0}
|
||||
try:
|
||||
result = db.execute(_DEACTIVATE_SQL, {"ttl_days": settings.avito_stale_ttl_days})
|
||||
# segments is None -> все сегменты (поведение avito). segments=[...] -> только
|
||||
# перечисленные сегменты. Используем `is not None` (НЕ truthy): пустой список []
|
||||
# означает "ни один сегмент" (= ANY(ARRAY[]) ничего не матчит, деактивирует 0),
|
||||
# а НЕ "все сегменты" — иначе случайный [] стёр бы весь источник.
|
||||
if segments is not None:
|
||||
params: dict[str, Any] = {
|
||||
"listing_source": listing_source,
|
||||
"ttl_days": ttl_days,
|
||||
"segments": segments,
|
||||
}
|
||||
result = db.execute(_DEACTIVATE_SQL_SEGMENTS, params)
|
||||
else:
|
||||
params = {"listing_source": listing_source, "ttl_days": ttl_days}
|
||||
result = db.execute(_DEACTIVATE_SQL_ALL_SEGMENTS, params)
|
||||
|
||||
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)",
|
||||
"deactivate_stale source=%s run_id=%d done: deactivated=%d "
|
||||
"(ttl_days=%d, segments=%r)",
|
||||
listing_source,
|
||||
run_id,
|
||||
counters["deactivated"],
|
||||
settings.avito_stale_ttl_days,
|
||||
ttl_days,
|
||||
segments,
|
||||
)
|
||||
return counters
|
||||
except Exception as exc:
|
||||
logger.exception("deactivate_stale_avito run_id=%d failed", run_id)
|
||||
logger.exception("deactivate_stale source=%s run_id=%d failed", listing_source, run_id)
|
||||
db.rollback()
|
||||
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
|
||||
raise
|
||||
|
||||
|
||||
def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]:
|
||||
"""Обёртка для обратной совместимости -- deactivate_stale_avito (все сегменты, TTL из settings).
|
||||
|
||||
Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад.
|
||||
Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
|
||||
Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed).
|
||||
|
||||
Returns {"deactivated": N} -- количество обновлённых строк.
|
||||
"""
|
||||
return deactivate_stale_listings(
|
||||
db,
|
||||
run_id,
|
||||
listing_source="avito",
|
||||
ttl_days=settings.avito_stale_ttl_days,
|
||||
segments=None,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,68 @@
|
|||
-- 115_scrape_schedules_seed_deactivate_stale_yandex_cian.sql
|
||||
-- Seed scrape_schedules rows for nightly segment-aware stale-listing deactivation
|
||||
-- of yandex and cian vtorichka (#759 generalisation).
|
||||
--
|
||||
-- Rationale (segment-aware vs blanket TTL):
|
||||
-- Yandex and Cian city sweeps do NOT maintain full inventory coverage -- they
|
||||
-- sample a rolling window rather than crawling every live listing on every run.
|
||||
-- A blanket TTL (as used for Avito) would therefore wrongly deactivate live
|
||||
-- novostroyki and unknown-segment (NULL) rows that simply were not re-visited
|
||||
-- in the sample window.
|
||||
--
|
||||
-- DECISION (user-approved): deactivate ONLY listing_segment='vtorichka' rows,
|
||||
-- with TTL=30 days. This is conservative enough that at first deploy it will
|
||||
-- touch 0 rows (no vtorichka listing has been absent for 30 days yet), but will
|
||||
-- catch genuinely-stale secondhand inventory going forward.
|
||||
--
|
||||
-- novostroyki rows (9659 active first-party rows as of 2026-06-16) are protected:
|
||||
-- they are managed by dedicated newbuilding pipelines and must not be deactivated
|
||||
-- by this task. NULL-segment rows are also excluded as their segment is unknown.
|
||||
--
|
||||
-- Schedule window 07:00-08:00 UTC:
|
||||
-- After avito deactivate_stale (06:00-07:00 UTC) and after the main city sweeps
|
||||
-- (02:00-05:00 UTC) and Rosreestr OS-cron (05:30/05:45 UTC).
|
||||
-- Pattern: next_run_at bootstrapped to tomorrow 07:00 UTC so the scheduler does
|
||||
-- not fire immediately on deploy.
|
||||
--
|
||||
-- Idempotent: ON CONFLICT (source) DO NOTHING -- safe to re-apply.
|
||||
--
|
||||
-- Dependencies:
|
||||
-- 052_scrape_schedules.sql (table + UNIQUE(source)).
|
||||
-- listings table with listing_segment column.
|
||||
-- app/tasks/deactivate_stale_avito.py -- deactivate_stale_listings() (generic fn).
|
||||
-- app/services/scheduler.py -- trigger_deactivate_stale_run() + dispatch branch
|
||||
-- source.startswith("deactivate_stale_").
|
||||
--
|
||||
-- Deploy order:
|
||||
-- Apply after deploying the scheduler.py + task changes.
|
||||
|
||||
BEGIN;
|
||||
|
||||
INSERT INTO scrape_schedules (
|
||||
source,
|
||||
enabled,
|
||||
window_start_hour,
|
||||
window_end_hour,
|
||||
next_run_at,
|
||||
default_params
|
||||
)
|
||||
VALUES
|
||||
(
|
||||
'deactivate_stale_yandex',
|
||||
true,
|
||||
7,
|
||||
8,
|
||||
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC',
|
||||
'{"listing_source":"yandex","ttl_days":30,"segments":["vtorichka"]}'::jsonb
|
||||
),
|
||||
(
|
||||
'deactivate_stale_cian',
|
||||
true,
|
||||
7,
|
||||
8,
|
||||
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC',
|
||||
'{"listing_source":"cian","ttl_days":30,"segments":["vtorichka"]}'::jsonb
|
||||
)
|
||||
ON CONFLICT (source) DO NOTHING;
|
||||
|
||||
COMMIT;
|
||||
|
|
@ -25,14 +25,18 @@ _MIGRATION_090 = _SQL_DIR / "090_scrape_schedules_seed_deactivate_stale_avito.sq
|
|||
|
||||
_DEACTIVATE_SQL = str(task_mod._DEACTIVATE_SQL.text)
|
||||
_TASK_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings)
|
||||
_GENERIC_TASK_SRC = inspect.getsource(task_mod.deactivate_stale_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
|
||||
"""Compat wrapper enforces source='avito' via listing_source param — cian/yandex not touched."""
|
||||
# _DEACTIVATE_SQL is now the generic all-segments SQL using :listing_source bind param.
|
||||
assert ":listing_source" in _DEACTIVATE_SQL
|
||||
# The avito constraint is enforced by the compat wrapper passing listing_source="avito".
|
||||
assert 'listing_source="avito"' in _TASK_SRC
|
||||
|
||||
|
||||
def test_sql_only_deactivates_not_deletes() -> None:
|
||||
|
|
@ -83,17 +87,20 @@ def test_task_reads_ttl_from_settings() -> None:
|
|||
|
||||
|
||||
def test_task_returns_deactivated_counter() -> None:
|
||||
assert '"deactivated"' in _TASK_SRC or "'deactivated'" in _TASK_SRC
|
||||
# counter logic lives in the generic deactivate_stale_listings function
|
||||
assert '"deactivated"' in _GENERIC_TASK_SRC or "'deactivated'" in _GENERIC_TASK_SRC
|
||||
|
||||
|
||||
def test_task_finalises_run() -> None:
|
||||
assert "mark_done" in _TASK_SRC
|
||||
assert "mark_failed" in _TASK_SRC
|
||||
# mark_done/mark_failed live in the generic deactivate_stale_listings function
|
||||
assert "mark_done" in _GENERIC_TASK_SRC
|
||||
assert "mark_failed" in _GENERIC_TASK_SRC
|
||||
|
||||
|
||||
def test_task_commits_and_rollbacks_on_error() -> None:
|
||||
assert "db.commit()" in _TASK_SRC
|
||||
assert "db.rollback()" in _TASK_SRC
|
||||
# commit/rollback live in the generic deactivate_stale_listings function
|
||||
assert "db.commit()" in _GENERIC_TASK_SRC
|
||||
assert "db.rollback()" in _GENERIC_TASK_SRC
|
||||
|
||||
|
||||
# ── Scheduler wiring ──────────────────────────────────────────────────────────
|
||||
|
|
@ -106,10 +113,10 @@ def test_trigger_fn_exists_and_is_async() -> None:
|
|||
|
||||
|
||||
def test_dispatch_branch_wired() -> None:
|
||||
"""scheduler_loop dispatches source='deactivate_stale_avito' to the trigger."""
|
||||
"""scheduler_loop dispatches deactivate_stale_* sources to the generic 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
|
||||
assert 'source.startswith("deactivate_stale_")' in loop_src
|
||||
assert "trigger_deactivate_stale_run(db, sch)" in loop_src
|
||||
|
||||
|
||||
def test_trigger_runs_sync_task_in_executor() -> None:
|
||||
|
|
|
|||
305
tradein-mvp/backend/tests/test_deactivate_stale_listings.py
Normal file
305
tradein-mvp/backend/tests/test_deactivate_stale_listings.py
Normal file
|
|
@ -0,0 +1,305 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
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_115 = _SQL_DIR / "115_scrape_schedules_seed_deactivate_stale_yandex_cian.sql"
|
||||
|
||||
_GENERIC_FN_SRC = inspect.getsource(task_mod.deactivate_stale_listings)
|
||||
_COMPAT_FN_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings)
|
||||
|
||||
|
||||
def test_segment_sql_uses_cast_interval_not_double_colon() -> None:
|
||||
sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text
|
||||
assert "CAST(:ttl_days || ' days' AS interval)" in sql
|
||||
assert not re.search(r"\w+::", sql)
|
||||
|
||||
|
||||
def test_segment_sql_uses_any_cast_text_array() -> None:
|
||||
sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text
|
||||
assert "ANY(CAST(:segments AS text[]))" in sql
|
||||
|
||||
|
||||
def test_segment_sql_filters_is_active_true() -> None:
|
||||
sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text
|
||||
assert "is_active = true" in sql
|
||||
|
||||
|
||||
def test_segment_sql_only_deactivates_not_deletes() -> None:
|
||||
sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text
|
||||
assert "SET is_active = false" in sql
|
||||
assert "DELETE" not in sql.upper()
|
||||
|
||||
|
||||
def test_all_segments_sql_uses_listing_source_param() -> None:
|
||||
sql = task_mod._DEACTIVATE_SQL_ALL_SEGMENTS.text
|
||||
assert ":listing_source" in sql
|
||||
assert "source = 'avito'" not in sql
|
||||
|
||||
|
||||
def test_deactivate_sql_alias_is_all_segments() -> None:
|
||||
assert task_mod._DEACTIVATE_SQL is task_mod._DEACTIVATE_SQL_ALL_SEGMENTS
|
||||
|
||||
|
||||
def test_generic_fn_exists() -> None:
|
||||
assert callable(task_mod.deactivate_stale_listings)
|
||||
|
||||
|
||||
def test_generic_fn_returns_deactivated_counter() -> None:
|
||||
assert "deactivated" in _GENERIC_FN_SRC
|
||||
|
||||
|
||||
def test_generic_fn_finalises_run() -> None:
|
||||
assert "mark_done" in _GENERIC_FN_SRC
|
||||
assert "mark_failed" in _GENERIC_FN_SRC
|
||||
|
||||
|
||||
def test_generic_fn_commits_and_rollbacks() -> None:
|
||||
assert "db.commit()" in _GENERIC_FN_SRC
|
||||
assert "db.rollback()" in _GENERIC_FN_SRC
|
||||
|
||||
|
||||
def test_compat_wrapper_calls_generic_with_avito_defaults() -> None:
|
||||
assert "deactivate_stale_listings(" in _COMPAT_FN_SRC
|
||||
assert 'listing_source="avito"' in _COMPAT_FN_SRC
|
||||
assert "settings.avito_stale_ttl_days" in _COMPAT_FN_SRC
|
||||
assert "segments=None" in _COMPAT_FN_SRC
|
||||
|
||||
|
||||
def test_trigger_deactivate_stale_run_exists_and_is_async() -> None:
|
||||
fn = getattr(scheduler, "trigger_deactivate_stale_run", None)
|
||||
assert fn is not None, "trigger_deactivate_stale_run missing from scheduler"
|
||||
assert inspect.iscoroutinefunction(fn)
|
||||
|
||||
|
||||
def test_dispatch_branch_uses_startswith() -> None:
|
||||
loop_src = inspect.getsource(scheduler.scheduler_loop)
|
||||
assert 'source.startswith("deactivate_stale_")' in loop_src
|
||||
assert "trigger_deactivate_stale_run(db, sch)" in loop_src
|
||||
|
||||
|
||||
def test_trigger_runs_sync_task_in_executor() -> None:
|
||||
trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_run)
|
||||
assert "run_in_executor(" in trig_src
|
||||
assert "deactivate_stale_listings" in trig_src
|
||||
assert "_claim_run(db, schedule_row)" in trig_src
|
||||
|
||||
|
||||
def test_trigger_reads_params_from_default_params() -> None:
|
||||
trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_run)
|
||||
assert "listing_source" in trig_src
|
||||
assert "ttl_days" in trig_src
|
||||
assert "segments" in trig_src
|
||||
|
||||
|
||||
class _FakeResult:
|
||||
def __init__(self, rowcount: int) -> None:
|
||||
self.rowcount = rowcount
|
||||
|
||||
|
||||
class _FakeDB:
|
||||
def __init__(self, rowcount: int = 0) -> None:
|
||||
self._rowcount = rowcount
|
||||
self.executed: list[tuple[Any, 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_vtorichka_stale_beyond_ttl_deactivated(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
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=5)
|
||||
out = task_mod.deactivate_stale_listings(
|
||||
db, run_id=1, listing_source="yandex", ttl_days=30, segments=["vtorichka"]
|
||||
) # type: ignore[arg-type]
|
||||
assert out == {"deactivated": 5}
|
||||
assert db.committed is True
|
||||
_stmt, params = db.executed[0]
|
||||
assert params is not None
|
||||
assert params["segments"] == ["vtorichka"]
|
||||
assert params["listing_source"] == "yandex"
|
||||
assert params["ttl_days"] == 30
|
||||
assert marked["counters"] == {"deactivated": 5}
|
||||
|
||||
|
||||
def test_novostroyki_not_in_segment_filter(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
|
||||
db = _FakeDB(rowcount=0)
|
||||
task_mod.deactivate_stale_listings(
|
||||
db, run_id=2, listing_source="yandex", ttl_days=30, segments=["vtorichka"]
|
||||
) # type: ignore[arg-type]
|
||||
_stmt, params = db.executed[0]
|
||||
assert params is not None
|
||||
assert "segments" in params
|
||||
assert "novostroyki" not in params["segments"]
|
||||
|
||||
|
||||
def test_null_segment_not_in_filter(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
|
||||
db = _FakeDB(rowcount=0)
|
||||
task_mod.deactivate_stale_listings(
|
||||
db, run_id=3, listing_source="cian", ttl_days=30, segments=["vtorichka"]
|
||||
) # type: ignore[arg-type]
|
||||
_stmt, params = db.executed[0]
|
||||
assert params is not None
|
||||
assert None not in params["segments"]
|
||||
|
||||
|
||||
def test_other_source_listing_source_param_matches(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
|
||||
db = _FakeDB(rowcount=0)
|
||||
task_mod.deactivate_stale_listings(
|
||||
db, run_id=4, listing_source="yandex", ttl_days=30, segments=["vtorichka"]
|
||||
) # type: ignore[arg-type]
|
||||
_stmt, params = db.executed[0]
|
||||
assert params is not None
|
||||
assert params["listing_source"] == "yandex"
|
||||
|
||||
|
||||
def test_fresh_listing_zero_rowcount(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
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_listings(
|
||||
db, run_id=5, listing_source="cian", ttl_days=30, segments=["vtorichka"]
|
||||
) # type: ignore[arg-type]
|
||||
assert out == {"deactivated": 0}
|
||||
assert db.committed is True
|
||||
assert marked["counters"] == {"deactivated": 0}
|
||||
|
||||
|
||||
def test_segments_none_uses_all_segments_sql(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
|
||||
db = _FakeDB(rowcount=3)
|
||||
out = task_mod.deactivate_stale_listings(
|
||||
db, run_id=6, listing_source="avito", ttl_days=10, segments=None
|
||||
) # type: ignore[arg-type]
|
||||
assert out == {"deactivated": 3}
|
||||
_stmt, params = db.executed[0]
|
||||
assert params is not None
|
||||
assert "segments" not in params
|
||||
assert params["listing_source"] == "avito"
|
||||
assert params["ttl_days"] == 10
|
||||
|
||||
|
||||
def test_avito_compat_wrapper_counter(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
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=42)
|
||||
out = task_mod.deactivate_stale_avito_listings(db, run_id=7) # type: ignore[arg-type]
|
||||
assert out == {"deactivated": 42}
|
||||
assert db.committed is True
|
||||
_stmt, params = db.executed[0]
|
||||
assert params is not None
|
||||
assert params["listing_source"] == "avito"
|
||||
assert "segments" not in params
|
||||
assert marked["run_id"] == 7
|
||||
|
||||
|
||||
def test_failure_path_rollback_and_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
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()
|
||||
with pytest.raises(RuntimeError, match="db exploded"):
|
||||
task_mod.deactivate_stale_listings(
|
||||
db, run_id=8, listing_source="yandex", ttl_days=30, segments=["vtorichka"]
|
||||
) # type: ignore[arg-type]
|
||||
assert db.rolled_back is True
|
||||
assert failed["run_id"] == 8
|
||||
|
||||
|
||||
def test_migration_115_exists() -> None:
|
||||
assert _MIGRATION_115.is_file(), f"missing migration: {_MIGRATION_115}"
|
||||
|
||||
|
||||
def test_migration_115_seeds_yandex_and_cian() -> None:
|
||||
sql = _MIGRATION_115.read_text("utf-8")
|
||||
assert "'deactivate_stale_yandex'" in sql
|
||||
assert "'deactivate_stale_cian'" in sql
|
||||
|
||||
|
||||
def test_migration_115_is_idempotent() -> None:
|
||||
sql = _MIGRATION_115.read_text("utf-8")
|
||||
assert "ON CONFLICT (source) DO NOTHING" in sql
|
||||
|
||||
|
||||
def test_migration_115_enabled_true() -> None:
|
||||
sql = _MIGRATION_115.read_text("utf-8")
|
||||
assert "true" in sql
|
||||
|
||||
|
||||
def test_migration_115_is_transactional() -> None:
|
||||
sql = _MIGRATION_115.read_text("utf-8")
|
||||
assert "BEGIN;" in sql
|
||||
assert "COMMIT;" in sql
|
||||
|
||||
|
||||
def test_migration_115_window_7_to_8_utc() -> None:
|
||||
sql = _MIGRATION_115.read_text("utf-8")
|
||||
assert re.search(r"\b7\b", sql), "window_start_hour 7 missing"
|
||||
assert re.search(r"\b8\b", sql), "window_end_hour 8 missing"
|
||||
|
||||
|
||||
def test_migration_115_segments_vtorichka_in_default_params() -> None:
|
||||
sql = _MIGRATION_115.read_text("utf-8")
|
||||
assert "vtorichka" in sql
|
||||
|
||||
|
||||
def test_migration_115_ttl_30_days() -> None:
|
||||
sql = _MIGRATION_115.read_text("utf-8")
|
||||
assert "30" in sql
|
||||
|
||||
|
||||
def test_migration_115_no_psycopg_trap() -> None:
|
||||
sql = _MIGRATION_115.read_text("utf-8")
|
||||
assert not re.search(r":\w+::", sql)
|
||||
Loading…
Add table
Reference in a new issue