From a8add671012d8cdae9d3ac5f7bea0b023798f9d6 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 15 Sep 2026 10:13:12 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein):=20=D1=81=D0=BD=D0=B8=D0=BC=D0=B0?= =?UTF-8?q?=D1=82=D1=8C=20=D1=83=D1=81=D1=82=D0=B0=D1=80=D0=B5=D0=B2=D1=88?= =?UTF-8?q?=D0=B8=D0=B5=20=D0=BE=D0=B1=D1=8A=D1=8F=D0=B2=D0=BB=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D1=8F=20=D1=82=D0=BE=D0=BB=D1=8C=D0=BA=D0=BE=20=D0=B2=20?= =?UTF-8?q?=D1=80=D0=B5=D0=B3=D0=B8=D0=BE=D0=BD=D0=B0=D1=85=20=D1=81=20?= =?UTF-8?q?=D0=BF=D0=B5=D1=80=D0=B5=D1=81=D0=B1=D0=BE=D1=80=D0=BE=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Джоба деактивации трактует молчание как «объявление ушло», что верно только там, где есть регулярный обход. Регионы 77 и 50 залиты разово 10-12.09 и не обновляются: около 26.09 кандидатами стали бы 40 410 строк ДомКлика, аварийный потолок max_deactivated сработал бы и остановил снятие ДЛЯ ВСЕХ регионов, включая ЕКБ (skipped_cap_exceeded не трогает ни одной строки). Позже то же с cian и yandex. Параметр region_codes (default_params расписания), дефолт (66,). Предикат добавлен и в preflight count(*), и в UPDATE во всех трёх сегментных ветках — потолок считает ровно то, что снимает. --- .../backend/app/services/product_handlers.py | 7 + .../app/tasks/deactivate_stale_avito.py | 100 ++++++- .../test_deactivate_stale_region_scope.py | 279 ++++++++++++++++++ 3 files changed, 376 insertions(+), 10 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_deactivate_stale_region_scope.py diff --git a/tradein-mvp/backend/app/services/product_handlers.py b/tradein-mvp/backend/app/services/product_handlers.py index 9d8fee48..da13dfb2 100644 --- a/tradein-mvp/backend/app/services/product_handlers.py +++ b/tradein-mvp/backend/app/services/product_handlers.py @@ -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, diff --git a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py index 38834eb8..bc0c35d8 100644 --- a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py +++ b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py @@ -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, ) diff --git a/tradein-mvp/backend/tests/test_deactivate_stale_region_scope.py b/tradein-mvp/backend/tests/test_deactivate_stale_region_scope.py new file mode 100644 index 00000000..b67936f6 --- /dev/null +++ b/tradein-mvp/backend/tests/test_deactivate_stale_region_scope.py @@ -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