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

"""Анонимный расчёт /api/public/mera/estimate* — что здесь запинено и почему.

1. СОГЛАСИЕ ДО ЗАПИСИ. Отказ без согласия обязан случиться ДО того, как
   адрес физлица дойдёт до БД. Поэтому проверяется не только код 422, но и
   то, что расчёт вообще не запускался и в сессию не ушло ни одного запроса:
   «422, но строка записана» выглядит в логах как успех приватности и им не
   является.

2. ФЛАГ. Выключенный `public_estimate_enabled` обязан давать 404 — иначе
   мерж этого кода сам по себе открывает наружу запись ПДн.

3. ТОКЕН. В БД уходит ХЭШ, а не токен; выборка отфильтрована по сроку
   жизни; «нет такого» и «протух» неотличимы снаружи.

4. БЕСПЛАТНАЯ ЧАСТЬ. Публичный ответ не содержит ни одного платного поля.
   Проверяется набором ключей на равенство: любое добавленное поле роняет
   тест, в том числе случайно добавленное платное.

5. ДЕЛЕГАЦИЯ. Весь анти-абузный рассказ ручки (анонимная квота на связку
   cookie+IP, семафор одновременности, 503 вместо 502, consent-гейт) держится
   ровно на ОДНОМ факте: в `app.api.v1.trade_in.estimate` уходит
   `x_authenticated_user=None`. `AsyncMock` съедает любую сигнатуру, поэтому
   «расчёт вызвали» тут ничего не значит — проверяются фактические kwargs
   вызова, причём запрос идёт С заголовком `X-Authenticated-User`: подмена
   `None` на чтение заголовка обязана красить тест, иначе публичная ручка
   молча начнёт считать чужим пользователем и мимо анонимной квоты.
"""

from __future__ import annotations

import hashlib
import os
import sys
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch

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.schemas.trade_in import AggregatedEstimate, CoverageProbeResponse  # noqa: E402

PREFIX = "/api/public/mera"

_BODY = {
    "address": "Малышева 51",
    "area_m2": 54.0,
    "rooms": 2,
    "city_hint": "Екатеринбург",
    "consent": True,
}

_FAKE_COVERAGE = CoverageProbeResponse(
    status="ok",
    n_listings=34,
    median_listing_age_days=44,
    n_with_age=6,
    radius_m=1000,
    city="Екатеринбург",
    threshold=10,
)

_TOKEN_EXPIRES = datetime(2026, 9, 5, 12, 0, tzinfo=UTC)
_FAKE_ESTIMATE_ID = "11111111-1111-1111-1111-111111111111"


def _fake_estimate_result() -> MagicMock:
    """Результат закрытого контура. MagicMock, а не собранный AggregatedEstimate:
    хендлеру нужны четыре атрибута, а перечисление всех полей платной модели
    здесь означало бы, что тест придётся править при каждой правке эстиматора."""
    result = MagicMock()
    result.estimate_id = _FAKE_ESTIMATE_ID
    result.n_analogs = 27
    result.target_lat = 56.838
    result.target_lon = 60.597
    return result


@pytest.fixture(autouse=True)
def _reset_limiters():
    public_mera._estimate_limiter._hits.clear()
    public_mera._estimate_read_limiter._hits.clear()
    public_mera._daily_estimate_limiter._hits.clear()
    yield
    public_mera._estimate_limiter._hits.clear()
    public_mera._estimate_read_limiter._hits.clear()
    public_mera._daily_estimate_limiter._hits.clear()


@pytest.fixture()
def db() -> MagicMock:
    session = MagicMock()
    session.execute.return_value.fetchone.return_value = MagicMock(
        public_token_expires_at=_TOKEN_EXPIRES,
        n_analogs=27,
        lat=56.838,
        lon=60.597,
        rooms=2,
        area_m2=54.0,
    )
    return session


@pytest.fixture()
def client(db: MagicMock) -> TestClient:
    """Без rbac-мидлвари: узость auth-исключения проверяет соседний
    tests/test_public_mera_api.py, здесь предмет — сами ручки."""
    app = FastAPI()
    app.include_router(public_mera.router, prefix=PREFIX)
    app.dependency_overrides[get_db] = lambda: db
    return TestClient(app)


@pytest.fixture()
def flag_on():
    with patch.object(public_mera.settings, "public_estimate_enabled", True):
        yield


# ── 1. Согласие фиксируется до первого обращения к БД ────────────────────────


def test_estimate_without_consent_is_422_and_writes_nothing(
    client: TestClient, db: MagicMock, flag_on: None
) -> None:
    body = {k: v for k, v in _BODY.items() if k != "consent"}
    with patch.object(public_mera, "estimate", AsyncMock()) as estimate_mock:
        resp = client.post(f"{PREFIX}/estimate", json=body)

    assert resp.status_code == 422, resp.text
    assert not estimate_mock.called, (
        "расчёт запустился без согласия — адрес физлица дошёл бы до "
        "trade_in_estimates раньше, чем человек что-либо разрешил"
    )
    assert db.execute.call_args_list == [], (
        "в БД ушёл запрос при отсутствии согласия: проверка согласия сдвинулась "
        "ПОСЛЕ работы с данными"
    )


def test_estimate_with_consent_false_is_also_422(client: TestClient, flag_on: None) -> None:
    """`consent: false` — это осознанный отказ, а не «поле не прислали»."""
    with patch.object(public_mera, "estimate", AsyncMock()) as estimate_mock:
        resp = client.post(f"{PREFIX}/estimate", json={**_BODY, "consent": False})
    assert resp.status_code == 422, resp.text
    assert not estimate_mock.called


# ── 2. Флаг ──────────────────────────────────────────────────────────────────


def test_estimate_is_404_while_flag_is_off(client: TestClient) -> None:
    with patch.object(public_mera, "estimate", AsyncMock()) as estimate_mock:
        resp = client.post(f"{PREFIX}/estimate", json=_BODY)
    assert resp.status_code == 404, resp.text
    assert not estimate_mock.called


def test_estimate_read_is_404_while_flag_is_off(client: TestClient, db: MagicMock) -> None:
    resp = client.post(f"{PREFIX}/estimate/read", json={"token": "x" * 32})
    assert resp.status_code == 404, resp.text
    assert db.execute.call_args_list == []


def test_flag_default_is_off() -> None:
    """Дефолт — выключено: мерж кода не должен открывать запись ПДн наружу."""
    from app.core.config import Settings

    assert Settings.model_fields["public_estimate_enabled"].default is False


# ── 3. Токен ─────────────────────────────────────────────────────────────────


def test_estimate_stores_token_hash_not_token(
    client: TestClient, db: MagicMock, flag_on: None
) -> None:
    with (
        patch.object(public_mera, "estimate", AsyncMock(return_value=_fake_estimate_result())),
        patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE),
    ):
        resp = client.post(f"{PREFIX}/estimate", json=_BODY)

    assert resp.status_code == 200, resp.text
    token = resp.json()["token"]
    assert len(token) >= 32, "короткий токен перебираем"

    params = db.execute.call_args_list[0].args[1]
    assert token not in str(params), "сам токен уехал в БД — дамп базы открывает чужие расчёты"
    assert params["token_hash"] == hashlib.sha256(token.encode()).hexdigest()
    assert params["id"] == _FAKE_ESTIMATE_ID, (
        "токен вешается не на ту строку: UPDATE обязан идти по id ТОЛЬКО ЧТО "
        "посчитанной оценки (result.estimate_id), иначе ссылка либо ведёт в чужой "
        "расчёт, либо не ведёт никуда"
    )


def test_read_filters_by_expiry_and_hash(client: TestClient, db: MagicMock, flag_on: None) -> None:
    """ЧТО проверено: в тексте отправленного SQL присутствует предикат срока жизни,
    а в параметры уехал ХЭШ токена, не сам токен.

    ЧЕГО НЕ проверено: что протухший токен действительно не читается. Сессия здесь —
    MagicMock, запрос не исполняется: SQL не разбирается, `fetchone()` отдаёт строку
    из фикстуры независимо от WHERE. Поэтому зелёными останутся, например, предикат,
    перенесённый туда, где он ничего не отсекает, сравнение NOW() с полем не того
    типа и любая ошибка в самом сравнении — текст-то совпадает.

    Поведенческой версии нет намеренно: она требует живого Postgres с миграцией 278
    (образец self-skip-теста — tests/test_purge_expired_trade_in_data.py::_live_session),
    а писать её вслепую, ни разу не прогнав, значит завести ещё один тест, зелёный
    по построению. Появится доступный Postgres — этот тест заменяется на вставку
    двух строк (свежий токен и просроченный) с проверкой 200 против 404.
    """
    token = "t" * 40
    with patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE):
        resp = client.post(f"{PREFIX}/estimate/read", json={"token": token})
    assert resp.status_code == 200, resp.text

    sql = str(db.execute.call_args_list[0].args[0])
    assert "public_token_expires_at > NOW()" in sql, (
        "из выборки пропал срок жизни ссылки — протухший токен снова читает расчёт"
    )
    params = db.execute.call_args_list[0].args[1]
    assert params["token_hash"] == hashlib.sha256(token.encode()).hexdigest()


def test_unknown_or_expired_token_is_404(client: TestClient, db: MagicMock, flag_on: None) -> None:
    """Чужой и протухший токен неотличимы: 404 в обоих случаях.

    Мок сессии отдаёт None ровно так же, как реальный SELECT с предикатом
    `public_token_expires_at > NOW()` — то есть и для несуществующего токена,
    и для просроченного.
    """
    db.execute.return_value.fetchone.return_value = None
    resp = client.post(f"{PREFIX}/estimate/read", json={"token": "z" * 40})
    assert resp.status_code == 404, resp.text
    assert "не найден" in resp.json()["detail"].lower()


# ── 4. В публичном ответе нет платного ───────────────────────────────────────

# Поля AggregatedEstimate, которые продукт продаёт. Список не для красоты: он
# сверяется с реальной моделью ниже, поэтому переименование поля в эстиматоре
# не оставит здесь мёртвую строку-обманку.
_PAID_FIELDS = {
    "median_price_rub",
    "range_low_rub",
    "range_high_rub",
    "median_price_per_m2",
    "market_percentile",
    "analogs",
    "actual_deals",
    "expected_sold_price_rub",
    "est_days_on_market",
    "price_trend",
}

_FREE_FIELDS = {"token", "token_expires_at", "n_analogs", "coverage"}


def test_paid_field_names_still_exist_in_estimator_model() -> None:
    """Контроль самого контроля: если поле переименовали, тест ниже сравнивал бы
    публичный ответ с несуществующими именами и был бы зелёным по построению."""
    missing = _PAID_FIELDS - set(AggregatedEstimate.model_fields)
    assert not missing, f"эти поля исчезли из AggregatedEstimate, обнови список: {missing}"


def test_public_result_model_exposes_only_free_fields() -> None:
    assert set(public_mera.PublicEstimateResult.model_fields) == _FREE_FIELDS, (
        "изменился состав публичного ответа. Любое поле отсюда видит любой аноним "
        "без оплаты — правь ОСОЗНАННО вместе с этим тестом"
    )


def test_public_estimate_response_leaks_no_paid_field(client: TestClient, flag_on: None) -> None:
    with (
        patch.object(public_mera, "estimate", AsyncMock(return_value=_fake_estimate_result())),
        patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE),
    ):
        resp = client.post(f"{PREFIX}/estimate", json=_BODY)
    body = resp.json()
    assert set(body) == _FREE_FIELDS, f"публичный ответ отдаёт лишнее: {set(body) - _FREE_FIELDS}"
    assert not _PAID_FIELDS & set(body)
    # Вложенный coverage тоже без цен — по построению CoverageProbeResponse,
    # но проверяем, а не верим: он мог обрасти полем «медиана ₽/м²».
    assert not _PAID_FIELDS & set(body["coverage"])
    assert body["n_analogs"] == 27
    assert body["coverage"]["median_listing_age_days"] == 44


def test_read_returns_same_free_shape(client: TestClient, flag_on: None) -> None:
    """POST и повторное чтение обязаны отдавать ОДНУ форму: разойдись они —
    фронт после перезагрузки страницы показал бы не то же самое."""
    with patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE):
        resp = client.post(f"{PREFIX}/estimate/read", json={"token": "t" * 40})
    assert resp.status_code == 200, resp.text
    assert set(resp.json()) == _FREE_FIELDS
    assert resp.json()["n_analogs"] == 27


# ── 5. Делегация в закрытый контур ───────────────────────────────────────────


def test_estimate_delegates_anonymously_despite_auth_header(
    client: TestClient, db: MagicMock, flag_on: None
) -> None:
    """Единственный тест, который смотрит НА АРГУМЕНТЫ делегации, а не на факт вызова.

    Запрос идёт с `X-Authenticated-User: admin` — то есть ровно тем заголовком,
    который в закрытом контуре означает «это авторизованный пользователь». Ручка
    обязана всё равно позвать эстиматор анонимом: `x_authenticated_user=None`
    включает там consent-гейт и анонимную квоту на связку cookie+IP. Пробрось
    сюда заголовок — и публичная форма начнёт считать от чужого имени в обход
    квоты, а все остальные тесты этого файла останутся зелёными: `AsyncMock`
    принимает любую сигнатуру.
    """
    with (
        patch.object(
            public_mera, "estimate", AsyncMock(return_value=_fake_estimate_result())
        ) as estimate_mock,
        patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE),
    ):
        resp = client.post(
            f"{PREFIX}/estimate", json=_BODY, headers={"X-Authenticated-User": "admin"}
        )

    assert resp.status_code == 200, resp.text
    call = estimate_mock.await_args
    assert call is not None, "делегации не было вовсе"
    assert call.args == (), (
        "аргументы поехали позиционно — сверка по именам ниже перестала что-либо "
        "проверять; зови эстиматор с ключевыми словами"
    )
    assert "x_authenticated_user" in call.kwargs, (
        "аргумент исчез из вызова: у эстиматора он по умолчанию None, но тогда "
        "гарантия анонимности держится на чужом дефолте, а не на этой ручке"
    )
    assert call.kwargs["x_authenticated_user"] is None, (
        "публичная ручка передала пользователя в закрытый контур: consent-гейт и "
        "анонимная квота выключаются, аноним считает от чужого имени "
        f"(пришло: {call.kwargs['x_authenticated_user']!r})"
    )
    # Остальное едет тем же вызовом: сессия — та, что отдана зависимостью (иначе
    # UPDATE токена ниже пишет в другую транзакцию), тело — то, что прислал аноним.
    assert call.kwargs["db"] is db
    assert call.kwargs["payload"].address == _BODY["address"]
    assert call.kwargs["payload"].area_m2 == _BODY["area_m2"]
    assert call.kwargs["payload"].rooms == _BODY["rooms"]


# ── 6. Бюджеты ───────────────────────────────────────────────────────────────


def test_estimate_rate_limited_per_ip(client: TestClient, flag_on: None) -> None:
    with (
        patch.object(public_mera, "estimate", AsyncMock(return_value=_fake_estimate_result())),
        patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE),
    ):
        codes = [
            client.post(f"{PREFIX}/estimate", json=_BODY).status_code
            for _ in range(public_mera._ESTIMATE_LIMIT + 1)
        ]
    assert codes[:-1] == [200] * public_mera._ESTIMATE_LIMIT
    assert codes[-1] == 429, f"бюджет не сработал: {codes}"


def test_daily_budget_stops_estimates_for_everyone(client: TestClient, flag_on: None) -> None:
    """Общий потолок — про сумму по всем IP, а не про одного клиента."""
    for _ in range(public_mera._DAILY_ESTIMATE_BUDGET):
        public_mera._daily_estimate_limiter.record(public_mera._ESTIMATE_GLOBAL_KEY)
    with patch.object(public_mera, "estimate", AsyncMock()) as estimate_mock:
        resp = client.post(f"{PREFIX}/estimate", json=_BODY)
    assert resp.status_code == 429, resp.text
    assert int(resp.headers["Retry-After"]) > 0
    assert not estimate_mock.called