from collections.abc import Iterator from typing import Any import pydantic import pytest from fastapi import Body, FastAPI from pytest_mock import MockerFixture from starlette import status from starlette.testclient import TestClient from campaigns.auth.exceptions import NotAuthenticated, PermissionDenied from campaigns.connectors.facebook.exceptions import FacebookClientError from campaigns.core.exceptions import FieldError, OwsError, ValidationError class SimpleModel(pydantic.BaseModel): field1: str field2: int def raise_exception() -> None: pass def exc_route() -> Any: raise_exception() def validation_exc_route(data: SimpleModel = Body()) -> Any: return data @pytest.fixture def test_client(app: FastAPI) -> Iterator[TestClient]: with TestClient(app, raise_server_exceptions=False) as client: yield client @pytest.fixture(autouse=True) def setup_error_routes(app: FastAPI) -> Iterator[None]: app.debug = False app.add_api_route("/exc", endpoint=exc_route) app.add_api_route( "/exc-validation", methods=["POST"], endpoint=validation_exc_route, ) yield app.debug = True @pytest.mark.parametrize( "exc, status_code, content", [ ( Exception("test"), status.HTTP_500_INTERNAL_SERVER_ERROR, { "code": "internal_error", "message": ( "The server encountered an internal error " "and was unable to complete your request." ), }, ), ( OwsError(), status.HTTP_500_INTERNAL_SERVER_ERROR, { "code": "ows_error", "message": "Ows error", }, ), ( OwsError(message="Some error", code="some_error"), status.HTTP_500_INTERNAL_SERVER_ERROR, { "code": "some_error", "message": "Some error", }, ), ( NotAuthenticated, status.HTTP_401_UNAUTHORIZED, { "code": "authorization_error", "message": "Unauthorized", }, ), ( NotAuthenticated("missing header"), status.HTTP_401_UNAUTHORIZED, { "code": "authorization_error", "message": "missing header", }, ), ( PermissionDenied("not allowed"), status.HTTP_403_FORBIDDEN, { "code": "permission_denied", "message": "not allowed", }, ), ( ValidationError( errors=[ FieldError("field1", message="Invalid field1", code="invalid1"), FieldError("field2", message="Invalid field2", code="invalid2"), ] ), status.HTTP_422_UNPROCESSABLE_ENTITY, { "code": "validation_error", "message": "Validation error", "fields": { "field1": { "code": "invalid1", "message": "Invalid field1", }, "field2": { "code": "invalid2", "message": "Invalid field2", }, }, }, ), ( FacebookClientError(), status.HTTP_400_BAD_REQUEST, { "code": "facebook_client_error", "message": "Facebook client error", "facebookCode": None, "facebookMessage": None, "facebookErrorTitle": None, "facebookErrorMessage": None, "facebookErrorData": None, }, ), ( FacebookClientError(context={"status_code": 422}), status.HTTP_422_UNPROCESSABLE_ENTITY, { "code": "facebook_client_error", "message": "Facebook client error", "facebookCode": None, "facebookMessage": None, "facebookErrorTitle": None, "facebookErrorMessage": None, "facebookErrorData": None, }, ), ( FacebookClientError( context={ "status_code": 422, "fb_error": { "code": 1000, "message": "error", "error_user_msg": "fb error msg", "error_user_title": "fb error title", "error_data": "fb error data", }, } ), status.HTTP_422_UNPROCESSABLE_ENTITY, { "code": "facebook_client_error", "message": "Facebook client error", "facebookCode": 1000, "facebookMessage": "error", "facebookErrorTitle": "fb error title", "facebookErrorMessage": "fb error msg", "facebookErrorData": "fb error data", }, ), ], ) def test_error_handler( exc: Exception | type[Exception], status_code: int, content: Any, test_client: TestClient, mocker: MockerFixture, ) -> None: mocker.patch(f"{__name__}.raise_exception", side_effect=exc) response = test_client.get("/exc") assert response.status_code == status_code assert response.json() == content def test_request_validation_error_handler(test_client: TestClient) -> None: response = test_client.post("/exc-validation", json={"field2": "invalid"}) assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY assert response.json() == { "code": "validation_error", "message": "Validation error", "fields": { "field1": { "code": "value_error.missing", "message": "Field required", }, "field2": { "code": "type_error.integer", "message": "Value is not a valid integer", }, }, }