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

"""Витринные метрики лэндинга — задача пересчёта + публичная ручка.

ЧТО ЗДЕСЬ ПРОВЕРЯЕТСЯ И ПОЧЕМУ ИМЕННО ЭТО

  1. ЗНАЧЕНИЯ. Арифметика витрины (доля снижавших, нормировка на 30 дней,
     округления, какой sample_n к какой метрике) живёт в Python, и она проверена
     по ЧИСЛАМ: подставляем агрегаты и сверяем ровно то, что уедет на страницу.
     Тест обязан краснеть, если share посчитать от не того знаменателя или
     приписать медиане общий n вместо числа снижавших.

  2. НЕТ ВХОДА — НЕТ СТРОКИ. Отдельная проверка на каждую пустую выборку:
     подстановка правдоподобного нуля — главный способ соврать на витрине, и
     запрещена она поведением задачи, а не комментарием.

  3. ГРАНИЦЫ ВЫБОРКИ В SQL. Условия «только domklik», «наблюдение >= 14 дней»,
     «|изменение| <= 30%» на mock-сессии не проявляются: их исполняет Postgres.
     Поэтому они запинены статически по тексту запроса — иначе их молчаливое
     исчезновение (а с ним и мусор от yandex-синтетики) прошло бы незамеченным.

  4. РУЧКА. Публичность (rbac), форма ответа, и главное — пустая таблица даёт
     200 и {}, а не 500: это штатное состояние сразу после накатки миграции.
"""

from __future__ import annotations

import os
import re
import sys
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)
sys.modules.setdefault("weasyprint.CSS", _wp_mock)
sys.modules.setdefault("weasyprint.HTML", _wp_mock)

import pytest  # noqa: E402
from fastapi import FastAPI  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402

from app.api.public import mera as public_mera  # noqa: E402
from app.core.db import get_db  # noqa: E402
from app.core.rbac import _PUBLIC_PATHS, rbac_guard  # noqa: E402
from app.tasks import landing_stats as ls  # noqa: E402

_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
_MIGRATION_275 = _SQL_DIR / "275_landing_stats.sql"

PREFIX = "/api/public/mera"


# ── Мок-сессия: отдаёт заранее заданную строку на каждый из запросов задачи ───
#
# Раскладываем ответы по ПОРЯДКУ вызовов, а не по тексту SQL: порядок — часть
# контракта collect_landing_metrics (он же порядок метрик на витрине), и его
# перестановка должна быть заметна.
class _FakeSession:
    def __init__(self, rows: list[Any], *, prune_rowcount: int = 0) -> None:
        self._rows = list(rows)
        self.upserts: list[dict[str, Any]] = []
        # Чистка протухших метрик: пишем сюда параметры каждого DELETE, чтобы
        # тест видел И факт вызова, И список оставляемых метрик.
        self.prunes: list[dict[str, Any] | None] = []
        self._prune_rowcount = prune_rowcount
        self.committed = 0

    # Считываем ТОЛЬКО запросы самой витрины: по этой же сессии ходит
    # runs_mod (heartbeat/mark_done пишут в scrape_runs), и если раздавать
    # заготовленные строки по любому execute, первый же heartbeat съест
    # агрегат оценок — тест краснел бы не по своей причине.
    _METRIC_SQL_MARKERS = (
        "FROM trade_in_estimates",
        "FROM listings",
        "offer_price_history",
        "FROM deals",
    )

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> Any:
        sql = str(stmt)
        if "INSERT INTO landing_stats" in sql:
            assert params is not None
            self.upserts.append(params)
            return MagicMock()
        if "DELETE FROM landing_stats" in sql:
            self.prunes.append(params)
            return SimpleNamespace(rowcount=self._prune_rowcount)
        if not any(marker in sql for marker in self._METRIC_SQL_MARKERS):
            return MagicMock()
        assert self._rows, f"неожиданный лишний SELECT: {sql[:80]}"
        row = self._rows.pop(0)
        return SimpleNamespace(first=lambda: row)

    def commit(self) -> None:
        self.committed += 1

    def rollback(self) -> None:  # pragma: no cover — путь ошибки здесь не гоняется
        pass


def _rows(**overrides: Any) -> list[Any]:
    """Пять агрегатов в порядке вызова. Значения — прод-срез на 29.08.2026."""
    base: dict[str, Any] = {
        "estimates": SimpleNamespace(total=1123, period_days=94.0),
        "analogs": SimpleNamespace(n=975, median=Decimal("12")),
        "listing_age": SimpleNamespace(n=29568, n_active=31068, median=Decimal("26")),
        "price": SimpleNamespace(n=6276, n_cut=3018, median_pct_per_month=Decimal("-2.174")),
        "deals": SimpleNamespace(n=18657),
    }
    base.update(overrides)
    return [base["estimates"], base["analogs"], base["listing_age"], base["price"], base["deals"]]


def _by_metric(metrics: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    return {m["metric"]: m for m in metrics}


# ── 1. Значения ──────────────────────────────────────────────────────────────


def test_all_metrics_computed_from_aggregates() -> None:
    """Каждая витринная цифра — ровно то, что следует из выборки."""
    got = _by_metric(ls.collect_landing_metrics(_FakeSession(_rows())))

    assert got["estimates_total"]["value_num"] == 1123.0
    assert got["estimates_total"]["sample_n"] == 1123
    assert got["estimates_period_days"]["value_num"] == 94.0
    assert got["analogs_median"]["value_num"] == 12.0
    assert got["analogs_median"]["sample_n"] == 975
    assert got["listing_age_median_days"]["value_num"] == 26.0
    assert got["deals_total_12m"]["value_num"] == 18657.0

    # 3018/6276 = 48.087...% → 48.1 после округления до десятых.
    assert got["price_cut_share_pct"]["value_num"] == 48.1
    assert got["price_cut_share_pct"]["sample_n"] == 6276
    assert got["price_cut_median_pct_per_month"]["value_num"] == -2.17
    # Медиана считается ТОЛЬКО по снижавшим — и выборка у неё их, а не общая.
    assert got["price_cut_median_pct_per_month"]["sample_n"] == 3018


def test_share_uses_full_observed_denominator_not_only_cutters() -> None:
    """Знаменатель доли — все наблюдавшиеся, а не только снижавшие.

    Если считать от снижавших, доля всегда 100% — ровно тот дефект, который на
    проде давал 85% вместо 48% (в выборку попадали только менявшие цену).
    """
    rows = _rows(price=SimpleNamespace(n=200, n_cut=50, median_pct_per_month=Decimal("-3")))
    got = _by_metric(ls.collect_landing_metrics(_FakeSession(rows)))
    assert got["price_cut_share_pct"]["value_num"] == 25.0


def test_every_metric_carries_sample_n() -> None:
    """Цифра без размера выборки неотличима от литерала, ради замены которого
    вся эта таблица и заведена."""
    for m in ls.collect_landing_metrics(_FakeSession(_rows())):
        assert isinstance(m["sample_n"], int) and m["sample_n"] > 0, m["metric"]
        assert m["note"], m["metric"]


def test_listing_age_note_says_exposure_not_time_to_sell() -> None:
    """Величина по построению — экспозиция ЕЩЁ ВИСЯЩЕГО объявления. Названная
    «сроком продажи», она врёт (и врёт в выгодную сторону)."""
    got = _by_metric(ls.collect_landing_metrics(_FakeSession(_rows())))
    note = got["listing_age_median_days"]["note"]
    assert "ЭКСПОЗИЦИЯ" in note
    assert "НЕ срок продажи" in note


def test_listing_age_note_names_its_coverage() -> None:
    """Знаменатель обязан быть в подписи: метрика видит не все активные.

    Дата публикации есть у 29 568 из 31 068 активных объявлений ЕКБ (замер
    2026-08-30). Полторы тысячи без даты — это не ноль, и «медиана по активным
    объявлениям» без охвата читается как «по всем».

    Ломать так: убрать из note подстановку n/n_active — тест покраснеет.
    """
    got = _by_metric(ls.collect_landing_metrics(_FakeSession(_rows())))
    note = got["listing_age_median_days"]["note"]
    assert "29568" in note and "31068" in note, f"охват не назван: {note}"


def test_listing_age_counts_both_publication_date_columns() -> None:
    """Дата публикации лежит в двух колонках, и обе значат одно.

    `listing_date` пишут cian/yandex/avito, `publish_date` — yandex (тем же
    значением) и Домклик. Там, где заполнены обе, они совпадают (прод
    2026-08-30: yandex 10 761 из 10 903, avito 474 из 569). По одному
    `listing_date` Домклик выпадал целиком — 3061 активное объявление, 0 с
    датой, охват 83.6% вместо 95.2%.

    Ломать так: вернуть `ORDER BY (CURRENT_DATE - listing_date)` без COALESCE —
    тест покраснеет.
    """
    sql = str(ls._LISTING_AGE_SQL)
    assert "COALESCE(listing_date, publish_date)" in sql, (
        "метрика снова считает по одной колонке — Домклик выпадает целиком"
    )


def test_forecast_accuracy_and_time_to_sell_are_never_produced() -> None:
    """Этих величин в данных нет; их считает бэктест со своими допущениями."""
    names = {m["metric"] for m in ls.collect_landing_metrics(_FakeSession(_rows()))}
    assert not {n for n in names if "accuracy" in n or "time_to_sell" in n or "days_to_sell" in n}


# ── 2. Нет входа — нет строки ────────────────────────────────────────────────


@pytest.mark.parametrize(
    ("kwargs", "absent"),
    [
        ({"estimates": SimpleNamespace(total=0, period_days=None)}, "estimates_total"),
        ({"analogs": SimpleNamespace(n=0, median=None)}, "analogs_median"),
        ({"listing_age": SimpleNamespace(n=0, n_active=0, median=None)}, "listing_age_median_days"),
        (
            {"price": SimpleNamespace(n=0, n_cut=0, median_pct_per_month=None)},
            "price_cut_share_pct",
        ),
        ({"deals": SimpleNamespace(n=0)}, "deals_total_12m"),
    ],
)
def test_empty_input_writes_no_row_instead_of_zero(kwargs: dict[str, Any], absent: str) -> None:
    """Ноль читается как измеренный ноль («никто не снижал цену») — а измерения
    не было. Строки просто нет, фронт не рисует блок."""
    names = {m["metric"] for m in ls.collect_landing_metrics(_FakeSession(_rows(**kwargs)))}
    assert absent not in names


def test_single_estimate_gives_no_period_metric() -> None:
    """Период между первым и последним расчётом при одном расчёте — 0 дней,
    что является артефактом единственной точки, а не сроком работы."""
    rows = _rows(estimates=SimpleNamespace(total=1, period_days=0.0))
    names = {m["metric"] for m in ls.collect_landing_metrics(_FakeSession(rows))}
    assert "estimates_total" in names
    assert "estimates_period_days" not in names


def test_no_cutters_leaves_share_but_drops_median() -> None:
    """Никто не снижал — доля 0% ИЗМЕРЕНА (наблюдения были), а медианы снижения
    не существует: писать её нулём значило бы выдумать «снижают на 0%»."""
    rows = _rows(price=SimpleNamespace(n=120, n_cut=0, median_pct_per_month=None))
    got = _by_metric(ls.collect_landing_metrics(_FakeSession(rows)))
    assert got["price_cut_share_pct"]["value_num"] == 0.0
    assert "price_cut_median_pct_per_month" not in got


def test_refresh_upserts_every_metric_and_commits() -> None:
    db = _FakeSession(_rows())
    counters = ls.refresh_landing_stats(db, run_id=1)  # type: ignore[arg-type]
    assert counters["metrics_written"] == len(db.upserts) == 7
    # >=1, а не ==1: runs_mod коммитит свои heartbeat/mark_done по той же сессии.
    assert db.committed >= 1
    assert {u["metric"] for u in db.upserts} == {
        "estimates_total",
        "estimates_period_days",
        "analogs_median",
        "listing_age_median_days",
        "price_cut_share_pct",
        "price_cut_median_pct_per_month",
        "deals_total_12m",
    }


def test_metric_that_stopped_computing_is_deleted_not_left_stale() -> None:
    """Пропал вход у метрики — строка УДАЛЯЕТСЯ, а не доживает со старым
    computed_at: иначе ручка отдаёт её неотличимо от посчитанной сегодня.

    Здесь сделок нет (`deals.n = 0`), значит `deals_total_12m` в наборе не
    появляется — и именно её обязан вынести DELETE, оставив ровно посчитанные.
    """
    rows = _rows(deals=SimpleNamespace(n=0))
    db = _FakeSession(rows, prune_rowcount=1)
    counters = ls.refresh_landing_stats(db, run_id=1)  # type: ignore[arg-type]

    assert len(db.prunes) == 1, "чистка протухших метрик не выполнена"
    kept = set(db.prunes[0]["kept"])  # type: ignore[index]
    assert kept == {u["metric"] for u in db.upserts}
    assert "deals_total_12m" not in kept, "метрика без входа осталась бы на витрине"
    assert counters["metrics_removed"] == 1


def test_totally_empty_run_keeps_the_showcase_instead_of_wiping_it() -> None:
    """Разом пропали ВСЕ входы — это похоже на поломку прогона (пустая или
    недоступная база), а не на пять одновременных «данных больше нет». По такому
    признаку витрина не стирается: DELETE не выполняется вовсе."""
    empty = _rows(
        estimates=SimpleNamespace(total=0, period_days=None),
        analogs=SimpleNamespace(n=0, median=None),
        listing_age=SimpleNamespace(n=0, n_active=0, median=None),
        price=SimpleNamespace(n=0, n_cut=0, median_pct_per_month=None),
        deals=SimpleNamespace(n=0),
    )
    db = _FakeSession(empty)
    counters = ls.refresh_landing_stats(db, run_id=1)  # type: ignore[arg-type]

    assert db.upserts == []
    assert db.prunes == [], "пустой прогон стёр бы всю витрину"
    assert counters["metrics_written"] == 0
    assert counters["metrics_removed"] == 0


# ── 3. Границы выборки, которые исполняет Postgres ───────────────────────────


def test_price_sql_takes_domklik_only() -> None:
    """avito/yandex сюда попасть не могут: у первого нет стартовой цены в
    истории, второй сеет синтетическую пару со сдвигом в сутки."""
    sql = str(ls._PRICE_MOVES_SQL)
    assert "source = 'domklik'" in sql
    assert "avito" not in sql and "yandex" not in sql


def test_price_sql_keeps_single_row_listings_in_denominator() -> None:
    """Знаменатель доли снижений включает объявления с ОДНОЙ записью истории.

    Это тот самый дефект, из-за которого на проде получалось бы 84.8% вместо
    48.1%: у domklik триггер пишет стартовую цену, поэтому одна запись означает
    «цену не менял» — наблюдение, а не отсутствие данных. Выкинув такие строки,
    считаешь долю снижавших ТОЛЬКО среди менявших цену, то есть почти единицу.

    Гейт текстовый, а не прогон на живой базе: DATABASE_URL в CI —
    заглушка (deploy-tradein.yml: `test:` job), Postgres в тестовой джобе нет,
    и живой тест по образцу test_purge_expired_trade_in_data.py тут молча
    скипался бы — то есть не гейтил бы ничего. Пин проверяет две половины
    дефекта: (1) однострочные попадают в `moved` через ветку CASE со значением
    0 («не снижал»), а не отбрасываются; (2) нигде в запросе нет фильтра по
    числу записей, который бы их отсёк.
    """
    sql = str(ls._PRICE_MOVES_SQL)

    case = re.search(r"CASE\b(?P<body>.*?)\bEND\b", sql, re.S | re.I)
    assert case is not None, "исчезла ветка для однострочных — они больше не «не снижал»"
    body = case.group("body")
    assert "n_rows" in body, "ветка перестала различать однострочные записи истории"
    assert re.search(r"\b(THEN|ELSE)\s+0\b", body), (
        "однострочным объявлениям больше не приписывается изменение 0% — "
        "они либо выпали из выборки, либо получили выдуманное значение"
    )

    rest = sql.replace(case.group(0), "")
    leftover = re.search(r"n_rows\s*(>=|>|<|<>|=|!=)", rest)
    assert leftover is None, (
        f"появился фильтр по числу записей истории вне ветки CASE ({leftover.group(0)!r}) — "
        "он выкидывает не менявших цену из знаменателя, доля вырастет с ~48% до ~85%"
    )
    assert not re.search(r"\bHAVING\b", rest, re.I), (
        "HAVING в агрегате истории отсекает однострочные ещё до знаменателя"
    )


def test_price_sql_keeps_span_and_outlier_gates() -> None:
    sql = str(ls._PRICE_MOVES_SQL)
    assert "span_days" in sql, "исчез порог наблюдения — короткоживущие дадут ложное «не снижал»"
    assert "max_abs_pct" in sql, "исчезла отсечка аномалий — перевыставленные объекты как торг"
    assert ls._PRICE_SPAN_DAYS == 14
    assert ls._PRICE_MAX_ABS_PCT == 30


def test_city_scoped_metrics_are_parameterised_by_ekb() -> None:
    for sql in (str(ls._LISTING_AGE_SQL), str(ls._DEALS_SQL)):
        assert "CAST(:city AS text)" in sql
    assert ls.EKB == "Екатеринбург"


def test_analogs_median_excludes_estimates_without_analogs() -> None:
    """n_analogs=0 — это отказ расчёта, а не «ноль аналогов»; в медиане он
    занизил бы величину наблюдением, где мерить было нечего."""
    assert "n_analogs > 0" in str(ls._ANALOGS_SQL)


# ── Миграция ─────────────────────────────────────────────────────────────────


def test_migration_275_is_idempotent_and_registers_the_job() -> None:
    sql = _MIGRATION_275.read_text("utf-8")
    assert "CREATE TABLE IF NOT EXISTS landing_stats" in sql
    assert "metric       text        PRIMARY KEY" in sql
    assert "ON CONFLICT (source) DO NOTHING" in sql
    assert "'landing_stats_refresh'" in sql


def test_migration_275_has_no_psycopg_cast_trap() -> None:
    """`:x::type` psycopg v3 разбирает как именованный параметр — в проекте
    разрешён только CAST(:x AS type)."""
    assert not re.search(r":\w+::", _MIGRATION_275.read_text("utf-8"))


def test_task_is_registered_in_the_scheduler_registry() -> None:
    """Без Handler'а строка расписания резолвится в никуда и джоба не бежит."""
    from app.services.product_handlers import build_product_handlers

    handlers = build_product_handlers(MagicMock())
    assert "landing_stats_refresh" in handlers


# ── 4. Публичная ручка ───────────────────────────────────────────────────────


_STAT_ROWS = [
    SimpleNamespace(
        metric="estimates_total",
        value_num=Decimal("1123"),
        value_text=None,
        sample_n=1123,
        note="Расчётов сделано",
        computed_at=datetime(2026, 8, 29, 5, 0, tzinfo=UTC),
    ),
    SimpleNamespace(
        metric="price_cut_share_pct",
        value_num=Decimal("48.1"),
        value_text=None,
        sample_n=6276,
        note="Только Домклик",
        computed_at=datetime(2026, 8, 29, 5, 0, tzinfo=UTC),
    ),
]


def _client(rows: list[Any]) -> TestClient:
    """Приложение с РЕАЛЬНЫМ rbac_guard — тем же, что вешает app/main.py."""
    app = FastAPI()
    app.middleware("http")(rbac_guard)
    app.include_router(public_mera.router, prefix=PREFIX)

    db = MagicMock()
    db.execute.return_value.fetchall.return_value = rows

    def _override_db():
        yield db

    app.dependency_overrides[get_db] = _override_db
    return TestClient(app)


@pytest.fixture(autouse=True)
def _reset_stats_limiter():
    public_mera._stats_limiter._hits.clear()
    yield
    public_mera._stats_limiter._hits.clear()


def test_stats_path_is_public_in_rbac() -> None:
    assert f"{PREFIX}/stats" in _PUBLIC_PATHS, (
        "без строки в rbac._PUBLIC_PATHS анониму прилетит 401 и лэндинг останется без чисел"
    )


def test_anonymous_gets_stats_keyed_by_metric() -> None:
    resp = _client(_STAT_ROWS).get(f"{PREFIX}/stats")
    assert resp.status_code == 200
    body = resp.json()
    assert set(body) == {"estimates_total", "price_cut_share_pct"}
    assert body["estimates_total"]["value"] == 1123.0
    assert body["price_cut_share_pct"]["value"] == 48.1
    assert body["price_cut_share_pct"]["sample_n"] == 6276
    assert body["price_cut_share_pct"]["note"] == "Только Домклик"
    assert body["estimates_total"]["computed_at"].startswith("2026-08-29T05:00")


def test_empty_table_is_a_valid_answer_not_an_error() -> None:
    """Состояние сразу после накатки миграции: задача ещё не отрабатывала.
    500 здесь сломал бы страницу целиком ради отсутствующего блока."""
    resp = _client([]).get(f"{PREFIX}/stats")
    assert resp.status_code == 200
    assert resp.json() == {}


def test_metric_without_numeric_value_falls_back_to_text_then_null() -> None:
    rows = [
        SimpleNamespace(
            metric="period_label",
            value_num=None,
            value_text="май–август 2026",
            sample_n=1123,
            note=None,
            computed_at=datetime(2026, 8, 29, tzinfo=UTC),
        ),
        SimpleNamespace(
            metric="nothing_measured",
            value_num=None,
            value_text=None,
            sample_n=None,
            note=None,
            computed_at=datetime(2026, 8, 29, tzinfo=UTC),
        ),
    ]
    body = _client(rows).get(f"{PREFIX}/stats").json()
    assert body["period_label"]["value"] == "май–август 2026"
    assert body["nothing_measured"]["value"] is None


def test_stats_rate_limited_per_ip() -> None:
    client = _client(_STAT_ROWS)
    codes = [client.get(f"{PREFIX}/stats").status_code for _ in range(public_mera._STATS_LIMIT + 1)]
    assert codes[-1] == 429
    assert set(codes[:-1]) == {200}