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

"""Публичный B2C-API МЕРЫ — /api/public/mera/* (issue #2911).

Что здесь запинено и почему именно это:

  1. ПЕРИМЕТР. Набор маршрутов под /api/public/ проверяется на РАВЕНСТВО, а не
     на вхождение. Третья ручка, добавленная сюда без правки этого теста,
     уронит его — а решение «эта ручка доступна анониму» не должно приниматься
     мимоходом, добавлением файла в каталог.

  2. СВЯЗКА С rbac. Публичность даёт не расположение файла, а строка в
     `rbac._PUBLIC_PATHS`; тест сверяет, что для каждого маршрута она есть, и
     наоборот — что `_PUBLIC_PATHS` не содержит /api/public/ путей, которых
     больше нет в роутере (иначе освободившийся путь остался бы открытым для
     любого, кто его потом займёт).

  3. УЗОСТЬ ИСКЛЮЧЕНИЯ. Соседний non-public путь в том же приложении обязан
     по-прежнему отдавать 401 анониму. Без этой проверки тест «аноним получает
     200» одинаково зелёный и когда исключение точечное, и когда auth-гейт
     снят целиком.

  4. БЮДЖЕТ. 429 после исчерпания per-IP окна — иначе внешний (платный)
     геокодер открыт наружу без ограничений.
"""

from __future__ import annotations

import os
import sys
from unittest.mock import AsyncMock, MagicMock, patch

# Settings требует DATABASE_URL на конструирование — stub до любого app-импорта
# (тот же паттерн, что в остальных tests/*.py).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

# WeasyPrint (GTK) в CI/Windows нет, а app.api.v1.trade_in тянет его на импорте —
# public/mera.py делегирует туда, значит стаб нужен и здесь.
_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.api.v1.geocode import SuggestResponse  # noqa: E402
from app.core.db import get_db  # noqa: E402
from app.core.rbac import _PUBLIC_PATHS, rbac_guard  # noqa: E402
from app.schemas.trade_in import CoverageProbeResponse  # noqa: E402

# Внешний префикс монтирования — ровно тот, что в app/main.py.
PREFIX = "/api/public/mera"

_COVERAGE_BODY = {"lat": 56.838, "lon": 60.597, "rooms": 2, "area_m2": 54.0}

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


@pytest.fixture(autouse=True)
def _reset_limiters():
    """Лимитеры — состояние МОДУЛЯ, переживающее тест.

    Без сброса порядок тестов начинает влиять на результат: тест бюджета
    выбирает окно, а следующий за ним «аноним получает 200» ловит 429 и падает
    не по своей вине.
    """
    public_mera._suggest_limiter._hits.clear()
    public_mera._coverage_limiter._hits.clear()
    yield
    public_mera._suggest_limiter._hits.clear()
    public_mera._coverage_limiter._hits.clear()


@pytest.fixture()
def client() -> TestClient:
    """Приложение с РЕАЛЬНЫМ rbac_guard — тем же, что регистрирует app/main.py.

    Рядом с публичным роутером намеренно висит закрытый маршрут-двойник: он и
    делает проверку 3 (узость исключения) возможной внутри одного приложения.
    """
    app = FastAPI()
    app.middleware("http")(rbac_guard)
    app.include_router(public_mera.router, prefix=PREFIX)

    @app.get("/api/v1/trade-in/coverage")
    def _closed_twin() -> dict[str, str]:  # pragma: no cover — до хендлера не доходит
        return {"ok": "should never be reachable anonymously"}

    def _override_db():
        yield MagicMock()

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


# ── 1-2. Периметр и его связка с rbac ────────────────────────────────────────


def test_public_router_exposes_exactly_two_routes() -> None:
    paths = {r.path for r in public_mera.router.routes}
    assert paths == {"/suggest", "/coverage"}, (
        "изменился набор публичных (анонимных) ручек МЕРЫ. Это не рефакторинг: "
        "всё под /api/public/ проксируется на meraocenka.ru целиком и доступно "
        "без идентичности. Обнови тест ОСОЗНАННО вместе с rbac._PUBLIC_PATHS."
    )


def test_every_public_route_is_exempt_in_rbac() -> None:
    for route in public_mera.router.routes:
        full = f"{PREFIX}{route.path}"
        assert full in _PUBLIC_PATHS, (
            f"{full} смонтирован в публичном пакете, но rbac его не пропускает — "
            "анониму прилетит 401, форма на лэндинге молча сломается"
        )


def test_rbac_has_no_stale_public_mera_exemptions() -> None:
    """Обратное направление: в `_PUBLIC_PATHS` нет /api/public/ путей-призраков.

    Удалённая ручка, чья строка осталась в исключениях, открывает этот путь
    заранее — для любого маршрута, который потом займёт то же место.
    """
    live = {f"{PREFIX}{r.path}" for r in public_mera.router.routes}
    stale = {p for p in _PUBLIC_PATHS if p.startswith("/api/public/")} - live
    assert not stale, f"в rbac._PUBLIC_PATHS остались несуществующие пути: {stale}"


# ── 3. Аноним проходит, но исключение узкое ──────────────────────────────────


def test_anonymous_gets_coverage(client: TestClient) -> None:
    with patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE):
        resp = client.post(f"{PREFIX}/coverage", json=_COVERAGE_BODY)
    assert resp.status_code == 200, resp.text
    assert resp.json()["city"] == "Екатеринбург"


def test_anonymous_gets_suggest(client: TestClient) -> None:
    with patch.object(
        public_mera,
        "suggest_addresses",
        AsyncMock(return_value=SuggestResponse(items=[])),
    ):
        resp = client.post(f"{PREFIX}/suggest", json={"q": "Малышева"})
    assert resp.status_code == 200, resp.text
    assert resp.json() == {"items": []}


def test_suggest_is_post_so_address_never_lands_in_access_log() -> None:
    """Адрес едет ТЕЛОМ, а не в query.

    На публичном домене включён access-лог Caddy, и он пишет URI целиком —
    `GET /suggest?q=Малышева+51` сохранил бы адрес квартиры в файл рядом с IP
    посетителя. На той же странице мы обещаем, что введённый адрес нигде не
    сохраняем. Метод — часть этого обещания, поэтому запинен: вернуть GET
    «для каноничности» нельзя, не сломав обещание.
    """
    route = next(r for r in public_mera.router.routes if r.path == "/suggest")
    assert route.methods == {"POST"}


def test_closed_twin_still_401_for_anonymous(client: TestClient) -> None:
    """Исключение точечное, а не «auth-гейт снят».

    Без этой проверки тесты выше остаются зелёными и в случае, когда
    rbac_guard перестал требовать идентичность вообще.
    """
    resp = client.get("/api/v1/trade-in/coverage")
    assert resp.status_code == 401, resp.text


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


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


def test_rate_limited_response_carries_retry_after(client: TestClient) -> None:
    """429 без Retry-After — это «попробуйте когда-нибудь»: клиент начнёт
    долбить сразу же, а лимит существует ровно чтобы этого не было."""
    with patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE):
        for _ in range(public_mera._COVERAGE_LIMIT):
            client.post(f"{PREFIX}/coverage", json=_COVERAGE_BODY)
        resp = client.post(f"{PREFIX}/coverage", json=_COVERAGE_BODY)
    assert resp.status_code == 429
    assert int(resp.headers["Retry-After"]) > 0


def test_suggest_and_coverage_budgets_are_independent(client: TestClient) -> None:
    """Общий лимитер на две ручки означал бы, что набор адреса (частые
    подсказки) выедает бюджет пробы — человек не смог бы нажать кнопку."""
    with patch.object(
        public_mera,
        "suggest_addresses",
        AsyncMock(return_value=SuggestResponse(items=[])),
    ):
        for _ in range(public_mera._SUGGEST_LIMIT):
            client.post(f"{PREFIX}/suggest", json={"q": "Малышева"})

    with patch.object(public_mera, "coverage_probe", return_value=_FAKE_COVERAGE):
        resp = client.post(f"{PREFIX}/coverage", json=_COVERAGE_BODY)
    assert resp.status_code == 200, resp.text


# ── Контракт ответа ──────────────────────────────────────────────────────────


def test_public_coverage_reuses_v1_response_model(client: TestClient) -> None:
    """Публичная ручка отдаёт ТОТ ЖЕ тип, что и v1.

    Это не косметика: на `CoverageProbeResponse` уже стоит гейт «ни одного
    price-подобного поля» (tests/test_coverage_probe_endpoint.py). Заведи здесь
    отдельную модель — и публичный ответ окажется вне того гейта, а именно он
    держит продуктовое правило «бесплатный шаг не отдаёт цену».
    """
    route = next(r for r in public_mera.router.routes if r.path == "/coverage")
    assert route.response_model is CoverageProbeResponse


def test_suggest_limit_ceiling_is_lower_than_v1(client: TestClient) -> None:
    """Потолок limit публичной ручки — 10 (у v1 15). Каждый лишний кандидат
    может стоить внешнего (платного) вызова, а публичный UI больше и не
    показывает."""
    with patch.object(
        public_mera,
        "suggest_addresses",
        AsyncMock(return_value=SuggestResponse(items=[])),
    ):
        at_ceiling = client.post(f"{PREFIX}/suggest", json={"q": "Малышева", "limit": 10})
        above = client.post(f"{PREFIX}/suggest", json={"q": "Малышева", "limit": 15})
    assert at_ceiling.status_code == 200, at_ceiling.text
    assert above.status_code == 422, above.text


# ── 5. Обещание «адрес нигде не сохраняется» ─────────────────────────────────
#
# Три пути утечки, найденные состязательным ревью 16.08.2026 и воспроизведённые
# на живом коде. Тесты сформулированы от обещания, а не от реализации: пока на
# публичной странице и в политике ПДн написано «не сохраняется», эти проверки
# обязаны быть зелёными.


def test_address_is_redacted_from_logs_inside_public_scope(caplog) -> None:
    """Геокодер логирует введённую строку открытым текстом на каждый вызов, а
    прод пишет stdout контейнеров в persistent journald — то есть адрес ложился
    на диск рядом с IP того же запроса в access-логе Caddy."""
    import logging

    from app.core.public_request import install_address_log_redaction, public_request_scope

    install_address_log_redaction()
    dadata_logger = logging.getLogger("app.services.dadata")

    with caplog.at_level(logging.INFO):
        with public_request_scope():
            dadata_logger.info("dadata suggest: %r → %d вариантов", "онуфриева 24", 5)
        # Вне публичного контура логи прежние — они нужны для разбора жалоб
        # пилотов на подсказки.
        dadata_logger.info("dadata suggest: %r → %d вариантов", "малышева 51", 3)

    text = "\n".join(r.getMessage() for r in caplog.records)
    assert "онуфриева" not in text.lower(), "адрес анонима попал в журнал"
    assert "малышева" in text.lower(), "редакция протекла на закрытый контур"


def test_public_scope_is_not_leaked_after_request() -> None:
    """Флаг обязан сниматься: иначе первый же публичный запрос заглушил бы логи
    процесса до перезапуска."""
    from app.core.public_request import is_public_request, public_request_scope

    assert is_public_request.get() is False
    with public_request_scope():
        assert is_public_request.get() is True
    assert is_public_request.get() is False


def test_sentry_scrub_drops_public_request_body_and_geocoder_query() -> None:
    """sentry_sdk кладёт в событие полное тело запроса, а `send_default_pii=False`
    этот путь не гейтит (он про куки и IP). Тело публичной ручки — ровно
    `{"q": "<адрес>"}`; брэдкрамб httpx несёт тот же адрес в query."""
    from app.observability.sentry_scrub import scrub_public_address

    event = {
        "request": {
            "url": "https://meraocenka.ru/trade-in/api/public/mera/suggest",
            "data": {"q": "Малышева 30"},
        },
        "breadcrumbs": {
            "values": [
                {
                    "category": "httpx",
                    "data": {"url": "https://nominatim.openstreetmap.org/search?q=Малышева+30"},
                },
                {"category": "httpx", "data": {"url": "https://example.com/x?a=1"}},
            ]
        },
    }

    scrubbed = scrub_public_address(event, {})
    assert scrubbed is not None
    assert "data" not in scrubbed["request"], "тело запроса с адресом уехало в GlitchTip"

    crumbs = scrubbed["breadcrumbs"]["values"]
    assert "Малышева" not in crumbs[0]["data"]["url"], "адрес уехал в брэдкрамбе геокодера"
    # Посторонние URL не трогаем — иначе разбирать чужие ошибки станет нечем.
    assert crumbs[1]["data"]["url"] == "https://example.com/x?a=1"


def test_sentry_scrub_keeps_closed_contour_body() -> None:
    """Редакция узкая: тела запросов закрытого контура нужны для разбора."""
    from app.observability.sentry_scrub import scrub_public_address

    event = {
        "request": {"url": "https://gendsgn.ru/trade-in/api/v1/trade-in/estimate", "data": {"x": 1}}
    }
    scrubbed = scrub_public_address(event, {})
    assert scrubbed is not None
    assert scrubbed["request"]["data"] == {"x": 1}


# ── 6. Бюджет внешнего геокодера ─────────────────────────────────────────────


def test_daily_suggest_budget_protects_shared_geocoder_quota(client: TestClient) -> None:
    """Per-IP окна ограничивают одного клиента, но не сумму: 20/мин с адреса —
    это 28 800 в сутки при общем бесплатном тире DaData в 10 000, ОБЩЕМ с
    закрытым контуром. Без суточного потолка один скрипт оставлял бы без
    подсказок платящих пилотов."""
    assert public_mera._DAILY_SUGGEST_BUDGET < 10_000, (
        "суточный потолок публичных подсказок обязан быть заметно меньше всего "
        "тира геокодера — иначе публичная форма может навредить закрытому контуру"
    )
    assert public_mera._SUGGEST_LIMIT * 60 * 24 > public_mera._DAILY_SUGGEST_BUDGET, (
        "если per-IP лимит сам по себе не может исчерпать суточный бюджет, "
        "потолок бессмысленен — проверь, что тест сторожит реальный сценарий"
    )


# ── 7. Город: то, что предлагаем, и то, на что умеем отвечать ────────────────


def test_offered_cities_match_coverage_cities() -> None:
    """Дропдаун на сайте и списки покрытия — одно множество, записанное дважды.

    Разошлись на проде 16.08.2026: `Серов` предлагался к выбору, но
    отсутствовал в порогах покрытия, поэтому проба резолвила город как
    неизвестный и отвечала «этот адрес вне области, по которой мы собираем
    данные» — про город В ТОЙ ЖЕ области, который мы сами же и предложили.
    В обратную сторону `Берёзовский`, `Среднеуральск` и `Ревда` бэкенд
    обслуживал, а выбрать их было нельзя.

    Шапка `city-registry.ts` этот риск прямо предсказывала («перед добавлением
    7-го города сверить оба списка вручную, теста на это пока нет»). Вот тест.
    """
    import pathlib
    import re

    from app.api.v1.trade_in import COVERAGE_GREEN_CITIES, COVERAGE_YELLOW_CITIES, _fold_city

    registry = (
        pathlib.Path(__file__).resolve().parents[2]
        / "frontend"
        / "src"
        / "lib"
        / "city-registry.ts"
    )
    text = registry.read_text(encoding="utf-8")
    block = re.search(r"OBLAST_CITIES:\s*readonly OblastCity\[\]\s*=\s*\[(.*?)\];", text, re.S)
    assert block, "не нашли OBLAST_CITIES — проверка стала бы пустой"

    offered = {_fold_city(m) for m in re.findall(r'label:\s*"([^"]+)"', block.group(1))}
    answerable = {_fold_city(c) for c in (*COVERAGE_GREEN_CITIES, *COVERAGE_YELLOW_CITIES)}

    assert offered == answerable, (
        "предлагаем выбрать и умеем отвечать — это должно быть одно множество.\n"
        f"  предлагаем, но не отвечаем: {sorted(offered - answerable)}\n"
        f"  отвечаем, но не предлагаем: {sorted(answerable - offered)}"
    )


def test_every_coverage_city_has_a_centroid() -> None:
    """Порог без центроида недостижим: город резолвится по координатам."""
    from app.api.v1.trade_in import (
        _CITY_CENTROIDS_DEG,
        COVERAGE_GREEN_CITIES,
        COVERAGE_YELLOW_CITIES,
        _fold_city,
    )

    centroids = {_fold_city(c) for c in _CITY_CENTROIDS_DEG}
    for city in (*COVERAGE_GREEN_CITIES, *COVERAGE_YELLOW_CITIES):
        assert _fold_city(city) in centroids, (
            f"{city} есть в порогах, но нет в центроидах — по координатам он "
            "никогда не резолвится, и порог мёртвый"
        )


# ── 8. Подсказки слушаются выбранного города ────────────────────────────────


def test_city_goes_into_the_query_itself() -> None:
    """`city_hint` на выдачу подсказок не влияет — его смотрит только
    екатеринбургский кадастровый тир, а DaData ограничена регионом целиком.

    Замер на проде: выбран Серов, введено «Ленина 1» → первой подсказкой
    «Невьянский р-н, пгт Верх-Нейвинский». Человек выбирает верхний вариант и
    считает чужой дом.
    """
    assert public_mera._query_with_city("Ленина 1", "Серов") == "Серов, Ленина 1"

    # Город уже назван — не дублируем («Серов, Серов Ленина 1»).
    assert public_mera._query_with_city("Серов Ленина 1", "Серов") == "Серов Ленина 1"
    assert public_mera._query_with_city("серов ленина 1", "Серов") == "серов ленина 1"

    # ёЁ нормализуются: «Березовский» в тексте и «Берёзовский» в списке — один город.
    assert public_mera._query_with_city("Березовский, Мира 5", "Берёзовский") == (
        "Березовский, Мира 5"
    )

    # Города нет — строка не трогается.
    assert public_mera._query_with_city("Ленина 1", None) == "Ленина 1"


def test_suggest_passes_city_prefixed_query_downstream(client: TestClient) -> None:
    """Проверяем не текст помощника, а то, что ручка реально отдаёт его вниз."""
    captured: dict[str, object] = {}

    async def _fake(**kwargs):
        captured.update(kwargs)
        return SuggestResponse(items=[])

    with patch.object(public_mera, "suggest_addresses", _fake):
        resp = client.post(f"{PREFIX}/suggest", json={"q": "Ленина 1", "city_hint": "Серов"})

    assert resp.status_code == 200, resp.text
    assert captured["q"] == "Серов, Ленина 1"
    # Сам хинт продолжаем передавать: от него зависит гейт кадастрового тира.
    assert captured["city_hint"] == "Серов"