tests/core/test_errors.py - master - systemsstrategyy/Treker

tests/core/test_errors.py
378 строк15 KB
SystemsStrategy
Синхронизация с актуальной линией разработки (август 2026)
06 авг 2026, 12:52
06 авг 2026, 12:5219af6aa
"""Тесты для api/core/errors.py — DomainException + RFC 9457 handler."""

from __future__ import annotations

import pytest
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from httpx import ASGITransport, AsyncClient
from pydantic import BaseModel, field_validator

from api.core.errors import (
    BusinessRuleViolation,
    DomainException,
    NotFoundError,
    PermissionDenied,
    QuotaExceeded,
    ValidationError,
    domain_exception_handler,
    unhandled_exception_handler,
    validation_exception_handler,
)


def test_hierarchy() -> None:
    for cls in (
        ValidationError,
        NotFoundError,
        PermissionDenied,
        QuotaExceeded,
        BusinessRuleViolation,
    ):
        assert issubclass(cls, DomainException)


def test_codes_unique() -> None:
    codes = [
        cls.code
        for cls in (
            ValidationError,
            NotFoundError,
            PermissionDenied,
            QuotaExceeded,
            BusinessRuleViolation,
        )
    ]
    assert len(codes) == len(set(codes))


def test_status_codes() -> None:
    assert ValidationError.status == 422
    assert NotFoundError.status == 404
    assert PermissionDenied.status == 403
    assert QuotaExceeded.status == 429
    assert BusinessRuleViolation.status == 409


def test_detail_preserved() -> None:
    e = NotFoundError("Карточка не найдена")
    assert e.detail == "Карточка не найдена"
    assert "Карточка не найдена" in str(e)


def test_default_detail_empty() -> None:
    e = ValidationError()
    assert e.detail == ""


@pytest.mark.asyncio
async def test_handler_returns_problem_details() -> None:
    app = FastAPI()
    app.add_exception_handler(DomainException, domain_exception_handler)

    @app.get("/raise-not-found")
    async def raise_not_found() -> None:
        raise NotFoundError("Карточка не найдена")

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/raise-not-found")

    assert response.status_code == 404
    assert response.headers["content-type"] == "application/problem+json"
    body = response.json()
    assert body == {
        "type": "urn:problem:tracker:not_found",
        "title": "Not found",
        "status": 404,
        "detail": "Карточка не найдена",
        "code": "not_found",
    }


@pytest.mark.asyncio
async def test_handler_passes_quota_exceeded() -> None:
    app = FastAPI()
    app.add_exception_handler(DomainException, domain_exception_handler)

    @app.get("/quota")
    async def raise_quota() -> None:
        raise QuotaExceeded("Превышен лимит карточек на тарифе Free")

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/quota")

    assert response.status_code == 429
    body = response.json()
    assert body["code"] == "quota_exceeded"
    assert "Free" in body["detail"]


@pytest.mark.asyncio
async def test_handler_unfolds_detail_payload() -> None:
    """Регрессия #125: subclass'ы (например LimitReached) могут поставить
    `self.detail_payload = {"resource": ..., "current": ..., "max": ...}` —
    эти поля должны попасть в body на одном уровне с type/title/detail,
    чтобы фронт мог строить actionable UI (quota modal с цифрами 1/1)."""

    class _LimitReachedFake(DomainException):
        code = "limit_reached"
        status = 402
        title = "Limit reached"

        def __init__(self, *, detail_payload: dict[str, object]) -> None:
            super().__init__(detail="Превышен лимит")
            self.detail_payload = detail_payload

    app = FastAPI()
    app.add_exception_handler(DomainException, domain_exception_handler)

    @app.get("/limit")
    async def raise_limit() -> None:
        raise _LimitReachedFake(detail_payload={"resource": "spaces", "current": 1, "max": 1})

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/limit")

    assert response.status_code == 402
    body = response.json()
    # Базовые RFC 9457 поля
    assert body["code"] == "limit_reached"
    assert body["detail"] == "Превышен лимит"
    # Unfolded extension fields
    assert body["resource"] == "spaces"
    assert body["current"] == 1
    assert body["max"] == 1


@pytest.mark.asyncio
async def test_handler_detail_payload_cannot_overwrite_reserved_keys() -> None:
    """Защита invariant'а: detail_payload не должен затереть базовые
    RFC 9457 поля (type/title/status/detail/code) даже если subclass
    случайно положил их в payload."""

    class _BadPayload(DomainException):
        code = "bad"
        status = 400
        title = "Original Title"

        def __init__(self) -> None:
            super().__init__(detail="Original Detail")
            self.detail_payload = {
                "code": "OVERWRITTEN",  # должно быть проигнорировано
                "detail": "OVERWRITTEN",
                "extra_field": "preserved",
            }

    app = FastAPI()
    app.add_exception_handler(DomainException, domain_exception_handler)

    @app.get("/bad")
    async def raise_bad() -> None:
        raise _BadPayload()

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/bad")

    body = response.json()
    # Базовые поля — оригинальные, не overwritten
    assert body["code"] == "bad"
    assert body["detail"] == "Original Detail"
    # Не-reserved extension прошёл
    assert body["extra_field"] == "preserved"


# ── RequestValidationError handler (issue #140) ──────────────────────────────


class _LoginIn(BaseModel):
    email: str
    password: str


def _build_validation_app() -> FastAPI:
    """Минимальное приложение с registered validation handler.

    Используется в тестах ниже чтобы изолировать handler-логику от main.py
    (там много middleware'а который мешает unit-тестам)."""
    app = FastAPI()
    app.add_exception_handler(RequestValidationError, validation_exception_handler)

    @app.post("/login")
    async def login(payload: _LoginIn) -> dict[str, str]:
        return {"email": payload.email}

    return app


@pytest.mark.asyncio
async def test_validation_handler_returns_problem_json_content_type() -> None:
    """RFC 9457 §3 требует `application/problem+json`. Дефолтный FastAPI
    handler возвращает `application/json` — schemathesis ловит как violation."""
    app = _build_validation_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/login", json={})  # missing both fields

    assert response.status_code == 422
    assert response.headers["content-type"] == "application/problem+json"


@pytest.mark.asyncio
async def test_validation_handler_body_shape_rfc9457() -> None:
    """Body должен содержать обязательные RFC 9457 поля + наш `code` + полный
    `errors` для frontend подсветки конкретных полей."""
    app = _build_validation_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/login", json={"email": "a@b.com"})  # missing password

    body = response.json()
    assert body["type"] == "urn:problem:tracker:validation_failed"
    assert body["title"] == "Validation Failed"
    assert body["status"] == 422
    assert body["code"] == "validation_failed"
    # `detail` — человекочитаемая агрегация первой ошибки
    assert "password" in body["detail"]
    # `errors` — полный Pydantic-list для frontend
    assert isinstance(body["errors"], list)
    assert any(e["loc"][-1] == "password" for e in body["errors"])


@pytest.mark.asyncio
async def test_validation_handler_detail_includes_field_path() -> None:
    """`detail` собирается из `loc` (минус первый элемент-кат: body/query/...)
    и `msg`. Frontend toast: «password: Field required»."""
    app = _build_validation_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/login", json={"email": "a@b.com"})

    body = response.json()
    assert "password" in body["detail"]
    assert "required" in body["detail"].lower()


class _ConsentIn(BaseModel):
    """Мини-схема, воспроизводящая `ConsentMixin` (api/modules/auth/schemas.py):
    `field_validator` с `raise ValueError` заставляет Pydantic v2 положить сам
    объект `ValueError` в `ctx['error']` каждой ошибки."""

    consent_accepted: bool

    @field_validator("consent_accepted")
    @classmethod
    def _must_be_true(cls, v: bool) -> bool:
        if v is not True:
            raise ValueError("Требуется согласие на обработку персональных данных")
        return v


@pytest.mark.asyncio
async def test_validation_handler_serializes_value_error_ctx() -> None:
    """Регресс: `field_validator` с `raise ValueError` кладёт несериализуемый
    объект `ValueError` в `ctx['error']`. Сырой `exc.errors()` в payload валил
    `JSONResponse` в 500 (`TypeError: ValueError is not JSON serializable`) —
    хотя пользователь должен получать понятный 422. Ломало 152-ФЗ-сценарий:
    register с `consent_accepted=false` отдавал 500 вместо validation_failed."""
    app = FastAPI()
    app.add_exception_handler(RequestValidationError, validation_exception_handler)

    @app.post("/consent")
    async def consent(payload: _ConsentIn) -> dict[str, bool]:
        return {"ok": payload.consent_accepted}

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/consent", json={"consent_accepted": False})

    assert response.status_code == 422, response.text
    assert response.headers["content-type"] == "application/problem+json"
    body = response.json()
    assert body["code"] == "validation_failed"
    # Сообщение из ValueError доезжает до клиента (не теряется и не валит сериализацию).
    assert "согласие" in body["detail"].lower()
    assert isinstance(body["errors"], list) and body["errors"]


@pytest.mark.asyncio
async def test_validation_handler_works_under_slowapi_limiter() -> None:
    """Проверка гипотезы «slowapi ломает 422»: эндпоинт под `@limiter.limit`
    с field_validator, кидающим ValueError, ВСЁ РАВНО должен отдавать 422,
    а не 500. slowapi-обёртка (functools.wraps пробрасывает сигнатуру) не
    мешает RequestValidationError дойти до кастомного handler'а — root cause
    500 на проде был в недеплоенном jsonable_encoder-фиксе, а не в slowapi."""
    from slowapi import Limiter
    from slowapi.util import get_remote_address

    # enabled=False: обёртка применяется (тестируем slowapi-сигнатуру), но
    # сам rate-check пропускается — Redis-стор не нужен.
    test_limiter = Limiter(key_func=get_remote_address, enabled=False)
    app = FastAPI()
    app.state.limiter = test_limiter
    app.add_exception_handler(RequestValidationError, validation_exception_handler)

    @app.post("/consent-limited")
    @test_limiter.limit("10/hour")
    async def consent_limited(request: Request, payload: _ConsentIn) -> dict[str, bool]:
        return {"ok": payload.consent_accepted}

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/consent-limited", json={"consent_accepted": False})

    assert response.status_code == 422, response.text
    body = response.json()
    assert body["code"] == "validation_failed"
    assert "согласие" in body["detail"].lower()


# ── Catch-all unhandled_exception_handler (issue #140 §3) ────────────────────


@pytest.mark.asyncio
async def test_unhandled_exception_handler_returns_problem_json() -> None:
    """Generic 500 должен быть `application/problem+json` (не text/plain) —
    schemathesis ловит дефолтный FastAPI «Internal Server Error» как
    Undocumented Content-Type."""
    app = FastAPI()
    app.add_exception_handler(Exception, unhandled_exception_handler)

    @app.get("/boom")
    async def boom() -> None:
        raise RuntimeError("неожиданная ошибка")

    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/boom")

    assert response.status_code == 500
    assert response.headers["content-type"] == "application/problem+json"


@pytest.mark.asyncio
async def test_unhandled_exception_handler_no_stacktrace_leak() -> None:
    """Body не должен содержать `detail` со stacktrace'ом — атакующий не
    должен видеть internals при probing endpoint'а."""
    app = FastAPI()
    app.add_exception_handler(Exception, unhandled_exception_handler)

    @app.get("/secret-boom")
    async def boom() -> None:
        raise RuntimeError("DATABASE_URL=postgresql://admin:secret@db/prod")

    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/secret-boom")

    body = response.json()
    assert body["status"] == 500
    assert body["code"] == "internal_error"
    # `detail` пустой — leak'а нет.
    assert body["detail"] == ""
    # И стектрейс не появится в любом другом поле.
    assert "DATABASE_URL" not in str(body)
    assert "admin:secret" not in str(body)