"""Error handler tests.""" import uuid from fractions import Fraction from typing import Any, Callable import pydantic import pytest from fastapi import FastAPI, status from fastapi.testclient import TestClient from gql.transport.exceptions import TransportError from httpx import HTTPStatusError, Request, RequestError, Response from pydantic import BaseModel, computed_field, model_validator from sqlalchemy.exc import TimeoutError as SQLAlchemyTimeoutError from delivery_metadata.api.error_handlers import DEFAULT_ERROR_MSG from delivery_metadata.exceptions import ( NotFoundException, ProductIneligible, SchemaValidationError, ) pydantic_minor_version = ".".join(pydantic.__version__.split(".")[0:2]) def exc_route() -> None: """Fake handler that only raises an exception.""" raise Exception("something internal to pdp happened!") def product_ineligible_exc_route() -> None: raise ProductIneligible("no metadata for you!") def http_exc_request_route() -> None: raise RequestError("no response", request=Request("GET", "/widgets")) def http_exc_status_route() -> None: raise HTTPStatusError( "bad response", request=Request("GET", "/widgets"), response=Response(401) ) def http_exc_gql_route() -> None: raise TransportError() def sqlalchemy_timeout_route() -> None: raise SQLAlchemyTimeoutError() def pydantic_validation_builtin_exc_route() -> None: class TestModel(BaseModel): name: str TestModel(name=None) # type: ignore[arg-type] def pydantic_validation_custom_exc_route() -> None: class VideoDimensions(BaseModel): width: int height: int @computed_field # type: ignore[prop-decorator] @property def aspect_ratio(self) -> str: aspect_ratio = Fraction(self.width, self.height) return f"{aspect_ratio.numerator}:{aspect_ratio.denominator}" @model_validator(mode="after") def validate_model(self): # type: ignore[no-untyped-def] if self.aspect_ratio != "16:9": raise ValueError(f"Aspect ratio must be 16:9, got {self.aspect_ratio}") return self VideoDimensions(width=4096, height=1716) def test_default_error_handler(app: FastAPI) -> None: """Verify an uncaught exception returns a 500 with expected payload.""" with TestClient(app, raise_server_exceptions=False) as test_client: app.add_api_route("/test_default_error_handler", endpoint=exc_route) response = test_client.get("/test_default_error_handler") assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR assert response.json() == { "code": "internal_error", "message": DEFAULT_ERROR_MSG, } @pytest.mark.parametrize( "mock_function,expected_status", [ (http_exc_status_route, status.HTTP_502_BAD_GATEWAY), (http_exc_request_route, status.HTTP_502_BAD_GATEWAY), (http_exc_gql_route, status.HTTP_502_BAD_GATEWAY), (sqlalchemy_timeout_route, status.HTTP_502_BAD_GATEWAY), ], ) def test_upstream_error_handling( mock_function: Callable[..., Any], expected_status: int, app: FastAPI ) -> None: """Verify an upstream request error returns as expected.""" with TestClient(app, raise_server_exceptions=False) as test_client: url = f"/{uuid.uuid4()}" app.add_api_route(url, endpoint=mock_function) response = test_client.get(url) assert response.status_code == expected_status if expected_status == status.HTTP_502_BAD_GATEWAY: assert response.json() == {} def test_pydantic_validation_error_handler_builtin_error(app: FastAPI) -> None: with TestClient(app, raise_server_exceptions=False) as test_client: app.add_api_route( "/test_pydantic_validation_error_handler_builtin_error", endpoint=pydantic_validation_builtin_exc_route, ) response = test_client.get( "/test_pydantic_validation_error_handler_builtin_error" ) assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT assert response.json() == { "code": "pydantic_validation_error", "message": "TestModel missing value", "schema": "TestModel", "errors": [ { "input": None, "loc": ["name"], "msg": "Input should be a valid string", "type": "string_type", "url": f"https://errors.pydantic.dev/{pydantic_minor_version}/v/string_type", }, ], } def test_pydantic_validation_error_handler_custom_error(app: FastAPI) -> None: with TestClient(app, raise_server_exceptions=False) as test_client: app.add_api_route( "/test_pydantic_validation_error_handler_custom_error", endpoint=pydantic_validation_custom_exc_route, ) response = test_client.get( "/test_pydantic_validation_error_handler_custom_error" ) assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT assert response.json() == { "code": "pydantic_validation_error", "message": "VideoDimensions missing value", "schema": "VideoDimensions", "errors": [ { "ctx": {"error": "Aspect ratio must be 16:9, got 1024:429"}, "input": {"width": 4096, "height": 1716}, "loc": [], "msg": "Value error, Aspect ratio must be 16:9, got 1024:429", "type": "value_error", "url": f"https://errors.pydantic.dev/{pydantic_minor_version}/v/value_error", }, ], } SCHEMA_VALIDATION_ERROR_MESSAGE = "Schema validation error - Element '{http://ddex.net/xml/ern/43}NewReleaseMessage': The attribute 'AvsVersionId' is required but missing." XML_CONTENT = "\n\n" def schema_validation_exc_route() -> None: class TestSchemaValidationError(SchemaValidationError): pass raise TestSchemaValidationError( message=SCHEMA_VALIDATION_ERROR_MESSAGE, xml_content=XML_CONTENT, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) @pytest.mark.parametrize( ("return_xml", "expected_response"), [ ( "return_xml=true", { "code": "TestSchemaValidationError", "message": SCHEMA_VALIDATION_ERROR_MESSAGE + "\n\n" + XML_CONTENT, }, ), ( "return_xml=false", { "code": "TestSchemaValidationError", "message": SCHEMA_VALIDATION_ERROR_MESSAGE, }, ), ( "", { "code": "TestSchemaValidationError", "message": SCHEMA_VALIDATION_ERROR_MESSAGE, }, ), ], ) def test_schema_validation_error_handler( return_xml: str, expected_response: str, app: FastAPI ) -> None: """Verify schema validation error returns a 500 code with expected payload.""" with TestClient(app, raise_server_exceptions=False) as test_client: app.add_api_route( "/test_schema_validation_error_handler", endpoint=schema_validation_exc_route, ) response = test_client.get( f"/test_schema_validation_error_handler?{return_xml}" ) assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR assert response.json() == expected_response def http_not_found_route() -> None: class TestNotFoundError(NotFoundException): pass raise TestNotFoundError("Not found") def test_not_found_error_handler(app: FastAPI) -> None: """Verify a 404 not found exception with expected payload.""" with TestClient(app, raise_server_exceptions=False) as test_client: app.add_api_route( "/test_not_found_error_handler", endpoint=http_not_found_route ) response = test_client.get("/test_not_found_error_handler") assert response.status_code == status.HTTP_404_NOT_FOUND assert response.json() == { "code": "TestNotFoundError", "message": "Not found", } def test_product_ineligible_error_handler(app: FastAPI) -> None: with TestClient(app, raise_server_exceptions=False) as test_client: app.add_api_route( "/test_product_ineligible_error_handler", endpoint=product_ineligible_exc_route, ) response = test_client.get("/test_product_ineligible_error_handler") assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT assert response.json() == { "code": "product_ineligible", "message": "no metadata for you!", }