"""Потолок эффективного TTL деактивации (найдено на проде 2026-08-15).

Пол TTL по измеренному циклу переобхода (#2659, deactivate_stale_avito.py) поднимает
эффективный TTL через max(ttl_days, пол) без верхней границы. На проде это оказалось
петлёй с положительной обратной связью: медленный обход поднимает пол, высокий пол
продлевает жизнь снятым лотам дольше, чем к ним успевает вернуться свежий обход, пул
«активных» раздувается протухшими строками. У yandex ttl_days_effective держали
75/75/75/39/52/54 шесть прогонов подряд при deactivated=0 -- это и есть разгон пола,
ради которого потолок написан. Цифру «23 687 из 44 744» из исходного разбора сюда НЕ
переносим: она про все источники сразу, две трети её -- новостройки вне выборки
оценщика, а у самого avito просроченных строк ноль (уточнено 2026-08-15).

Этот файл проверяет CAP_MULT -- потолок, не пускающий эффективный TTL выше
ttl_days * CAP_MULT, независимо от того, насколько высоко посчитанный пол.
"""

from __future__ import annotations

import os
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

# ── Фейковая сессия (тот же контракт, что в test_deactivate_stale_revisit_floor.py) ──


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 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", "avito"),
        ttl_days=kwargs.pop("ttl_days", 10),
        **kwargs,
    )


# ── Контракт из задачи ─────────────────────────────────────────────────────────


def test_high_floor_is_capped_at_double_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
    """revisit_floor=75, ttl_days=10 -> итог 20 (потолок 2x), НЕ 75."""
    db = _FakeDB(floor_days=75.0)
    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 20
    assert out["ttl_floor_capped"] == 1
    assert out["ttl_days_floor_raw"] == 75
    _, update_params = db.update_query
    assert update_params is not None
    assert update_params["ttl_days"] == 20, "UPDATE обязан получить капнутый TTL, не сырой пол"


def test_low_floor_leaves_ttl_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
    """revisit_floor=5, ttl_days=10 -> итог 10 (пол ниже заданного TTL, max() его не поднимает)."""
    db = _FakeDB(floor_days=5.0)
    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 10
    assert "ttl_floor_capped" not in out
    assert "ttl_days_floor_raw" not in out
    _, update_params = db.update_query
    assert update_params is not None
    assert update_params["ttl_days"] == 10


# ── Контракт потолка ────────────────────────────────────────────────────────────


def test_cap_mult_is_named_module_constant_equal_two() -> None:
    assert task_mod.CAP_MULT == 2


def test_floor_between_ttl_and_cap_is_not_flagged_capped(monkeypatch: pytest.MonkeyPatch) -> None:
    """Пол поднял TTL, но не дотянулся до потолка -- capped-флаг НЕ выставляется."""
    db = _FakeDB(floor_days=15.0)
    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 15
    assert "ttl_floor_capped" not in out


def test_floor_exactly_at_cap_boundary_is_not_flagged_capped(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Пол ровно на потолке (2x ttl) -- это ещё "поднят до потолка", не "срезан выше него".

    Формула -- min(raw, cap): при raw == cap срезания не происходит (raw > cap ложно),
    капнутый флаг предназначен сигналить именно "потолок реально что-то отрезал".
    """
    db = _FakeDB(floor_days=20.0)
    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 20
    assert "ttl_floor_capped" not in out


def test_cap_logs_warning_containing_both_numbers(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    """WARNING при срезании содержит и сырой пол, и капнутый результат -- не только counters."""
    db = _FakeDB(floor_days=75.0)
    with caplog.at_level("WARNING", logger=task_mod.logger.name):
        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
    messages = " ".join(r.getMessage() for r in caplog.records)
    assert "75" in messages, "лог обязан называть сырой пол"
    assert "20" in messages, "лог обязан называть итоговый (капнутый) TTL"


def test_cap_never_lowers_ttl_below_configured_value(monkeypatch: pytest.MonkeyPatch) -> None:
    """Потолок -- верхняя граница, не альтернативный источник истины: заданный TTL
    (10) остаётся нижней границей независимо от того, насколько низко ушёл пол."""
    db = _FakeDB(floor_days=1.0)
    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 10


# ── avito self-descend (52 -> ... -> 10) не должен ломаться потолком ────────────


def test_avito_high_transient_floor_is_capped_not_left_unbounded(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Наблюдённый на проде транзиентный пик avito (счётчики видели ttl_days_effective=52)
    теперь капается на 2x ttl=20, а не пропускается в UPDATE как есть."""
    db = _FakeDB(floor_days=52.0)
    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 20
    assert out["ttl_floor_capped"] == 1
    assert out["ttl_days_floor_raw"] == 52


def test_avito_recovered_low_floor_still_reaches_configured_ttl(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """После восстановления обхода (пол опустился ниже ttl_days=10, как на проде 52->10)
    потолок не мешает нормальному пути -- эффективный TTL просто равен заданному."""
    db = _FakeDB(floor_days=9.0)
    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 10
    assert "ttl_floor_capped" not in out


def test_avito_floor_above_ttl_but_under_cap_passes_through_uncapped(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Промежуточная точка того же самопонижения (пол между ttl и потолком, например 18)
    поднимает TTL как раньше -- потолок не мешает нормальному постепенному пути."""
    db = _FakeDB(floor_days=18.0)
    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 18
    assert "ttl_floor_capped" not in out


# ── cap_mult конфигурируем per-source (найдено ревью 2026-08-15) ────────────────
# Дефолтный CAP_MULT=2 даёт разный АБСОЛЮТНЫЙ потолок на разных источниках
# (cian/yandex 60 сут, avito 20 сут), а хвост переобхода не пропорционален
# ttl_days: avito p99=42.1 -- выше его же дефолтного потолка 20. cap_mult -- ручка
# для конкретно такого источника, без изменения дефолта для остальных.


def test_cap_mult_defaults_to_module_constant_when_not_overridden(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Без явного cap_mult поведение не меняется: потолок = ttl_days * CAP_MULT (2)."""
    db = _FakeDB(floor_days=75.0)
    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
    assert out["ttl_days_effective"] == 10 * task_mod.CAP_MULT


def test_cap_mult_override_raises_the_ceiling_for_a_long_tailed_source(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """avito p99=42.1: cap_mult=6 (потолок 60) больше не режет пол ниже хвоста обхода,
    в отличие от дефолтного cap_mult=2 (потолок 20)."""
    db = _FakeDB(floor_days=45.0)
    out = _run(
        db,
        monkeypatch,
        listing_source="avito",
        ttl_days=10,
        revisit_floor_quantile=0.99,
        cap_mult=6,
    )
    assert out["ttl_days_effective"] == 45
    assert "ttl_floor_capped" not in out


def test_cap_mult_override_still_caps_when_floor_exceeds_the_wider_ceiling(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """cap_mult поднимает потолок, но не убирает его -- пол выше 60 всё равно срезается."""
    db = _FakeDB(floor_days=90.0)
    out = _run(
        db,
        monkeypatch,
        listing_source="avito",
        ttl_days=10,
        revisit_floor_quantile=0.99,
        cap_mult=6,
    )
    assert out["ttl_days_effective"] == 60
    assert out["ttl_floor_capped"] == 1
    assert out["ttl_days_floor_raw"] == 90


def test_cap_mult_is_threaded_into_update_params(monkeypatch: pytest.MonkeyPatch) -> None:
    """Капнутый по override'нутому потолку TTL реально уходит в UPDATE, не только считается."""
    db = _FakeDB(floor_days=90.0)
    _run(
        db,
        monkeypatch,
        listing_source="avito",
        ttl_days=10,
        revisit_floor_quantile=0.99,
        cap_mult=6,
    )
    _, update_params = db.update_query
    assert update_params is not None
    assert update_params["ttl_days"] == 60


# ── ttl_days <= 0 (LOW из ревью 2026-08-15) ──────────────────────────────────────
# До потолка max(ttl_days, floor) прикрывал ttl_days<=0, если пол посчитан и
# положителен. С потолком min(raw, ttl_days * cap_mult) при ttl_days<=0 капнутый
# потолок тоже <= 0 и побеждает в min() -- защита пола пропадает молча. Явный guard
# ловит это ДО любого SQL, тем же путём, что и невалидный staleness_column.


def test_ttl_days_zero_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    db = _FakeDB(floor_days=75.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=0, revisit_floor_quantile=0.99)
    assert db.executed == []


def test_ttl_days_negative_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    db = _FakeDB(floor_days=75.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=-5, revisit_floor_quantile=0.99)
    assert db.executed == []


def test_ttl_days_zero_fails_the_run_via_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
    """Тот же контракт, что и невалидный staleness_column: run помечается failed,
    а не остаётся 'running'."""
    marked_failed: list[Any] = []
    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(
        task_mod.runs_mod,
        "mark_failed",
        lambda db, run_id, err, counters: marked_failed.append((run_id, err, counters)),
    )
    db = _FakeDB(floor_days=75.0)
    with pytest.raises(ValueError):
        task_mod.deactivate_stale_listings(
            db,  # type: ignore[arg-type]
            7,
            listing_source="avito",
            ttl_days=0,
        )
    assert len(marked_failed) == 1
    assert marked_failed[0][0] == 7


# ── cap_mult < 1 (HIGH из ревью круга 2, 2026-08-15) ─────────────────────────────
# Тот же класс дыры, что и ttl_days<=0 выше, но со стороны потолка: cap_mult -- ЕДИНСТВЕННЫЙ
# запланированный способ его задать -- руками вписать в jsonb default_params расписания
# (см. миграцию для avito), т.е. именно там опечатка 0 / 0.5 вместо 6 доходит до прода.
# cap_mult=0 -> capped=0 -> effective_ttl_days=0 -> UPDATE снимает весь активный пул
# источника молча. cap_mult<1 (например 0.5) опускает потолок НИЖЕ заданного оператором
# ttl_days -- прямое нарушение инварианта, который проверяет
# test_cap_never_lowers_ttl_below_configured_value для пола, но не было проверено для
# потолка при некорректном cap_mult.


def test_cap_mult_zero_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=0)
    assert db.executed == []


def test_cap_mult_negative_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=-2)
    assert db.executed == []


def test_cap_mult_below_one_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    """cap_mult=0.5 опустил бы потолок НИЖЕ заданного ttl_days -- та самая инверсия,
    которую тест test_cap_never_lowers_ttl_below_configured_value гарантирует для пола."""
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=0.5)
    assert db.executed == []


def test_cap_mult_non_numeric_fails_safe_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    """Опечатка в jsonb default_params (строка вместо числа) не должна молча пройти
    в SQL -- TypeError из сравнения `cap_mult < 1` ловится тем же except Exception,
    что и ValueError-гварды, и маршрутизируется через mark_failed. Никакого SQL не
    исполняется, ни один active-лот не тронут."""
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(TypeError):
        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult="6")
    assert db.executed == []


# ── cap_mult / ttl_days -- bool (найдено ревью круга 3, 2026-08-15) ─────────────
# bool -- подкласс int в Python: `True < 1` ложно, `True <= 0` ложно. Числовые
# guard'ы выше (`cap_mult < 1`, `ttl_days <= 0`) поэтому НЕ ловят jsonb `true` в
# default_params расписания -- ровно тот класс опечатки, ради которого guard'ы
# вообще написаны. `cap_mult=True` даёт потолок == ttl_days (ttl_days * True ==
# ttl_days) -- пол молча отключается без единого ValueError. `ttl_days=True` даёт
# ttl_days == 1 -- TTL молча меняется на 1 сутки. Явная type-проверка ловит оба
# ДО числового сравнения и ДО любого SQL.


def test_cap_mult_true_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    """cap_mult=True: `True < 1` ложно -- без явной type-проверки потолок = ttl_days
    (пол молча отключается) вместо ValueError. Воспроизведено на HEAD ветки."""
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=True)
    assert db.executed == []


def test_cap_mult_false_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    """cap_mult=False уже ловится `cap_mult < 1` (False == 0), но type-guard идёт
    первым -- проверяем, что путь всё равно ValueError, а не иной exception."""
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=False)
    assert db.executed == []


def test_ttl_days_true_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    """ttl_days=True: `True <= 0` ложно -- без явной type-проверки TTL молча
    становится 1 сутки (True ведёт себя как int 1) вместо ValueError."""
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=True, revisit_floor_quantile=0.99)
    assert db.executed == []


def test_ttl_days_false_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
    """ttl_days=False уже ловится `ttl_days <= 0` (False == 0), но type-guard идёт
    первым -- проверяем, что путь всё равно ValueError."""
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(ValueError):
        _run(db, monkeypatch, ttl_days=False, revisit_floor_quantile=0.99)
    assert db.executed == []


def test_cap_mult_zero_fails_the_run_via_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
    """Тот же контракт, что и ttl_days<=0: run помечается failed, а не остаётся 'running',
    и НИ ОДНА строка не деактивируется (в отличие от воспроизведённого на HEAD дефекта, где
    cap_mult=0 давало effective_ttl_days=0 и снимало весь активный пул источника)."""
    marked_failed: list[Any] = []
    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(
        task_mod.runs_mod,
        "mark_failed",
        lambda db, run_id, err, counters: marked_failed.append((run_id, err, counters)),
    )
    db = _FakeDB(floor_days=52.0)
    with pytest.raises(ValueError):
        task_mod.deactivate_stale_listings(
            db,  # type: ignore[arg-type]
            9,
            listing_source="avito",
            ttl_days=10,
            revisit_floor_quantile=0.99,
            cap_mult=0,
        )
    assert len(marked_failed) == 1
    assert marked_failed[0][0] == 9
    assert db.executed == []


# ── проводка cap_mult в product_handlers ─────────────────────────────────────────


def test_handler_wires_cap_mult_from_schedule_params() -> None:
    """Тот же приём, что test_handler_wires_revisit_floor_from_schedule_params:
    читаем исходник файлом (product_handlers тянет scraper_kit, которого в юнит-
    окружении может не быть) и проверяем именно проводку default_params -> вызов."""
    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("cap_mult", CAP_MULT)' in flat
    assert "cap_mult=cap_mult" in job