gendesign/tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py

"""Пол 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
    # = '<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