"""Снятие устаревших ограничено регионами, где есть регулярный пересбор. Джоба трактует молчание как «объявление ушло»: строку давно не видели свежей -> 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