gendesign/tradein-mvp/backend/tests/test_deactivate_stale_avito.py
bot-backend 652ea75f54 feat(tradein): nightly task to deactivate stale avito listings (#759)
Avito listings re-seen by the scraper get is_active=true but were never
deactivated when they stop appearing (sold/removed) — unlike cian's snapshot
scope. Result: all avito rows stay is_active=true, inflating the coverage
denominator and polluting estimator analog medians with dead lots.

Add scheduled task deactivate_stale_avito_listings: marks source='avito'
is_active=false when last_seen_at older than AVITO_STALE_TTL_DAYS (10d),
window 06:00-07:00 UTC (after avito_city_sweep). No DELETE — history kept
for backtest #667. cian/yandex untouched. Wired into scheduler_loop dispatch
+ seed migration 090 (ON CONFLICT(source) idempotent).

Refs #759
2026-05-31 10:10:59 +03:00

254 lines
9.6 KiB
Python

"""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