fix: tests

Author: rubenfonseca

aws_lambda_powertools/event_handler/api_gateway.py

@@ -210,7 +210,7 @@ def __init__(
         self,
         status_code: int,
         content_type: Optional[str] = None,
-        body: Union[str, bytes, None] = None,
+        body: Union[Any, None] = None,
         headers: Optional[Dict[str, Union[str, List[str]]]] = None,
         cookies: Optional[List[Cookie]] = None,
         compress: Optional[bool] = None,
@@ -1317,7 +1317,7 @@ def __init__(
         self._strip_prefixes = strip_prefixes
         self.context: Dict = {}  # early init as customers might add context before event resolution
         self.processed_stack_frames = []
-        self.response_builder_class = ResponseBuilder
+        self._response_builder_class = ResponseBuilder
 
         # Allow for a custom serializer or a concise json serialization
         self._serializer = serializer or partial(json.dumps, separators=(",", ":"), cls=Encoder)
@@ -1875,9 +1875,9 @@ def _not_found(self, method: str) -> ResponseBuilder:
 
         handler = self._lookup_exception_handler(NotFoundError)
         if handler:
-            return self.response_builder_class(handler(NotFoundError()))
+            return self._response_builder_class(handler(NotFoundError()))
 
-        return self.response_builder_class(
+        return self._response_builder_class(
             Response(
                 status_code=HTTPStatus.NOT_FOUND.value,
                 content_type=content_types.APPLICATION_JSON,
@@ -1892,7 +1892,7 @@ def _call_route(self, route: Route, route_arguments: Dict[str, str]) -> Response
             # Reset Processed stack for Middleware (for debugging purposes)
             self._reset_processed_stack()
 
-            return self.response_builder_class(
+            return self._response_builder_class(
                 self._to_response(
                     route(router_middlewares=self._router_middlewares, app=self, route_arguments=route_arguments),
                 ),
@@ -1909,7 +1909,7 @@ def _call_route(self, route: Route, route_arguments: Dict[str, str]) -> Response
                 # If the user has turned on debug mode,
                 # we'll let the original exception propagate, so
                 # they get more information about what went wrong.
-                return self.response_builder_class(
+                return self._response_builder_class(
                     Response(
                         status_code=500,
                         content_type=content_types.TEXT_PLAIN,
@@ -1948,12 +1948,12 @@ def _call_exception_handler(self, exp: Exception, route: Route) -> Optional[Resp
         handler = self._lookup_exception_handler(type(exp))
         if handler:
             try:
-                return self.response_builder_class(handler(exp), route)
+                return self._response_builder_class(handler(exp), route)
             except ServiceError as service_error:
                 exp = service_error
 
         if isinstance(exp, ServiceError):
-            return self.response_builder_class(
+            return self._response_builder_class(
                 Response(
                     status_code=exp.status_code,
                     content_type=content_types.APPLICATION_JSON,

aws_lambda_powertools/event_handler/bedrock_agent.py

@@ -1,6 +1,8 @@
 import logging
 from typing import Any, Dict, Optional, cast
 
+from typing_extensions import override
+
 from aws_lambda_powertools.event_handler import ApiGatewayResolver
 from aws_lambda_powertools.event_handler.api_gateway import CORSConfig, ProxyEventType, ResponseBuilder
 from aws_lambda_powertools.utilities.data_classes import BedrockAgentEvent
@@ -10,6 +12,7 @@
 
 
 class BedrockResponseBuilder(ResponseBuilder):
+    @override
     def build(self, event: BaseProxyEvent, cors: Optional[CORSConfig] = None) -> Dict[str, Any]:
         """Build the full response dict to be returned by the lambda"""
         self._route(event, cors)
@@ -68,4 +71,4 @@ def __init__(self, debug: bool = False, enable_validation: bool = True):
             None,
             enable_validation,
         )
-        self.response_builder_class = BedrockResponseBuilder
+        self._response_builder_class = BedrockResponseBuilder

tests/functional/event_handler/test_bedrock_agent.py

@@ -30,13 +30,13 @@ def claims() -> Dict[str, Any]:
     assert result["response"]["httpStatusCode"] == 200
 
     body = result["response"]["responseBody"]["application/json"]["body"]
-    assert body == {"output": claims_response}
+    assert body == json.dumps({"output": claims_response})
 
 
 def test_bedrock_agent_event_with_response():
     # GIVEN a Bedrock Agent event
     app = BedrockAgentResolver()
-    output = json.dumps({"output": claims_response})
+    output = {"output": claims_response}
 
     @app.get("/claims")
     def claims():
@@ -56,7 +56,7 @@ def claims():
     assert result["response"]["httpStatusCode"] == 200
 
     body = result["response"]["responseBody"]["application/json"]["body"]
-    assert body == output
+    assert body == json.dumps(output)
 
 
 def test_bedrock_agent_event_with_no_matches():