"""Витринные метрики лэндинга — задача пересчёта + публичная ручка. ЧТО ЗДЕСЬ ПРОВЕРЯЕТСЯ И ПОЧЕМУ ИМЕННО ЭТО 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.*?)\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}