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

"""Снятие устаревших ограничено регионами, где есть регулярный пересбор.

Джоба трактует молчание как «объявление ушло»: строку давно не видели свежей ->
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