Bug(HttpResolverLocal): OpenAPIResponseValidationMiddleware is not working as intended

[amin-farjadi]

Expected Behaviour

The following tests to pass:

"""
The imports and utility functions used can found in:
tests/functional/event_handler/_pydantic/test_http_resolver_pydantic.py
"""

from __future__ import annotations

import asyncio
from typing import Any

import pytest
from pydantic import BaseModel, Field

from aws_lambda_powertools.event_handler import HttpResolverLocal
from aws_lambda_powertools.event_handler.middlewares.openapi_validation import OpenAPIResponseValidationMiddleware


def make_asgi_receive(body: bytes = b""):
    """Create an ASGI receive callable."""

    async def receive() -> dict[str, Any]:
        await asyncio.sleep(0)
        return {"type": "http.request", "body": body, "more_body": False}

    return receive


def make_asgi_send():
    """Create an ASGI send callable that captures response."""
    captured: dict[str, Any] = {"status_code": None, "body": b""}

    async def send(message: dict[str, Any]) -> None:
        await asyncio.sleep(0)
        if message["type"] == "http.response.start":
            captured["status_code"] = message["status"]
        elif message["type"] == "http.response.body":
            captured["body"] = message["body"]

    return send, captured


class UserModel(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    age: int = Field(ge=0, le=150)
    email: str | None = None


class UserResponse(BaseModel):
    id: str
    user: UserModel
    created: bool = True


@pytest.mark.asyncio
async def test_openapi_middleware_async_handler_with_validation(mocker):
    # GIVEN an app with async handler and validation
    app = HttpResolverLocal(enable_validation=True)

    # WHEN a handler for a route is defined with a return type as Pydantic model
    @app.post("/users")
    async def create_user(user: UserModel) -> UserResponse:
        await asyncio.sleep(0.001)
        return {"id": "async-123","name": "AsyncUser", "age": 25}

    scope = {
        "type": "http",
        "method": "POST",
        "path": "/users",
        "query_string": b"",
        "headers": [(b"content-type", b"application/json")],
    }

    receive = make_asgi_receive(b'{"name": "AsyncUser", "age": 25}')
    send, captured = make_asgi_send()

    spy = mocker.spy(OpenAPIResponseValidationMiddleware, "_handle_response")
    # WHEN called via ASGI interface
    await app(scope, receive, send)
    response_passed_to_openapi_validation = spy.call_args.kwargs["response"]

    # THEN the response of the route is expected to be passed onto the OpenAPI response validation middleware
    spy.assert_called_once()
    assert response_passed_to_openapi_validation.body == '{"id": "async-123", "name": "AsyncUser", "age": 25}'


@pytest.mark.asyncio
async def test_validate_invalid_return_model_http_resolver_local():
    # GIVEN an app with async handler and validation
    app = HttpResolverLocal(enable_validation=True)

    class Model(BaseModel):
        name: str
        age: int

    # WHEN a handler for a route is defined with a return type as Pydantic model but, returning data of different type
    @app.get("/user")
    async def handler() -> Model:
        await asyncio.sleep(0.001)
        return {"name": "John"}  # type: ignore

    scope = {
        "type": "http",
        "method": "GET",
        "path": "/user",
        "query_string": b"",
        "headers": [(b"content-type", b"application/json")],
    }

    receive = make_asgi_receive()
    send, captured = make_asgi_send()

    # WHEN called via ASGI interface
    await app(scope, receive, send)

    # THEN the response of the route is expected to be 422 Unprocessable Entity
    assert captured["status_code"] == 422
    assert "missing" in captured["body"]


if __name__ == "__main__":
    pytest.main(["-v", __file__])

Current Behaviour

FAILED tests/functional/event_handler/_pydantic/standalone.py::test_openapi_middleware_async_handler_with_validation - assert '' == '{"id": "async-123", "name": "AsyncUser", "age": 25}'
  
  - {"name": "AsyncUser", "age": 25}
FAILED tests/functional/event_handler/_pydantic/standalone.py::test_validate_invalid_return_model_http_resolver_local - assert <HTTPStatus.OK: 200> == 422

Code snippet

See code in Expected Behaviour

Possible Solution

I might have a fix for this, happy to contribute

Steps to Reproduce

copy and paste the tests into a new file at: tests/functional/event_handler/_pydantic/standalone.py and run python ./tests/functional/event_handler/_pydantic/standalone.py

Powertools for AWS Lambda (Python) version

latest

AWS Lambda function runtime

3.12

Packaging format used

PyPi

Debugging logs