"""Пол TTL по измеренному циклу переобхода (#2659). Гейт здоровья (#2710) отвечает «источник собирается?». Этот пол отвечает на второй вопрос issue — «а достаточно ли ttl_days, чтобы молчание означало снятие?». Пока свип возвращается к строке реже, чем раз в ttl_days, TTL меряет нашу выборку, а не жизнь объявления, и источник при этом ЗДОРОВ — гейт молчит. Ключевой тест — test_effective_ttl_covers_every_proven_false_kill: он проигрывает РЕАЛЬНЫЙ прод-случай (127 строк, снятых на возрасте 29.9..30.3 суток при TTL=30 и доказанно вернувшихся живыми) и требует, чтобы эффективный TTL накрыл каждую. На старом коде — без пола — эффективный TTL остаётся 30, и тест падает на всех срезах. """ from __future__ import annotations 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.tasks import deactivate_stale_avito as task_mod # ── Прод-замер 2026-08-09 (read-only), из-за которого пол существует ─────────── # С деплоя гейта 06.08 TTL снял 1 028 строк, 127 уже снова активны (12.4%). # В Екатеринбурге — единственном городе с настоящим покрытием — доля ложных 100%. _FALSE_KILLS_BY_CITY: dict[str, tuple[int, int]] = { # срез -> (снято, снова активны) "cian/Екатеринбург": (103, 103), "yandex/Екатеринбург": (24, 24), "cian/без города": (560, 0), "yandex/без города": (341, 0), } # Возраст строки на момент снятия у всех 127 доказанно ложных снятий, суток. _FALSE_KILL_AGE_MIN = 29.9 _FALSE_KILL_AGE_MAX = 30.3 # Пол, который отдаёт ТОТ ЖЕ запрос на проде (percentile_disc 0.99, окно 3 суток, # срез = срез TTL-джобы). Ключ -> (listing_source, segments, ttl_days сейчас, пол). _PROD_FLOORS: dict[str, tuple[str, list[str] | None, int, float]] = { "cian/vtorichka": ("cian", ["vtorichka"], 30, 34.0), "yandex/vtorichka": ("yandex", ["vtorichka"], 30, 74.3), "avito/все сегменты": ("avito", None, 10, 69.7), } # Разрывы переобхода по срезу TTL-джобы (listing_source_snapshots, 40 суток): # источник -> (p90, p99, ttl_days сейчас). Домклик — контрольная группа: почти # полное суточное покрытие, TTL лежит в 4.5 раза выше хвоста, и снятие у него # действительно означает снятие. _REVISIT_TAIL: dict[str, tuple[float, float, int]] = { "domklik/vtorichka": (1.9, 3.1, 14), "cian/vtorichka": (10.9, 26.6, 30), "yandex/vtorichka": (5.7, 43.0, 30), "avito/vtorichka": (29.1, 42.1, 10), } # ── Фейковая сессия ─────────────────────────────────────────────────────────── 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 _FakeDB: """Session-заглушка: percentile_disc -> пол, count(*) -> подтверждения, UPDATE -> rowcount.""" def __init__( self, *, floor_days: float | None, confirmations: int = 10_000, rowcount: int = 137, ) -> None: self._floor = floor_days self._confirmations = confirmations self._rowcount = rowcount self.executed: list[tuple[str, dict[str, Any] | None]] = [] self.committed = False self.rolled_back = False 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 "SELECT count(*)" in sql: return _FakeResult(scalar_value=self._confirmations) return _FakeResult(rowcount=self._rowcount) def commit(self) -> None: self.committed = True def rollback(self) -> None: self.rolled_back = True @property def floor_query(self) -> tuple[str, dict[str, Any] | None]: return next((e for e in self.executed if "percentile_disc" in e[0]), ("", None)) @property def update_query(self) -> tuple[str, dict[str, Any] | None]: return next((e for e in self.executed if "UPDATE listings" in e[0]), ("", None)) def _run(db: _FakeDB, 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) return task_mod.deactivate_stale_listings( db, # type: ignore[arg-type] 1, listing_source=kwargs.pop("listing_source", "cian"), ttl_days=kwargs.pop("ttl_days", 30), **kwargs, ) # ── Исторический случай: 127 доказанных ложных снятий ───────────────────────── def test_effective_ttl_covers_every_proven_false_kill(monkeypatch: pytest.MonkeyPatch) -> None: """Ни одно из 127 доказанно ложных снятий (cian/yandex) не должно повториться. Все они произошли на возрасте 29.9..30.3 суток. Эффективный TTL обязан быть строго выше этого возраста на cian/yandex-срезах — иначе следующий прогон снимет ту же строку снова. avito из этого цикла исключён намеренно: 127 доказанных ложных снятий (_FALSE_KILLS_BY_CITY) измерены только по cian/yandex, у avito другой сценарий и своя проверка ниже (test_avito_prod_floor_is_capped_by_calibrated_cap_mult) -- калибровка cap_mult=6 для avito (миграция 264_deactivate_stale_avito_cap_mult.sql) пиннится ТАМ, а не здесь, чтобы не смешивать два разных замера под одним порогом _FALSE_KILL_AGE_MAX, который к avito не относится. """ for slice_name, (source, segments, ttl_days, floor) in _PROD_FLOORS.items(): if source == "avito": continue db = _FakeDB(floor_days=floor) out = _run( db, monkeypatch, listing_source=source, ttl_days=ttl_days, segments=segments, revisit_floor_quantile=task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE, ) effective = out["ttl_days_effective"] assert effective > _FALSE_KILL_AGE_MAX, ( f"{slice_name}: эффективный TTL {effective} не накрывает возраст ложного " f"снятия {_FALSE_KILL_AGE_MAX} — 127 строк снимутся снова" ) _, update_params = db.update_query assert update_params is not None assert update_params["ttl_days"] == effective, ( f"{slice_name}: UPDATE получил не поднятый TTL — пол посчитан и выброшен" ) def _read_cap_mult_from_migration(filename: str, *, source: str) -> int: """Читает cap_mult из UPDATE default_params миграции -- НЕ хардкодит дубль в тесте. Найдено ревью круга 3 2026-08-15: раньше тест ниже принимал cap_mult=6 как аргумент напрямую, захардкоженный прямо в теле теста. Мутация значения в 264_deactivate_stale_avito_cap_mult.sql (6 -> 2) НЕ трогала вход теста вовсе -- набор оставался зелёным при любом реальном значении в миграции, то есть калибровка нигде не была пином, только упоминанием в комментарии. Здесь значение читается ИЗ ФАЙЛА миграции regex'ом, а ожидаемый результат (ttl_days_effective, ttl_floor_capped) остаётся зафиксированным числом в самом тесте -- так дрейф калибровки в миграции ломает тест, как и задумано. """ migration = Path(__file__).resolve().parents[1] / "data" / "sql" / filename src = migration.read_text("utf-8") # Порядок в файле -- jsonb_build_object('cap_mult', N) в SET, ЗАТЕМ WHERE source # = '' ниже (см. 264/265_*.sql). DOTALL матчит перевод строки между ними; # source в regex -- страховка от чтения не того UPDATE, если файл когда-нибудь # станет мульти-source (сейчас в каждом файле ровно один UPDATE). match = re.search( r"jsonb_build_object\('cap_mult',\s*(\d+)\).*?WHERE\s+source\s*=\s*'" + re.escape(source) + r"'", src, re.DOTALL, ) assert match is not None, ( f"{filename} сменил формат UPDATE default_params для source={source!r} -- " "обнови regex в _read_cap_mult_from_migration" ) return int(match.group(1)) def test_avito_prod_floor_is_capped_by_calibrated_cap_mult(monkeypatch: pytest.MonkeyPatch) -> None: """Пиннит калибровку cap_mult=6 для avito (миграция 264_deactivate_stale_avito_cap_mult.sql) на измеренном прод-поле _PROD_FLOORS ("avito/все сегменты" = 69.7, замер 2026-08-09). cap_mult -- ВХОД теста, читается ИЗ ФАЙЛА миграции (regex), не хардкодится здесь: дрейф калибровки в 264_*.sql (например 6 -> 2) меняет вход, но НЕ ожидаемый результат ниже (60/70) -- эти числа пинят калибровку саму по себе, поэтому дрейф ломает тест, как и задумано (см. _read_cap_mult_from_migration). С дефолтным cap_mult=2 потолок avito (20 сут) РЕЖЕТ ниже собственного хвоста переобхода p99=42.1 (_REVISIT_TAIL) -- ровно тот false-kill, ради которого пол заведён. С калиброванным cap_mult=6 потолок 60 сут -- выше и p99=42.1, и живого прод-пика 52 (замер 08-10..08-12), и этого гипотетического замера 69.7 (капается ровно на 60, не пропускается как есть). """ calibrated_cap_mult = _read_cap_mult_from_migration( "264_deactivate_stale_avito_cap_mult.sql", source="deactivate_stale_avito" ) source, segments, ttl_days, floor = _PROD_FLOORS["avito/все сегменты"] db = _FakeDB(floor_days=floor) out = _run( db, monkeypatch, listing_source=source, ttl_days=ttl_days, segments=segments, revisit_floor_quantile=task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE, cap_mult=calibrated_cap_mult, ) assert out["ttl_days_effective"] == 60, "cap_mult из миграции 264 обязан дать потолок 60" assert out["ttl_floor_capped"] == 1 assert out["ttl_days_floor_raw"] == 70, "ceil(69.7) == 70 -- пол считается по real-числу" def test_yandex_prod_floor_is_not_capped_by_calibrated_cap_mult( monkeypatch: pytest.MonkeyPatch, ) -> None: """Пиннит калибровку cap_mult=3 для yandex (миграция 265_deactivate_stale_yandex_cap_mult.sql, найдено ревью круга 3 2026-08-15) на измеренном прод-поле _PROD_FLOORS ("yandex/vtorichka" = 74.3). cap_mult -- ВХОД теста, читается ИЗ ФАЙЛА миграции 265 (тот же приём, что и у avito выше): дрейф калибровки в 265_*.sql ломает тест. С дефолтным cap_mult=2 потолок yandex (60 сут) РЕЖЕТ живой пол (75-79 сут, scrape_runs.counters 08-10..08-15 и live-замер 08-15) -- та же дыра, что у avito, найдена позже (первая версия 264 ошибочно считала yandex безопасным по устаревшему статическому p99=43.0). С калиброванным cap_mult=3 потолок 90 сут выше живого пика 79.2 -- пол 74.3 из этого теста НЕ капается, эффективный TTL равен сырому полу (75, ceil(74.3)). """ calibrated_cap_mult = _read_cap_mult_from_migration( "265_deactivate_stale_yandex_cap_mult.sql", source="deactivate_stale_yandex" ) source, segments, ttl_days, floor = _PROD_FLOORS["yandex/vtorichka"] db = _FakeDB(floor_days=floor) out = _run( db, monkeypatch, listing_source=source, ttl_days=ttl_days, segments=segments, revisit_floor_quantile=task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE, cap_mult=calibrated_cap_mult, ) assert out["ttl_days_effective"] == 75, "ceil(74.3) == 75, потолок 90 не должен резать" assert "ttl_floor_capped" not in out, "потолок 90 выше живого пола 74.3 -- капать нечего" def test_false_kill_ages_sit_inside_the_old_ttl(monkeypatch: pytest.MonkeyPatch) -> None: """Замер согласован сам с собой: снимали ровно на границе TTL=30, не раньше.""" assert _FALSE_KILL_AGE_MIN < 30.0 <= _FALSE_KILL_AGE_MAX for source, _segments, ttl_days, _floor in _PROD_FLOORS.values(): if source in ("cian", "yandex"): assert ttl_days == 30, f"{source}: прод-TTL разошёлся с замером" def test_false_kill_rate_is_total_where_coverage_is_real() -> None: """В городе с настоящим покрытием ложны ВСЕ снятия — это и есть корень #2659.""" for slice_name in ("cian/Екатеринбург", "yandex/Екатеринбург"): killed, returned = _FALSE_KILLS_BY_CITY[slice_name] assert returned == killed, f"{slice_name}: замер разошёлся" total_killed = sum(k for k, _ in _FALSE_KILLS_BY_CITY.values()) total_returned = sum(r for _, r in _FALSE_KILLS_BY_CITY.values()) assert total_killed == 1028 assert total_returned == 127 def test_control_group_domklik_needs_no_floor() -> None: """Домклик — контроль: при почти полном суточном обходе TTL с запасом выше хвоста. Если бы пол был нужен всем подряд, он был бы нужен и источнику со сплошным покрытием. Ему не нужен — значит меряем именно покрытие, а не «TTL маловат». Разделяет источники ЗАПАС над хвостом, а не сам факт превышения: у домклика TTL/p99 = 4.5, у остальных трёх 0.24..1.13, то есть порог сидит вплотную к хвосту или внутри него. Именно у самого «благополучного» из трёх (cian, 1.13) и случились 103 доказанно ложных снятия: p99 запаса не даёт, снимает 1% живых. """ _p90, p99, ttl = _REVISIT_TAIL["domklik/vtorichka"] assert ttl / p99 > 4, "домклик перестал быть контрольной группой — перемерить" for name, (_p90, p99, ttl) in _REVISIT_TAIL.items(): if name.startswith("domklik"): continue assert ttl / p99 < 1.5, f"{name}: TTL отошёл от хвоста обхода, замер устарел" # ── Контракт пола ───────────────────────────────────────────────────────────── def test_floor_never_lowers_configured_ttl(monkeypatch: pytest.MonkeyPatch) -> None: """Пол поднимает порог и только. Здоровый источник с коротким циклом не даёт поводу снять больше, чем задано оператором.""" db = _FakeDB(floor_days=3.1) out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99) assert out["ttl_days_effective"] == 30 _, update_params = db.update_query assert update_params is not None assert update_params["ttl_days"] == 30 def test_floor_is_rounded_up_not_down(monkeypatch: pytest.MonkeyPatch) -> None: """34.0 суток разрыва -> TTL 34, а 34.2 -> 35: округление в сторону осторожности.""" db = _FakeDB(floor_days=34.2) out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99) assert out["revisit_floor_days"] == 35 assert out["ttl_days_effective"] == 35 def test_floor_disabled_keeps_old_behaviour(monkeypatch: pytest.MonkeyPatch) -> None: """quantile=0 -> ни одного лишнего запроса, поведение как до правки.""" db = _FakeDB(floor_days=74.3) out = _run(db, monkeypatch, ttl_days=30) assert out == {"deactivated": 137} assert len(db.executed) == 1 assert "percentile_disc" not in db.executed[0][0] def test_missing_snapshot_history_leaves_ttl_as_configured( monkeypatch: pytest.MonkeyPatch, ) -> None: """NULL (истории снимков за окно нет) -> пол не выдумывается, TTL как задан.""" db = _FakeDB(floor_days=None) out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99) assert "revisit_floor_days" not in out _, update_params = db.update_query assert update_params is not None assert update_params["ttl_days"] == 30 def test_floor_runs_before_any_write(monkeypatch: pytest.MonkeyPatch) -> None: """Пол считается ДО UPDATE: снятое объявление возвращает только повторный сбор.""" db = _FakeDB(floor_days=74.3) _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99) kinds = ["floor" if "percentile_disc" in sql else "update" for sql, _ in db.executed] assert kinds.index("floor") < kinds.index("update") def test_floor_measures_same_slice_as_update(monkeypatch: pytest.MonkeyPatch) -> None: """Срез пола совпадает со срезом UPDATE: тот же source и те же сегменты.""" db = _FakeDB(floor_days=74.3) _run( db, monkeypatch, listing_source="yandex", segments=["vtorichka"], revisit_floor_quantile=0.99, ) floor_sql, floor_params = db.floor_query assert "ANY(CAST(:segments AS text[]))" in floor_sql assert floor_params is not None assert floor_params["segments"] == ["vtorichka"] assert floor_params["listing_source"] == "yandex" def test_floor_uses_same_staleness_column_as_ttl(monkeypatch: pytest.MonkeyPatch) -> None: """domklik считает свежесть по scraped_at (#2204) — пол обязан мерить ту же колонку, иначе bulk-touch по last_seen_at показал бы цикл обхода там, где сбора нет.""" db = _FakeDB(floor_days=20.0) _run( db, monkeypatch, listing_source="domklik", ttl_days=14, staleness_column="scraped_at", revisit_floor_quantile=0.99, ) floor_sql, _ = db.floor_query assert "l.scraped_at" in floor_sql assert "l.last_seen_at" not in floor_sql def test_floor_rejects_invalid_staleness_column(monkeypatch: pytest.MonkeyPatch) -> None: """Whitelist колонки работает и на пути пола — интерполяции чужого имени нет.""" db = _FakeDB(floor_days=20.0) with pytest.raises(ValueError): _run(db, monkeypatch, staleness_column="is_active", revisit_floor_quantile=0.99) assert db.executed == [] def test_health_gate_still_wins_over_floor(monkeypatch: pytest.MonkeyPatch) -> None: """Нездоровый источник блокируется гейтом ДО того, как считается пол: лишний тяжёлый запрос по мёртвому источнику не нужен.""" db = _FakeDB(floor_days=74.3, confirmations=10) out = _run( db, monkeypatch, min_confirmations=500, revisit_floor_quantile=0.99, ) assert out["skipped_unhealthy"] == 1 assert db.floor_query[0] == "" assert db.update_query[0] == "" def test_floor_sql_is_psycopg_v3_safe() -> None: sql = str(task_mod._build_revisit_floor_sql("last_seen_at", with_segments=True).text) assert "CAST(:revisit_quantile AS double precision)" in sql assert "CAST(:health_window_days AS integer)" in sql assert not re.search(r":\w+::", sql) assert "UPDATE" not in sql.upper() assert "DELETE" not in sql.upper() def test_floor_only_counts_rows_the_sweep_actually_refound() -> None: """Пол меряет выживших, а не «мы к ним не приходили»: свежесть обязана СДВИНУТЬСЯ относительно предыдущего снимка, иначе разрыв нулевой и хвост схлопнется в ноль.""" sql = str(task_mod._build_revisit_floor_sql("last_seen_at", with_segments=False).text) assert "l.last_seen_at > prev.last_seen_at" in sql def test_default_quantile_is_high_enough_for_the_prod_case() -> None: """Ниже 0.99 опускать нельзя без нового замера: именно на 0.99 прод-запрос даёт 34.0 для cian/vtorichka, что накрывает возраст ложных снятий 30.3.""" assert task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE >= 0.99 assert _PROD_FLOORS["cian/vtorichka"][3] > _FALSE_KILL_AGE_MAX def test_handler_wires_revisit_floor_from_schedule_params() -> None: """Читаем исходник файлом: product_handlers тянет scraper_kit, которого в юнит-окружении может не быть, а проверяем мы проводку, а не импорт.""" handlers = Path(__file__).resolve().parents[1] / "app" / "services" / "product_handlers.py" src = handlers.read_text("utf-8") job = src.split("async def _job_deactivate_stale")[1].split("\nasync def ")[0] flat = " ".join(job.split()) assert 'params.get( "revisit_floor_quantile", DEFAULT_REVISIT_FLOOR_QUANTILE )' in flat assert "revisit_floor_quantile=revisit_floor_quantile" in job