Снятие устаревших объявлений работает только там, где есть пересбор #3526
3 changed files with 376 additions and 10 deletions
|
|
@ -239,6 +239,12 @@ async def _job_deactivate_stale(
|
|||
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")
|
||||
# Регионы снятия. None (ключа нет в default_params) -> дефолт самой джобы,
|
||||
# DEFAULT_DEACTIVATION_REGION_CODES=(66,): снимать устаревшее можно только там,
|
||||
# где идёт регулярный пересбор -- см. развёрнутый комментарий у константы в
|
||||
# deactivate_stale_avito.py. Расширять список ВМЕСТЕ с запуском сбора по
|
||||
# региону, ключом "region_codes" (jsonb-массив кодов) в строке расписания.
|
||||
region_codes: list[int] | None = params.get("region_codes")
|
||||
staleness_column: str = params.get("staleness_column", "last_seen_at")
|
||||
# Гейт по здоровью сбора (#2659) включён по умолчанию: незасеянное расписание
|
||||
# получает страховочный порог, а не «деактивируй вслепую». Посчитанные по
|
||||
|
|
@ -278,6 +284,7 @@ async def _job_deactivate_stale(
|
|||
listing_source=listing_source,
|
||||
ttl_days=ttl_days,
|
||||
segments=segments,
|
||||
region_codes=region_codes,
|
||||
staleness_column=staleness_column,
|
||||
min_confirmations=min_confirmations,
|
||||
revisit_floor_quantile=revisit_floor_quantile,
|
||||
|
|
|
|||
|
|
@ -488,12 +488,39 @@ def _build_confirmations_sql(
|
|||
)
|
||||
|
||||
|
||||
# ── Регионы, где снятие устаревших вообще осмысленно ──────────────────────────
|
||||
# Снятие читает МОЛЧАНИЕ как «объявление ушло»: строку давно не видели свежей ->
|
||||
# is_active=false. Такой вывод верен ТОЛЬКО там, где есть механизм пересбора: без
|
||||
# регулярного обхода «давно не видели» неотличимо от «мы туда ни разу не ходили
|
||||
# после разовой загрузки», и джоба снимет разом весь загруженный корпус региона.
|
||||
#
|
||||
# На 15.09.2026 регулярный сбор идёт только по Свердловской области (66): реестр
|
||||
# якорей CITY_ANCHORS в ките покрывает её одну, у ДомКлика гео-скоуп зашит в ЕКБ.
|
||||
# Объявления регионов 77 и 50 попали в listings разовой ручной загрузкой 10-12.09
|
||||
# и с тех пор не обновлялись. Без фильтра ниже около 26.09 кандидатами разом стали
|
||||
# бы 40 410 строк ДомКлика (ttl_days=14 по scraped_at), следом 60 174 cian и
|
||||
# 24 152 yandex (ttl_days=30). Пострадала бы не только Москва с областью:
|
||||
# аварийный потолок max_deactivated (DEFAULT_MAX_DEACTIVATED=15000, см. ниже)
|
||||
# сработал бы раньше, а прогон при skipped_cap_exceeded НЕ ТРОГАЕТ НИ ОДНОЙ
|
||||
# строки -- то есть чистка Екатеринбурга встала бы вместе с московской.
|
||||
#
|
||||
# Отсюда дефолт «снимаем только там, где сами собираем». Расширять список нужно
|
||||
# ВМЕСТЕ с появлением расписания регулярного сбора по региону (якоря кита +
|
||||
# гео-скоуп ДомКлика), а не раньше: пока пересбора нет, нечем отличить ушедшее
|
||||
# объявление от неопрошенного. Переопределяется ключом "region_codes" в
|
||||
# default_params строки расписания (jsonb-массив целых кодов регионов).
|
||||
DEFAULT_DEACTIVATION_REGION_CODES: tuple[int, ...] = (66,)
|
||||
|
||||
|
||||
def _build_all_segments_sql(staleness_column: str) -> Any:
|
||||
"""UPDATE без фильтра по сегменту: все сегменты для данного source.
|
||||
|
||||
staleness_column уже прошёл whitelist-проверку в deactivate_stale_listings,
|
||||
поэтому f-string-подстановка имени колонки безопасна. Значения (:listing_source,
|
||||
:ttl_days, :run_id) остаются param-binding — psycopg v3 safe (никаких :param::type).
|
||||
:region_codes, :ttl_days, :run_id) остаются param-binding — psycopg v3 safe
|
||||
(никаких :param::type). :region_codes -- см. DEFAULT_DEACTIVATION_REGION_CODES:
|
||||
снятие идёт только по регионам с регулярным пересбором. NULL region_code не
|
||||
матчится ANY(...) намеренно -- неизвестный регион нечем отнести к покрытому.
|
||||
"""
|
||||
return text(
|
||||
f"""
|
||||
|
|
@ -501,6 +528,7 @@ def _build_all_segments_sql(staleness_column: str) -> Any:
|
|||
UPDATE listings
|
||||
SET is_active = false
|
||||
WHERE source = :listing_source
|
||||
AND region_code = ANY(CAST(:region_codes AS int[]))
|
||||
AND is_active = true
|
||||
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||
RETURNING id, price_rub
|
||||
|
|
@ -514,7 +542,8 @@ def _build_segments_sql(staleness_column: str) -> Any:
|
|||
"""UPDATE с фильтром по сегменту (segments задан): только указанные сегменты.
|
||||
|
||||
staleness_column уже прошёл whitelist-проверку. = ANY(CAST(:segments AS text[]))
|
||||
-- psycopg v3 адаптирует Python list -> text[].
|
||||
-- psycopg v3 адаптирует Python list -> text[]; то же для :region_codes -> int[]
|
||||
(см. DEFAULT_DEACTIVATION_REGION_CODES).
|
||||
"""
|
||||
return text(
|
||||
f"""
|
||||
|
|
@ -522,6 +551,7 @@ def _build_segments_sql(staleness_column: str) -> Any:
|
|||
UPDATE listings
|
||||
SET is_active = false
|
||||
WHERE source = :listing_source
|
||||
AND region_code = ANY(CAST(:region_codes AS int[]))
|
||||
AND is_active = true
|
||||
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||
AND listing_segment = ANY(CAST(:segments AS text[]))
|
||||
|
|
@ -541,7 +571,8 @@ def _build_null_segment_sql(staleness_column: str) -> Any:
|
|||
отдельный явный предикат. Целенаправленно НЕ трогает 'vtorichka'/'novostroyki' --
|
||||
их деактивация идёт через _build_segments_sql в отдельных, уже существующих джобах.
|
||||
|
||||
staleness_column уже прошёл whitelist-проверку. Без :segments-параметра вовсе.
|
||||
staleness_column уже прошёл whitelist-проверку. Без :segments-параметра вовсе,
|
||||
но с :region_codes (см. DEFAULT_DEACTIVATION_REGION_CODES).
|
||||
"""
|
||||
return text(
|
||||
f"""
|
||||
|
|
@ -549,6 +580,7 @@ def _build_null_segment_sql(staleness_column: str) -> Any:
|
|||
UPDATE listings
|
||||
SET is_active = false
|
||||
WHERE source = :listing_source
|
||||
AND region_code = ANY(CAST(:region_codes AS int[]))
|
||||
AND is_active = true
|
||||
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||
AND listing_segment IS NULL
|
||||
|
|
@ -639,13 +671,16 @@ def _build_all_segments_candidates_count_sql(staleness_column: str) -> Any:
|
|||
_build_all_segments_sql: рефакторинг уже протестированных UPDATE-builder'ов
|
||||
вне скоупа PR-B. Синхронность с UPDATE закреплена тестом
|
||||
test_candidates_predicate_matches_update_predicate
|
||||
(tests/test_deactivate_stale_deactivation_cap.py).
|
||||
(tests/test_deactivate_stale_deactivation_cap.py). Фильтр по :region_codes
|
||||
обязан стоять и здесь: считать кандидатов по всем регионам, а снимать по
|
||||
покрытым сбором -- значит кормить аварийный потолок чужими числами.
|
||||
"""
|
||||
return text(
|
||||
f"""
|
||||
SELECT count(*)
|
||||
FROM listings
|
||||
WHERE source = :listing_source
|
||||
AND region_code = ANY(CAST(:region_codes AS int[]))
|
||||
AND is_active = true
|
||||
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||
"""
|
||||
|
|
@ -661,6 +696,7 @@ def _build_segments_candidates_count_sql(staleness_column: str) -> Any:
|
|||
SELECT count(*)
|
||||
FROM listings
|
||||
WHERE source = :listing_source
|
||||
AND region_code = ANY(CAST(:region_codes AS int[]))
|
||||
AND is_active = true
|
||||
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||
AND listing_segment = ANY(CAST(:segments AS text[]))
|
||||
|
|
@ -677,6 +713,7 @@ def _build_null_segment_candidates_count_sql(staleness_column: str) -> Any:
|
|||
SELECT count(*)
|
||||
FROM listings
|
||||
WHERE source = :listing_source
|
||||
AND region_code = ANY(CAST(:region_codes AS int[]))
|
||||
AND is_active = true
|
||||
AND {staleness_column} < NOW() - CAST(:ttl_days || ' days' AS interval)
|
||||
AND listing_segment IS NULL
|
||||
|
|
@ -691,6 +728,7 @@ def deactivate_stale_listings(
|
|||
listing_source: str,
|
||||
ttl_days: int,
|
||||
segments: list[str] | None = None,
|
||||
region_codes: list[int] | None = None,
|
||||
staleness_column: str = "last_seen_at",
|
||||
min_confirmations: int = 0,
|
||||
health_window_days: int = _HEALTH_WINDOW_DAYS,
|
||||
|
|
@ -710,6 +748,24 @@ def deactivate_stale_listings(
|
|||
segments: если задан -- деактивировать только объявления с указанными
|
||||
listing_segment значениями. None -> все сегменты (поведение avito по умолчанию).
|
||||
Несовместимо с null_segment_only=True (см. ниже).
|
||||
region_codes: регионы (listings.region_code), в которых снятие вообще
|
||||
разрешено. None -> DEFAULT_DEACTIVATION_REGION_CODES = (66,), то есть
|
||||
только Свердловская область. ЭТО НЕ КОСМЕТИЧЕСКИЙ ФИЛЬТР: снятие
|
||||
трактует молчание как «объявление ушло», а это верно лишь там, где
|
||||
есть механизм пересбора -- без регулярного обхода «давно не видели»
|
||||
неотличимо от «мы туда не ходили с разовой загрузки» (см. подробный
|
||||
комментарий у DEFAULT_DEACTIVATION_REGION_CODES: что именно случилось
|
||||
бы с корпусом 77/50 и почему вместе с ним встала бы чистка ЕКБ).
|
||||
Расширять список нужно ВМЕСТЕ с запуском регулярного сбора по региону,
|
||||
через ключ "region_codes" в default_params строки расписания. None
|
||||
означает дефолт, а НЕ «без фильтра»: отключить региональное
|
||||
ограничение целиком нельзя намеренно -- только перечислить регионы
|
||||
явно. Строки с region_code IS NULL не снимаются никогда (ANY(...) не
|
||||
матчит NULL): неизвестный регион нечем отнести к покрытому сбором.
|
||||
Гейты здоровья/пола (min_confirmations, revisit_floor_quantile) и
|
||||
active_pool ОСТАЮТСЯ общими по источнику, без региона -- их пороги
|
||||
откалиброваны на источнике целиком, региональная нарезка потребовала
|
||||
бы пересчёта калибровки и здесь не делается.
|
||||
staleness_column: колонка-таймстемп, по которой считается свежесть. Whitelist
|
||||
{"last_seen_at", "scraped_at"} — иначе ValueError ДО любого SQL. Дефолт
|
||||
last_seen_at. Для domklik (#2204) — scraped_at: нетрекаемый bulk-touch
|
||||
|
|
@ -815,7 +871,11 @@ def deactivate_stale_listings(
|
|||
min_floor_pairs < 0 / floor_drop_ratio < 1 / max_deactivated <= 0, ИЛИ
|
||||
любой из этих трёх -- bool (тот же класс jsonb-опечатки true/false
|
||||
вместо числа, что и у ttl_days/cap_mult выше -- default_params
|
||||
расписания это единственный запланированный способ их переопределить).
|
||||
расписания это единственный запланированный способ их переопределить),
|
||||
ЛИБО region_codes -- пустой список (= ANY(ARRAY[]) не матчит ничего,
|
||||
прогон молча снял бы 0 и выглядел бы здоровым) или содержит не-int /
|
||||
bool (та же jsonb-опечатка, что выше: "region_codes": [true] прошло бы
|
||||
как [1] -- регион 1 существует).
|
||||
"""
|
||||
counters: dict[str, int] = {"deactivated": 0}
|
||||
try:
|
||||
|
|
@ -850,9 +910,7 @@ def deactivate_stale_listings(
|
|||
raise ValueError(f"min_floor_pairs must be >= 0, got {min_floor_pairs!r}")
|
||||
|
||||
if isinstance(floor_drop_ratio, bool):
|
||||
raise ValueError(
|
||||
f"floor_drop_ratio must be a number, not bool: {floor_drop_ratio!r}"
|
||||
)
|
||||
raise ValueError(f"floor_drop_ratio must be a number, not bool: {floor_drop_ratio!r}")
|
||||
if floor_drop_ratio < 1:
|
||||
raise ValueError(f"floor_drop_ratio must be >= 1, got {floor_drop_ratio!r}")
|
||||
|
||||
|
|
@ -876,6 +934,21 @@ def deactivate_stale_listings(
|
|||
if null_segment_only and segments is not None:
|
||||
raise ValueError("null_segment_only=True несовместимо с заданным segments")
|
||||
|
||||
# Регионы снятия (см. DEFAULT_DEACTIVATION_REGION_CODES): None -> дефолт,
|
||||
# а НЕ «без фильтра» -- снимать можно только там, где идёт пересбор.
|
||||
# Пустой список отвергаем явно: он матчил бы 0 строк, и прогон выглядел бы
|
||||
# здоровым (deactivated=0) вместо того, чтобы упасть на очевидной ошибке
|
||||
# конфигурации. bool отсекаем до isinstance(int), как у ttl_days/cap_mult
|
||||
# выше: jsonb `true` прошёл бы как регион 1.
|
||||
effective_region_codes: list[int] = (
|
||||
list(DEFAULT_DEACTIVATION_REGION_CODES) if region_codes is None else list(region_codes)
|
||||
)
|
||||
if not effective_region_codes:
|
||||
raise ValueError("region_codes must be a non-empty list, got empty")
|
||||
for code in effective_region_codes:
|
||||
if isinstance(code, bool) or not isinstance(code, int):
|
||||
raise ValueError(f"region_codes must contain ints, got {code!r}")
|
||||
|
||||
# Гейт по здоровью сбора (#2659) — ДО любого UPDATE. Деактивация необратима
|
||||
# на практике (вернуть «живость» может только повторный сбор), поэтому
|
||||
# проверяем ПЕРЕД записью, а не откатываем после.
|
||||
|
|
@ -1060,6 +1133,9 @@ def deactivate_stale_listings(
|
|||
preflight_params: dict[str, Any] = {
|
||||
"listing_source": listing_source,
|
||||
"ttl_days": effective_ttl_days,
|
||||
# Тот же срез регионов, что и у UPDATE ниже -- иначе потолок
|
||||
# считал бы кандидатов по регионам, которых снятие не коснётся.
|
||||
"region_codes": effective_region_codes,
|
||||
}
|
||||
if null_segment_only:
|
||||
candidates_sql = _build_null_segment_candidates_count_sql(staleness_column)
|
||||
|
|
@ -1108,6 +1184,7 @@ def deactivate_stale_listings(
|
|||
params: dict[str, Any] = {
|
||||
"listing_source": listing_source,
|
||||
"ttl_days": effective_ttl_days,
|
||||
"region_codes": effective_region_codes,
|
||||
"run_id": run_id,
|
||||
}
|
||||
result = db.execute(_build_null_segment_sql(staleness_column), params)
|
||||
|
|
@ -1115,6 +1192,7 @@ def deactivate_stale_listings(
|
|||
params = {
|
||||
"listing_source": listing_source,
|
||||
"ttl_days": effective_ttl_days,
|
||||
"region_codes": effective_region_codes,
|
||||
"segments": segments,
|
||||
"run_id": run_id,
|
||||
}
|
||||
|
|
@ -1123,6 +1201,7 @@ def deactivate_stale_listings(
|
|||
params = {
|
||||
"listing_source": listing_source,
|
||||
"ttl_days": effective_ttl_days,
|
||||
"region_codes": effective_region_codes,
|
||||
"run_id": run_id,
|
||||
}
|
||||
result = db.execute(_build_all_segments_sql(staleness_column), params)
|
||||
|
|
@ -1133,14 +1212,15 @@ def deactivate_stale_listings(
|
|||
runs_mod.mark_done(db, run_id, counters)
|
||||
logger.info(
|
||||
"deactivate_stale source=%s run_id=%d done: deactivated=%d "
|
||||
"(ttl_days=%d эффективный, задан %d, segments=%r, null_segment_only=%s, "
|
||||
"staleness_column=%s)",
|
||||
"(ttl_days=%d эффективный, задан %d, segments=%r, region_codes=%r, "
|
||||
"null_segment_only=%s, staleness_column=%s)",
|
||||
listing_source,
|
||||
run_id,
|
||||
counters["deactivated"],
|
||||
effective_ttl_days,
|
||||
ttl_days,
|
||||
segments,
|
||||
effective_region_codes,
|
||||
null_segment_only,
|
||||
staleness_column,
|
||||
)
|
||||
|
|
|
|||
279
tradein-mvp/backend/tests/test_deactivate_stale_region_scope.py
Normal file
279
tradein-mvp/backend/tests/test_deactivate_stale_region_scope.py
Normal file
|
|
@ -0,0 +1,279 @@
|
|||
"""Снятие устаревших ограничено регионами, где есть регулярный пересбор.
|
||||
|
||||
Джоба трактует молчание как «объявление ушло»: строку давно не видели свежей ->
|
||||
is_active=false. Вывод верен ТОЛЬКО там, где есть механизм пересбора. Регионы 77 и
|
||||
50 попали в listings разовой загрузкой 10-12.09.2026, регулярного сбора по ним нет
|
||||
(CITY_ANCHORS кита покрывает только 66, гео-скоуп ДомКлика зашит в ЕКБ) -- без
|
||||
фильтра около 26.09 кандидатами разом стали бы 40 410 строк ДомКлика, и аварийный
|
||||
потолок max_deactivated остановил бы снятие ДЛЯ ВСЕХ регионов, включая ЕКБ
|
||||
(skipped_cap_exceeded не трогает ни одной строки).
|
||||
|
||||
Здесь проверяется: дефолт = только 66; предикат стоит и в preflight count(*), и в
|
||||
UPDATE с ОДНИМ И ТЕМ ЖЕ значением параметра (иначе потолок считает одно, а снимает
|
||||
другое); явный список расширяет охват; мусор в region_codes отвергается.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
from app.tasks import deactivate_stale_avito as task_mod
|
||||
|
||||
_REGION_PREDICATE = "region_code = ANY(CAST(:region_codes AS int[]))"
|
||||
|
||||
|
||||
# ── Фейковая сессия поверх набора строк ──────────────────────────────────────────
|
||||
# В отличие от _FakeDB прочих файлов, эта считает кандидатов и rowcount НЕ из
|
||||
# фикстуры, а применяя предикат к одному и тому же набору строк по параметрам
|
||||
# КАЖДОГО запроса. Поэтому расхождение preflight и UPDATE (разный :region_codes,
|
||||
# потерянный фильтр в одном из двух) проявляется как разные числа, а не прячется
|
||||
# за заранее заданным rowcount.
|
||||
|
||||
|
||||
class _FakeResult:
|
||||
def __init__(self, rowcount: int = 0, scalar_value: Any = None) -> None:
|
||||
self.rowcount = rowcount
|
||||
self._scalar = scalar_value
|
||||
|
||||
def scalar(self) -> Any:
|
||||
return self._scalar
|
||||
|
||||
|
||||
class _RowsDB:
|
||||
"""rows: список dict'ов region_code / stale / active / segment."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
rows: list[dict[str, Any]],
|
||||
*,
|
||||
floor_days: float | None = 0.0,
|
||||
floor_n_pairs: int = 5000,
|
||||
prev_floor_n_pairs: int | None = 5000,
|
||||
confirmations: int = 10_000,
|
||||
) -> None:
|
||||
self._rows = rows
|
||||
self._floor = floor_days
|
||||
self._floor_n_pairs = floor_n_pairs
|
||||
self._prev_floor_n_pairs = prev_floor_n_pairs
|
||||
self._confirmations = confirmations
|
||||
self.executed: list[tuple[str, dict[str, Any] | None]] = []
|
||||
self.committed = False
|
||||
self.rolled_back = False
|
||||
|
||||
def _matching(self, sql: str, params: dict[str, Any] | None) -> int:
|
||||
params = params or {}
|
||||
region_codes = params.get("region_codes")
|
||||
segments = params.get("segments")
|
||||
matched = 0
|
||||
for row in self._rows:
|
||||
if not row.get("active", True):
|
||||
continue
|
||||
if not row.get("stale", True):
|
||||
continue
|
||||
# Фильтр применяется, только если он реально есть в тексте запроса --
|
||||
# выкинутый из одного из двух запросов предикат обязан дать расхождение.
|
||||
if _REGION_PREDICATE in sql:
|
||||
# ANY(...) никогда не матчит NULL -- та же семантика, что в SQL.
|
||||
if row["region_code"] is None or row["region_code"] not in (region_codes or []):
|
||||
continue
|
||||
if "listing_segment IS NULL" in sql and row.get("segment") is not None:
|
||||
continue
|
||||
if "listing_segment = ANY(CAST(:segments AS text[]))" in sql and (
|
||||
row.get("segment") not in (segments or [])
|
||||
):
|
||||
continue
|
||||
matched += 1
|
||||
return matched
|
||||
|
||||
def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
|
||||
sql = str(stmt.text)
|
||||
self.executed.append((sql, params))
|
||||
if "percentile_disc" in sql:
|
||||
return _FakeResult(scalar_value=self._floor)
|
||||
if "FROM scrape_runs prev" in sql:
|
||||
return _FakeResult(scalar_value=self._prev_floor_n_pairs)
|
||||
if "JOIN LATERAL" in sql:
|
||||
return _FakeResult(scalar_value=self._floor_n_pairs)
|
||||
if "health_window_days" in sql:
|
||||
return _FakeResult(scalar_value=self._confirmations)
|
||||
if "SELECT count(*)" in sql and "ttl_days" in sql:
|
||||
return _FakeResult(scalar_value=self._matching(sql, params))
|
||||
if "SELECT count(*)" in sql: # active_pool -- весь активный пул источника
|
||||
return _FakeResult(scalar_value=sum(1 for r in self._rows if r.get("active", True)))
|
||||
return _FakeResult(rowcount=self._matching(sql, params))
|
||||
|
||||
def commit(self) -> None:
|
||||
self.committed = True
|
||||
|
||||
def rollback(self) -> None:
|
||||
self.rolled_back = True
|
||||
|
||||
def _query(self, needle: str) -> tuple[str, dict[str, Any]]:
|
||||
for sql, params in self.executed:
|
||||
if needle in sql and ("SELECT count(*)" not in sql or "ttl_days" in sql):
|
||||
return sql, dict(params or {})
|
||||
return "", {}
|
||||
|
||||
@property
|
||||
def update_query(self) -> tuple[str, dict[str, Any]]:
|
||||
return self._query("UPDATE listings")
|
||||
|
||||
@property
|
||||
def candidates_query(self) -> tuple[str, dict[str, Any]]:
|
||||
for sql, params in self.executed:
|
||||
if sql.lstrip().startswith("SELECT count(*)") and "ttl_days" in sql:
|
||||
return sql, dict(params or {})
|
||||
return "", {}
|
||||
|
||||
|
||||
def _run(db: _RowsDB, monkeypatch: pytest.MonkeyPatch, **kwargs: Any) -> dict[str, int]:
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
|
||||
# Пол включён, чтобы исполнился preflight-блок (он живёт внутри
|
||||
# revisit_floor_quantile > 0) -- именно его согласованность с UPDATE и проверяем.
|
||||
kwargs.setdefault("revisit_floor_quantile", task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE)
|
||||
return task_mod.deactivate_stale_listings(
|
||||
db, # type: ignore[arg-type]
|
||||
1,
|
||||
listing_source=kwargs.pop("listing_source", "domklik"),
|
||||
ttl_days=kwargs.pop("ttl_days", 14),
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
def _corpus() -> list[dict[str, Any]]:
|
||||
"""4 устаревших строки ЕКБ + по 3 строки «залитых руками» 77 и 50."""
|
||||
return (
|
||||
[{"region_code": 66, "stale": True} for _ in range(4)]
|
||||
+ [{"region_code": 77, "stale": True} for _ in range(3)]
|
||||
+ [{"region_code": 50, "stale": True} for _ in range(3)]
|
||||
)
|
||||
|
||||
|
||||
# ── Дефолт: только регион с регулярным сбором ────────────────────────────────────
|
||||
|
||||
|
||||
def test_default_region_scope_is_sverdlovsk_only() -> None:
|
||||
assert task_mod.DEFAULT_DEACTIVATION_REGION_CODES == (66,)
|
||||
|
||||
|
||||
def test_default_run_does_not_touch_rows_of_other_regions(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
db = _RowsDB(_corpus())
|
||||
out = _run(db, monkeypatch)
|
||||
assert out["deactivated"] == 4, "сняты должны быть только 4 строки региона 66"
|
||||
assert db.update_query[1]["region_codes"] == [66]
|
||||
assert db.committed is True
|
||||
|
||||
|
||||
def test_rows_with_null_region_code_are_never_deactivated(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""ANY(...) не матчит NULL -- неизвестный регион нечем отнести к покрытому сбором."""
|
||||
db = _RowsDB([{"region_code": 66, "stale": True}, {"region_code": None, "stale": True}])
|
||||
out = _run(db, monkeypatch)
|
||||
assert out["deactivated"] == 1
|
||||
|
||||
|
||||
# ── Preflight и UPDATE согласованы ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_preflight_counts_exactly_what_update_deactivates(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
db = _RowsDB(_corpus())
|
||||
out = _run(db, monkeypatch)
|
||||
assert out["deactivation_candidates"] == out["deactivated"] == 4
|
||||
|
||||
|
||||
def test_preflight_and_update_bind_the_same_region_codes(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
db = _RowsDB(_corpus())
|
||||
_run(db, monkeypatch, region_codes=[66, 77])
|
||||
assert db.candidates_query[1]["region_codes"] == db.update_query[1]["region_codes"] == [66, 77]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"build_update, build_count",
|
||||
[
|
||||
(task_mod._build_all_segments_sql, task_mod._build_all_segments_candidates_count_sql),
|
||||
(task_mod._build_segments_sql, task_mod._build_segments_candidates_count_sql),
|
||||
(task_mod._build_null_segment_sql, task_mod._build_null_segment_candidates_count_sql),
|
||||
],
|
||||
)
|
||||
def test_region_predicate_present_in_both_update_and_count_sql(
|
||||
build_update: Any, build_count: Any
|
||||
) -> None:
|
||||
"""Предикат продублирован текстуально в обеих ветках -- потерять его в одной
|
||||
значит считать кандидатов по одному срезу, а снимать по другому."""
|
||||
for sql in (str(build_update("scraped_at").text), str(build_count("scraped_at").text)):
|
||||
assert _REGION_PREDICATE in sql
|
||||
# psycopg v3: никакого :param::type
|
||||
assert ":region_codes::" not in sql
|
||||
|
||||
|
||||
def test_cap_counts_only_covered_regions(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Ради чего всё: чужой регион не должен пробивать аварийный потолок и глушить
|
||||
снятие по своему (skipped_cap_exceeded не трогает НИ ОДНОЙ строки)."""
|
||||
rows = [{"region_code": 66, "stale": True} for _ in range(3)]
|
||||
rows += [{"region_code": 77, "stale": True} for _ in range(100)]
|
||||
db = _RowsDB(rows)
|
||||
out = _run(db, monkeypatch, max_deactivated=10)
|
||||
assert "skipped_cap_exceeded" not in out
|
||||
assert out["deactivated"] == 3
|
||||
|
||||
|
||||
# ── Явный список расширяет охват ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_explicit_region_codes_widen_the_scope(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
db = _RowsDB(_corpus())
|
||||
out = _run(db, monkeypatch, region_codes=[66, 77])
|
||||
assert out["deactivated"] == 7
|
||||
assert out["deactivation_candidates"] == 7
|
||||
|
||||
|
||||
def test_explicit_region_codes_can_target_a_single_foreign_region(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
db = _RowsDB(_corpus())
|
||||
out = _run(db, monkeypatch, region_codes=[50])
|
||||
assert out["deactivated"] == 3
|
||||
|
||||
|
||||
# ── Валидация (jsonb-опечатки в default_params) ──────────────────────────────────
|
||||
|
||||
|
||||
def test_empty_region_codes_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""[] матчил бы 0 строк и выглядел бы здоровым прогоном -- явный ValueError."""
|
||||
db = _RowsDB(_corpus())
|
||||
with pytest.raises(ValueError, match="region_codes"):
|
||||
_run(db, monkeypatch, region_codes=[])
|
||||
|
||||
|
||||
def test_bool_region_code_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""jsonb `true` прошёл бы как регион 1 -- тот же класс опечатки, что ttl_days."""
|
||||
db = _RowsDB(_corpus())
|
||||
with pytest.raises(ValueError, match="region_codes"):
|
||||
_run(db, monkeypatch, region_codes=[True])
|
||||
|
||||
|
||||
def test_non_int_region_code_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
db = _RowsDB(_corpus())
|
||||
with pytest.raises(ValueError, match="region_codes"):
|
||||
_run(db, monkeypatch, region_codes=["66"]) # type: ignore[list-item]
|
||||
|
||||
|
||||
def test_invalid_region_codes_touch_no_rows(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
db = _RowsDB(_corpus())
|
||||
with pytest.raises(ValueError):
|
||||
_run(db, monkeypatch, region_codes=[])
|
||||
assert db.update_query[0] == ""
|
||||
assert db.committed is False
|
||||
Loading…
Add table
Reference in a new issue