"""Публичный 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 datetime import UTC, datetime 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.http_errors import install_validation_error_handler # 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() public_mera._stats_limiter._hits.clear() public_mera._showcase_limiter._hits.clear() yield public_mera._suggest_limiter._hits.clear() public_mera._coverage_limiter._hits.clear() public_mera._stats_limiter._hits.clear() public_mera._showcase_limiter._hits.clear() @pytest.fixture() def client() -> TestClient: """Приложение с РЕАЛЬНЫМ rbac_guard — тем же, что регистрирует app/main.py. Рядом с публичным роутером намеренно висит закрытый маршрут-двойник: он и делает проверку 3 (узость исключения) возможной внутри одного приложения. """ app = FastAPI() app.middleware("http")(rbac_guard) # Тот же обработчик 422, что вешает app/main.py. Без него тестовое # приложение отвечало бы на нефинитные числа иначе, чем прод, и проверка # «422, а не 500» была бы зелёной по построению. install_validation_error_handler(app) 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_the_declared_routes() -> None: paths = {r.path for r in public_mera.router.routes} assert paths == { "/suggest", "/coverage", "/stats", "/showcase", "/estimate", "/estimate/read", }, ( "изменился набор публичных (анонимных) ручек МЕРЫ. Это не рефакторинг: " "всё под /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": []} _SHOWCASE_ROW = { "district": None, "rooms": 2, "area_m2": 54.0, "floor": 5, "total_floors": None, "deal_quarter": "II квартал 2026", "predicted_rub": 6_100_000, "fact_rub": 5_900_000, "err_pct": 3.39, "n_analogs": 41, "note": "не point-in-time", "street_name": "улица Краснолесья", "street_scheme": { "street": "улица Краснолесья", "w": 1000, "h": 1000, "target": ["M10 20L30 40"], "roads": [{"c": "primary", "d": "M0 0L100 100"}], "water": [], "labels": [{"t": "улица Чкалова", "x": 431.2, "y": 88.0}], }, # Центроид улицы, не дом (миграция 280). ЕКБ: широта ~56.8, долгота ~60.6 — # величины намеренно непохожие, чтобы перестановка ловилась по значению. "lat": 56.8386, "lon": 60.6055, } _SHOWCASE_RUN = { "computed_at": datetime(2026, 8, 29, 10, 0, tzinfo=UTC), "considered": 200, "priced": 173, "no_prediction": 27, "incomplete": 4, "eligible": 169, "written": 20, "with_district": 18, "rejection_rule": "данных нет: нет прогноза / квартала / площади", } def _showcase_db(run: dict | None = _SHOWCASE_RUN, rows: list | None = None) -> MagicMock: db = MagicMock() chain = db.execute.return_value.mappings.return_value chain.first.return_value = run chain.all.return_value = [_SHOWCASE_ROW] if rows is None else rows return db def test_anonymous_gets_showcase(client: TestClient) -> None: """Витрина открыта анониму и отдаёт то, что лежит в таблице. Пустое поле района проходит НАСКВОЗЬ как null: витрина не имеет права подставить правдоподобный район там, где его не удалось определить. """ client.app.dependency_overrides[get_db] = lambda: _showcase_db() resp = client.get(f"{PREFIX}/showcase") assert resp.status_code == 200, resp.text body = resp.json() assert body["computed_at"].startswith("2026-08-29T10:00") assert body["deals"][0]["district"] is None assert body["deals"][0]["fact_rub"] == 5_900_000 # Адреса в контракте ручки нет вовсе — в `deals` дом известен у 2.7% строк. assert "address" not in body["deals"][0] # Схема улицы едет готовыми SVG-путями, а не геометрией. assert body["deals"][0]["street_name"] == "улица Краснолесья" assert body["deals"][0]["street_scheme"]["target"] == ["M10 20L30 40"] def test_showcase_row_without_street_stays_null(client: TestClient) -> None: """Несматчившаяся улица — 8% сделок — проходит насквозь как null. Ни подстановки соседней улицы, ни падения: фронт по этому null показывает район. Ломать так: сделать street_name/street_scheme обязательными в `ShowcaseDeal` — тест покраснеет на 500. """ row = {**_SHOWCASE_ROW, "street_name": None, "street_scheme": None} client.app.dependency_overrides[get_db] = lambda: _showcase_db(rows=[row]) body = client.get(f"{PREFIX}/showcase").json() assert body["deals"][0]["street_name"] is None assert body["deals"][0]["street_scheme"] is None def test_showcase_carries_coords_without_swapping_them(client: TestClient) -> None: """Координата доезжает до JSON и не переставлена местами. Обе величины float, поэтому перестановку не поймает ни схема, ни тип — только значение: 56.8 (широта ЕКБ) против 60.6 (долгота). Ломать так: поменять местами `lat=` и `lon=` в сборке `ShowcaseDeal` — покраснеет здесь. Точка при этом — ЦЕНТРОИД УЛИЦЫ, а не дом (991 различная координата на 34 017 сделок выборки); ровно это написано в `note`, который едет рядом. """ client.app.dependency_overrides[get_db] = lambda: _showcase_db() deal = client.get(f"{PREFIX}/showcase").json()["deals"][0] assert deal["lat"] == 56.8386, "широта приехала не в то поле" assert deal["lon"] == 60.6055, "долгота приехала не в то поле" def test_showcase_row_without_coords_reaches_the_client_as_null( client: TestClient, ) -> None: """Сделка без координаты доезжает до ответа целиком, с lat=lon=null. Карта переживёт строку без точки; выбрасывание такой сделки было бы отбором по признаку, не связанному с качеством оценки. Подстановка центра города вместо отсутствующей точки — тем более враньё. Ломать так: заменить null на любую подстановку или отфильтровать строку в ручке — покраснеет по значению, а не по форме. """ row = {**_SHOWCASE_ROW, "lat": None, "lon": None} client.app.dependency_overrides[get_db] = lambda: _showcase_db(rows=[row]) deals = client.get(f"{PREFIX}/showcase").json()["deals"] assert len(deals) == 1, "строка без координаты выброшена из витрины" assert deals[0]["lat"] is None and deals[0]["lon"] is None assert deals[0]["fact_rub"] == 5_900_000, "остальные величины строки не пострадали" def test_showcase_carries_counters_so_20_rows_cannot_read_as_all_there_was( client: TestClient, ) -> None: """Счётчики прогона доезжают до фронта, а не остаются в логе бэкенда. Без них «20 отличных строк» неотличимо от «столько и было»: посетитель не может отличить выборку из работы оценщика от её лучшего хвоста. Здесь показано 20 из 169 годных — и оба числа обязаны быть в ответе, вместе с правилом, по которому отсеяно остальное. Ломать так: убрать `stats` из `ShowcaseResponse` (или перестать его заполнять) — тест покраснеет на отсутствующем ключе, а не на форме. """ client.app.dependency_overrides[get_db] = lambda: _showcase_db() body = client.get(f"{PREFIX}/showcase").json() stats = body["stats"] assert stats is not None, "витрина отдаёт строки без счётчиков — подписаться нечем" assert stats["considered"] == 200 assert stats["eligible"] == 169 assert stats["written"] == 20 assert stats["eligible"] - stats["written"] == 149, "не видно, сколько годных не влезло" assert stats["no_prediction"] == 27 assert stats["rejection_rule"], "правило отбраковки не подписано" def test_showcase_without_a_run_shows_nothing_and_says_so(client: TestClient) -> None: """Пересчёта не было — ни строк, ни счётчиков, и это штатный ответ. Обратная сторона предыдущего теста: `stats` не выдумывается там, где прогона не было. Заодно это гейт на связку «строки берутся ИЗ прогона»: строки в таблице есть, но прогона нет — показывать их не из чего. """ client.app.dependency_overrides[get_db] = lambda: _showcase_db(run=None) body = client.get(f"{PREFIX}/showcase").json() assert body == {"computed_at": None, "deals": [], "stats": None} def test_showcase_rate_limited_per_ip(client: TestClient) -> None: db = _showcase_db(rows=[]) client.app.dependency_overrides[get_db] = lambda: db codes = [ client.get(f"{PREFIX}/showcase").status_code for _ in range(public_mera._SHOWCASE_LIMIT + 1) ] assert codes[: public_mera._SHOWCASE_LIMIT] == [200] * public_mera._SHOWCASE_LIMIT assert codes[-1] == 429, f"бюджет витрины не сработал: {codes}" 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"] == "Серов" # ── 9. Невалидный вход: 422 и всё тот же бюджет ────────────────────────────── # # Аудит живого сайта 30.08.2026: POST /coverage с `"lon":1e400` отвечал 500, и # двенадцать таких запросов подряд дали двенадцать пятисоток и ни одного 429. # Две разные поломки в одном месте, поэтому и проверок здесь две. _JSON = {"content-type": "application/json"} # `json.loads` принимает нестандартные литералы Infinity/NaN, а `1e400` — это # переполнение float. Ни одно из этих чисел не сериализуется обратно в JSON, # поэтому они и роняли ответ об ошибке. Поля берём разные намеренно: чинить # должно не поле, а сериализацию перечня ошибок. _NON_FINITE_BODIES = [ b'{"lat":56.838,"lon":1e400,"rooms":2,"area_m2":54.0}', b'{"lat":56.838,"lon":NaN,"rooms":2,"area_m2":54.0}', b'{"lat":Infinity,"lon":60.597,"rooms":2,"area_m2":54.0}', b'{"lat":56.838,"lon":60.597,"rooms":2,"area_m2":-Infinity}', ] @pytest.mark.parametrize("body", _NON_FINITE_BODIES) def test_non_finite_number_is_422_not_500(client: TestClient, body: bytes) -> None: """Infinity/NaN во входе — это невалидный вход, а не отказ сервера. Красный вид этого теста без починки — не «assert 500 != 422», а необработанный ValueError из `json.dumps(..., allow_nan=False)`: он летит сквозь TestClient. Оба исхода одинаково красные и оба про одно: ответ об ошибке не собрался. """ resp = client.post(f"{PREFIX}/coverage", content=body, headers=_JSON) assert resp.status_code == 422, resp.text # Форма ответа остаётся стандартной, иначе фронт разбирает её иначе. assert isinstance(resp.json()["detail"], list) @pytest.mark.parametrize("route", ["/coverage", "/suggest"]) def test_invalid_body_still_spends_the_per_ip_budget(client: TestClient, route: str) -> None: """Бюджет обязан срабатывать РАНЬШЕ разбора тела. Иначе он не защищает ровно от того, что на разборе тела и падает: клиент льёт неограниченный поток отказов с одного адреса. Двусторонность: верните `_enforce(...)` первой строкой тела хендлера — и все ответы станут 422, ни одного 429, тест покраснеет. """ limit = public_mera._COVERAGE_LIMIT if route == "/coverage" else public_mera._SUGGEST_LIMIT codes = [ client.post( f"{PREFIX}{route}", content=b'{"lat":56.838,"lon":1e400}', headers=_JSON ).status_code for _ in range(limit + 3) ] assert 429 in codes, f"бюджет не сработал: {codes}" # Ничего третьего быть не должно — ни 500, ни внезапной 200 на мусоре. assert set(codes) <= {422, 429}, codes # Отказ начинается ровно после исчерпания окна, а не «когда-нибудь». assert codes.index(429) == limit, codes