")
+ def get_path(p: Annotated[str_field, Path()]):
+ return {"value": p}
+
+ @app.get("/query")
+ def get_query(q: Annotated[str_field, Query()]):
+ return {"value": q}
+
+ @app.post("/body")
+ def post_body(b: Annotated[str_field, Body()]):
+ return {"value": b}
+
+ del gw_event["multiValueHeaders"]
+ del gw_event["multiValueQueryStringParameters"]
+
+ # Header
+ gw_event["path"] = "/header"
+ gw_event["httpMethod"] = "GET"
+ gw_event["headers"] = {"h": "test"}
+ assert app(gw_event, {})["statusCode"] == 200
+
+ # Path
+ gw_event["path"] = "/path/test"
+ gw_event["pathParameters"] = {"p": "test"}
+ assert app(gw_event, {})["statusCode"] == 200
+
+ # Query
+ gw_event["path"] = "/query"
+ gw_event["pathParameters"] = None
+ gw_event["queryStringParameters"] = {"q": "test"}
+ assert app(gw_event, {})["statusCode"] == 200
+
+ # Body
+ gw_event["path"] = "/body"
+ gw_event["httpMethod"] = "POST"
+ gw_event["headers"]["content-type"] = "application/json"
+ gw_event["body"] = '"test"'
+ assert app(gw_event, {})["statusCode"] == 200
+
+
+def test_field_constraints_apply_with_param_type(gw_event):
+ """Constraints declared on a Field are enforced when paired with a location marker."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/items")
+ def get_items(quantity: Annotated[int, Field(gt=0), Query()]):
+ return {"quantity": quantity}
+
+ gw_event["path"] = "/items"
+ gw_event["httpMethod"] = "GET"
+
+ # Passes the gt=0 constraint
+ gw_event["queryStringParameters"] = {"quantity": "5"}
+ assert app(gw_event, {})["statusCode"] == 200
+
+ # Violates gt=0
+ gw_event["queryStringParameters"] = {"quantity": "-1"}
+ assert app(gw_event, {})["statusCode"] == 422
+
+
def test_validate_pydantic_query_params_with_config_dict_and_validators(gw_event):
"""Test that Pydantic models with ConfigDict, aliases, and validators work correctly"""
@@ -2833,3 +2935,1423 @@ def handler(query_dt: datetime.datetime):
# THEN validation should fail because the encoded string is not a valid datetime
result = app(raw_event, {})
assert result["statusCode"] == 422
+
+
+def test_validate_union_single_or_list_body_with_list(gw_event):
+ """Test that Union[Model, List[Model]] correctly handles a list of items"""
+ # GIVEN an APIGatewayRestResolver with validation enabled
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ class Item(BaseModel):
+ name: str
+ value: int
+
+ # WHEN a handler is defined with Union[Model, List[Model]] body parameter
+ @app.post("/items")
+ def handler(items: Annotated[Union[Item, List[Item]], Body()]) -> Dict[str, Any]:
+ # Should receive the full list, not just the first element
+ if isinstance(items, list):
+ return {"count": len(items), "items": [item.model_dump() for item in items]}
+ else:
+ return {"count": 1, "items": [items.model_dump()]}
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/items"
+ # Send a list of items
+ gw_event["body"] = json.dumps(
+ [
+ {"name": "item1", "value": 10},
+ {"name": "item2", "value": 20},
+ {"name": "item3", "value": 30},
+ ],
+ )
+
+ # THEN the handler should receive all items in the list, not just the first one
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["count"] == 3
+ assert len(body["items"]) == 3
+ assert body["items"][0]["name"] == "item1"
+ assert body["items"][1]["name"] == "item2"
+ assert body["items"][2]["name"] == "item3"
+
+
+def test_validate_union_single_or_list_body_with_single(gw_event):
+ """Test that Union[Model, List[Model]] correctly handles a single item"""
+ # GIVEN an APIGatewayRestResolver with validation enabled
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ class Item(BaseModel):
+ name: str
+ value: int
+
+ # WHEN a handler is defined with Union[Model, List[Model]] body parameter
+ @app.post("/items")
+ def handler(items: Annotated[Union[Item, List[Item]], Body()]) -> Dict[str, Any]:
+ if isinstance(items, list):
+ return {"count": len(items), "items": [item.model_dump() for item in items]}
+ else:
+ return {"count": 1, "items": [items.model_dump()]}
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/items"
+ # Send a single item
+ gw_event["body"] = json.dumps({"name": "single_item", "value": 42})
+
+ # THEN the handler should receive the single item
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["count"] == 1
+ assert len(body["items"]) == 1
+ assert body["items"][0]["name"] == "single_item"
+ assert body["items"][0]["value"] == 42
+
+
+def test_validate_rootmodel_list_body(gw_event):
+ """Test that RootModel[List[Model]] correctly handles a list of items"""
+ # GIVEN an APIGatewayRestResolver with validation enabled
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ class Item(BaseModel):
+ name: str
+ value: int
+
+ class ItemCollection(RootModel[List[Item]]):
+ root: List[Item]
+
+ # WHEN a handler is defined with RootModel[List[Model]] body parameter
+ @app.post("/items")
+ def handler(collection: Annotated[ItemCollection, Body()]) -> Dict[str, Any]:
+ # collection.root should contain the full list
+ items = collection.root
+ return {"count": len(items), "items": [item.model_dump() for item in items]}
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/items"
+ # Send a list of items
+ gw_event["body"] = json.dumps(
+ [
+ {"name": "item1", "value": 100},
+ {"name": "item2", "value": 200},
+ ],
+ )
+
+ # THEN the handler should receive all items in the collection
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["count"] == 2
+ assert len(body["items"]) == 2
+ assert body["items"][0]["name"] == "item1"
+ assert body["items"][0]["value"] == 100
+ assert body["items"][1]["name"] == "item2"
+ assert body["items"][1]["value"] == 200
+
+
+def test_validate_nested_union_with_sequence(gw_event):
+ """Test that nested Union types containing sequences are handled correctly"""
+ # GIVEN an APIGatewayRestResolver with validation enabled
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ class Person(BaseModel):
+ name: str
+ age: int
+
+ # WHEN a handler is defined with a complex Union including List
+ @app.post("/people")
+ def handler(
+ data: Annotated[Union[str, List[Person], Person], Body()],
+ ) -> Dict[str, Any]:
+ if isinstance(data, str):
+ return {"type": "string", "value": data}
+ elif isinstance(data, list):
+ return {"type": "list", "count": len(data)}
+ else:
+ return {"type": "person", "name": data.name}
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/people"
+ # Send a list
+ gw_event["body"] = json.dumps(
+ [
+ {"name": "Alice", "age": 30},
+ {"name": "Bob", "age": 25},
+ ],
+ )
+
+ # THEN the handler should receive the full list
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["type"] == "list"
+ assert body["count"] == 2
+
+
+# ────────────────────────────────────────────────────────────────────
+# Regression tests for Union / RootModel / Optional sequence body
+# See: https://github.com/aws-powertools/powertools-lambda-python/issues/8057
+# ────────────────────────────────────────────────────────────────────
+
+
+class _Item(BaseModel):
+ name: str
+ value: int
+
+
+class _ItemCollection(RootModel[List[_Item]]):
+ pass
+
+
+_THREE_ITEMS = [
+ {"name": "a", "value": 1},
+ {"name": "b", "value": 2},
+ {"name": "c", "value": 3},
+]
+
+
+def _post_json(app, path, payload):
+ """Helper: build a minimal APIGW REST event, POST JSON, return parsed result."""
+ from tests.functional.utils import load_event
+
+ event = load_event("apiGatewayProxyEvent.json")
+ event["httpMethod"] = "POST"
+ event["path"] = path
+ event["body"] = json.dumps(payload)
+ result = app(event, {})
+ return result["statusCode"], json.loads(result["body"])
+
+
+# ---------- List[Model] | None ----------
+
+
+def test_optional_list_body_with_list():
+ """List[Model] | None must preserve the full list."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[List[_Item] | None, Body()]) -> Dict[str, Any]:
+ assert isinstance(items, list)
+ return {"count": len(items)}
+
+ status, body = _post_json(app, "/items", _THREE_ITEMS)
+ assert status == 200
+ assert body["count"] == 3
+
+
+def test_optional_list_body_with_none():
+ """List[Model] | None must accept a null body gracefully."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[List[_Item] | None, Body()] = None) -> Dict[str, Any]:
+ return {"received_none": items is None}
+
+ status, body = _post_json(app, "/items", None)
+ assert status == 200
+ assert body["received_none"] is True
+
+
+# ---------- Union[Model, List[Model]] | None ----------
+
+
+def test_optional_union_model_or_list_with_list():
+ """Union[Model, List[Model]] | None — send list, get full list."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[Union[_Item, List[_Item]] | None, Body()]) -> Dict[str, Any]:
+ assert isinstance(items, list)
+ return {"count": len(items)}
+
+ status, body = _post_json(app, "/items", _THREE_ITEMS)
+ assert status == 200
+ assert body["count"] == 3
+
+
+def test_optional_union_model_or_list_with_single():
+ """Union[Model, List[Model]] | None — send single obj, get single obj."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[Union[_Item, List[_Item]] | None, Body()]) -> Dict[str, Any]:
+ assert not isinstance(items, list)
+ return {"name": items.name}
+
+ status, body = _post_json(app, "/items", {"name": "solo", "value": 99})
+ assert status == 200
+ assert body["name"] == "solo"
+
+
+def test_optional_union_model_or_list_with_none():
+ """Union[Model, List[Model]] | None — send null, get None."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[Union[_Item, List[_Item]] | None, Body()] = None) -> Dict[str, Any]:
+ return {"is_none": items is None}
+
+ status, body = _post_json(app, "/items", None)
+ assert status == 200
+ assert body["is_none"] is True
+
+
+# ---------- List[Model] directly (no Union / Optional) ----------
+
+
+def test_plain_list_body_preserves_all_items():
+ """List[Model] — baseline: must never truncate."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[List[_Item], Body()]) -> Dict[str, Any]:
+ return {"count": len(items)}
+
+ status, body = _post_json(app, "/items", _THREE_ITEMS)
+ assert status == 200
+ assert body["count"] == 3
+
+
+# ---------- Empty list ----------
+
+
+def test_union_model_or_list_with_empty_list():
+ """Union[Model, List[Model]] with [] — must not crash on value[0]."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[Union[_Item, List[_Item]], Body()]) -> Dict[str, Any]:
+ if isinstance(items, list):
+ return {"count": len(items)}
+ return {"count": 1}
+
+ status, body = _post_json(app, "/items", [])
+ assert status == 200
+ assert body["count"] == 0
+
+
+def test_plain_list_with_empty_list():
+ """List[Model] with [] — must accept empty list."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[List[_Item], Body()]) -> Dict[str, Any]:
+ return {"count": len(items)}
+
+ status, body = _post_json(app, "/items", [])
+ assert status == 200
+ assert body["count"] == 0
+
+
+# ---------- Single-element list (boundary) ----------
+
+
+def test_union_model_or_list_with_single_element_list():
+ """Union[Model, List[Model]] with [single_item] — must NOT unwrap to scalar."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[Union[_Item, List[_Item]], Body()]) -> Dict[str, Any]:
+ if isinstance(items, list):
+ return {"type": "list", "count": len(items)}
+ return {"type": "single"}
+
+ status, body = _post_json(app, "/items", [{"name": "only", "value": 1}])
+ assert status == 200
+ # Pydantic may match as single Item or list — either is valid,
+ # but it must NOT crash or lose data
+ assert body.get("count", 1) == 1
+
+
+# ---------- Union with primitive sequences ----------
+
+
+def test_union_str_or_list_dict():
+ """Union[str, List[dict]] — list of dicts must arrive intact."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/data")
+ def handler(data: Annotated[Union[str, List[Dict[str, Any]]], Body()]) -> Dict[str, Any]:
+ if isinstance(data, list):
+ return {"type": "list", "count": len(data)}
+ return {"type": "str"}
+
+ payload = [{"key": "v1"}, {"key": "v2"}]
+ status, body = _post_json(app, "/data", payload)
+ assert status == 200
+ assert body["type"] == "list"
+ assert body["count"] == 2
+
+
+# ---------- RootModel edge cases ----------
+
+
+def test_optional_rootmodel_list_body():
+ """RootModel[List[Model]] | None — list must not be truncated."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[_ItemCollection | None, Body()]) -> Dict[str, Any]:
+ return {"count": len(items.root)}
+
+ status, body = _post_json(app, "/items", _THREE_ITEMS)
+ assert status == 200
+ assert body["count"] == 3
+
+
+def test_union_rootmodel_and_model():
+ """Union[RootModel[List[Model]], Model] — list must not be truncated."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[Union[_ItemCollection, _Item], Body()]) -> Dict[str, Any]:
+ if isinstance(items, _ItemCollection):
+ return {"type": "collection", "count": len(items.root)}
+ return {"type": "single", "name": items.name}
+
+ status, body = _post_json(app, "/items", _THREE_ITEMS)
+ assert status == 200
+ assert body["type"] == "collection"
+ assert body["count"] == 3
+
+
+# ---------- Python 3.10+ pipe Union syntax ----------
+
+
+def test_pipe_union_syntax_model_or_list():
+ """Model | List[Model] (PEP 604 syntax) — list must not be truncated."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[_Item | List[_Item], Body()]) -> Dict[str, Any]: # noqa: FA102
+ if isinstance(items, list):
+ return {"count": len(items)}
+ return {"count": 1}
+
+ status, body = _post_json(app, "/items", _THREE_ITEMS)
+ assert status == 200
+ assert body["count"] == 3
+
+
+def test_pipe_union_optional_list():
+ """List[Model] | None (PEP 604 Optional) — list must not be truncated."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[List[_Item] | None, Body()]) -> Dict[str, Any]: # noqa: FA102
+ if items is None:
+ return {"count": 0}
+ return {"count": len(items)}
+
+ status, body = _post_json(app, "/items", _THREE_ITEMS)
+ assert status == 200
+ assert body["count"] == 3
+
+
+# ---------- Deeply nested: RootModel[Union[Model, List[Model]]] ----------
+
+
+def test_rootmodel_wrapping_union_with_sequence():
+ """RootModel[Union[Model, List[Model]]] — inner Union sequence must be detected."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ class FlexiblePayload(RootModel[Union[_Item, List[_Item]]]):
+ pass
+
+ @app.post("/items")
+ def handler(payload: Annotated[FlexiblePayload, Body()]) -> Dict[str, Any]:
+ data = payload.root
+ if isinstance(data, list):
+ return {"type": "list", "count": len(data)}
+ return {"type": "single", "name": data.name}
+
+ status, body = _post_json(app, "/items", _THREE_ITEMS)
+ assert status == 200
+ assert body["type"] == "list"
+ assert body["count"] == 3
+
+
+# ---------- Multiple resolvers (ALB, HTTP API, etc.) ----------
+
+
+def test_union_list_body_works_across_resolvers():
+ """Regression: ensure fix works for ALB and HTTP API resolvers too."""
+ for ResolverClass in [APIGatewayHttpResolver, ALBResolver]:
+ app = ResolverClass(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[Union[_Item, List[_Item]], Body()]) -> Dict[str, Any]:
+ if isinstance(items, list):
+ return {"count": len(items)}
+ return {"count": 1}
+
+ # Build event appropriate for resolver
+ if ResolverClass is APIGatewayHttpResolver:
+ event = load_event("apiGatewayProxyV2Event.json")
+ event["requestContext"]["http"]["method"] = "POST"
+ event["requestContext"]["http"]["path"] = "/items"
+ event["rawPath"] = "/items"
+ else:
+ event = load_event("albEvent.json")
+ event["httpMethod"] = "POST"
+ event["path"] = "/items"
+
+ event["body"] = json.dumps(_THREE_ITEMS)
+ result = app(event, {})
+ assert result["statusCode"] == 200
+ body_result = json.loads(result["body"])
+ assert body_result["count"] == 3, f"Failed for {ResolverClass.__name__}"
+
+
+# ---------- Large list (stress boundary) ----------
+
+
+def test_union_list_body_large_payload():
+ """Union[Model, List[Model]] with 100 items — no truncation."""
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/items")
+ def handler(items: Annotated[Union[_Item, List[_Item]], Body()]) -> Dict[str, Any]:
+ assert isinstance(items, list)
+ return {"count": len(items)}
+
+ big_payload = [{"name": f"item-{i}", "value": i} for i in range(100)]
+ status, body = _post_json(app, "/items", big_payload)
+ assert status == 200
+ assert body["count"] == 100
+
+
+# ---------- File upload (multipart/form-data) ----------
+
+
+def _build_multipart_body(fields: List[Dict], boundary: str = "----TestBoundary") -> Tuple[str, str]:
+ """
+ Build a multipart/form-data body and return (base64_body, content_type).
+
+ Each field dict can have:
+ - name: field name (required)
+ - value: str or bytes (required)
+ - filename: optional filename (makes it a file part)
+ - content_type: optional content type for the part
+ """
+ parts = []
+ for field in fields:
+ headers = f'Content-Disposition: form-data; name="{field["name"]}"'
+ if "filename" in field:
+ headers += f'; filename="{field["filename"]}"'
+ if "content_type" in field:
+ headers += f"\r\nContent-Type: {field['content_type']}"
+ value = field["value"]
+ if isinstance(value, str):
+ value = value.encode("utf-8")
+ parts.append((headers, value))
+
+ body = b""
+ for headers, value in parts:
+ body += f"--{boundary}\r\n".encode()
+ body += f"{headers}\r\n\r\n".encode()
+ body += value
+ body += b"\r\n"
+ body += f"--{boundary}--\r\n".encode()
+
+ content_type = f"multipart/form-data; boundary={boundary}"
+ return base64.b64encode(body).decode("utf-8"), content_type
+
+
+def test_file_upload_basic(gw_event):
+ """Test basic file upload with File() parameter."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[bytes, File()]):
+ return {"size": len(file_data)}
+
+ body, content_type = _build_multipart_body(
+ [
+ {"name": "file_data", "value": b"hello world", "filename": "test.txt"},
+ ],
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = content_type
+ gw_event["body"] = body
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"size": 11}
+
+
+def test_file_upload_with_form_field(gw_event):
+ """Test file upload mixed with a regular form field."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(
+ description: Annotated[str, Form()],
+ file_data: Annotated[bytes, File()],
+ ):
+ return {"description": description, "size": len(file_data)}
+
+ body, content_type = _build_multipart_body(
+ [
+ {"name": "description", "value": "my file"},
+ {"name": "file_data", "value": b"\x89PNG\r\n\x1a\n", "filename": "image.png", "content_type": "image/png"},
+ ],
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = content_type
+ gw_event["body"] = body
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ parsed = json.loads(result["body"])
+ assert parsed["description"] == "my file"
+ assert parsed["size"] == 8
+
+
+def test_file_upload_missing_required(gw_event):
+ """Test that missing required File() parameter returns 422."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[bytes, File()]):
+ return {"size": len(file_data)}
+
+ # Send empty multipart body (no file_data field)
+ body, content_type = _build_multipart_body(
+ [
+ {"name": "other_field", "value": "some value"},
+ ],
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = content_type
+ gw_event["body"] = body
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 422
+ assert "missing" in result["body"]
+
+
+def test_file_upload_openapi_schema():
+ """Test that File() parameters generate correct OpenAPI schema."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[bytes, File(description="The file to upload")]):
+ return {"size": len(file_data)}
+
+ schema = app.get_openapi_schema()
+ path = schema.paths["/upload"]
+ post_op = path.post
+
+ # Should have a request body with multipart/form-data
+ assert post_op.requestBody is not None
+ content = post_op.requestBody.content
+ assert "multipart/form-data" in content
+
+ # The schema should reference a binary format field
+ multipart_schema = content["multipart/form-data"].schema_
+ assert multipart_schema is not None
+
+
+def test_file_upload_non_base64(gw_event):
+ """Test file upload when body is not base64-encoded (edge case)."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[bytes, File()]):
+ return {"size": len(file_data)}
+
+ # Build multipart body without base64 encoding
+ boundary = "----TestBoundary"
+ raw_body = (
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="test.txt"\r\n'
+ f"\r\n"
+ f"hello world\r\n"
+ f"--{boundary}--\r\n"
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}"
+ gw_event["body"] = raw_body
+ gw_event["isBase64Encoded"] = False
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"size": 11}
+
+
+def test_file_upload_non_base64_emits_warning(gw_event):
+ """Test that non-base64 multipart body emits a warning about API Gateway config."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[bytes, File()]):
+ return {"size": len(file_data)}
+
+ boundary = "----TestBoundary"
+ raw_body = (
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="test.txt"\r\n'
+ f"\r\n"
+ f"hello world\r\n"
+ f"--{boundary}--\r\n"
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}"
+ gw_event["body"] = raw_body
+ gw_event["isBase64Encoded"] = False
+
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ result = app(gw_event, {})
+
+ assert result["statusCode"] == 200
+ assert len(w) == 1
+ assert "Binary Media Types" in str(w[0].message)
+
+
+def test_file_upload_non_base64_binary_content(gw_event):
+ """Test file upload with raw binary bytes (e.g. JPEG) without base64 encoding."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[bytes, File()]):
+ return {"size": len(file_data)}
+
+ # Simulate binary content with bytes that are NOT valid UTF-8 (like JPEG header 0xFF 0xD8)
+ binary_content = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00"
+ boundary = "----TestBoundary"
+ raw_bytes = (
+ (
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="photo.jpg"\r\n'
+ f"Content-Type: image/jpeg\r\n"
+ f"\r\n"
+ ).encode("latin-1")
+ + binary_content
+ + f"\r\n--{boundary}--\r\n".encode("latin-1")
+ )
+
+ # Without binary mode, API Gateway passes body as latin-1 compatible string
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}"
+ gw_event["body"] = raw_bytes.decode("latin-1")
+ gw_event["isBase64Encoded"] = False
+
+ with warnings.catch_warnings(record=True):
+ warnings.simplefilter("always")
+ result = app(gw_event, {})
+
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"size": len(binary_content)}
+
+
+def test_upload_file_with_metadata(gw_event):
+ """Test UploadFile annotation provides filename and content_type."""
+ from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[UploadFile, File()]):
+ return {
+ "filename": file_data.filename,
+ "content_type": file_data.content_type,
+ "size": len(file_data),
+ }
+
+ body, content_type = _build_multipart_body(
+ [
+ {"name": "file_data", "value": b"fake jpeg", "filename": "photo.jpg", "content_type": "image/jpeg"},
+ ],
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = content_type
+ gw_event["body"] = body
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ parsed = json.loads(result["body"])
+ assert parsed["filename"] == "photo.jpg"
+ assert parsed["content_type"] == "image/jpeg"
+ assert parsed["size"] == 9
+
+
+def test_upload_file_mixed_with_form(gw_event):
+ """Test UploadFile + Form fields together."""
+ from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(
+ file_data: Annotated[UploadFile, File()],
+ title: Annotated[str, Form()],
+ ):
+ return {
+ "title": title,
+ "filename": file_data.filename,
+ "size": len(file_data),
+ }
+
+ body, content_type = _build_multipart_body(
+ [
+ {"name": "title", "value": "My Document"},
+ {
+ "name": "file_data",
+ "value": b"pdf content here",
+ "filename": "doc.pdf",
+ "content_type": "application/pdf",
+ },
+ ],
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = content_type
+ gw_event["body"] = body
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ parsed = json.loads(result["body"])
+ assert parsed["title"] == "My Document"
+ assert parsed["filename"] == "doc.pdf"
+ assert parsed["size"] == 16
+
+
+def test_upload_file_openapi_schema():
+ """Test UploadFile generates correct OpenAPI schema."""
+ from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[UploadFile, File(description="A file")]):
+ return {}
+
+ schema = app.get_openapi_schema()
+ schema_dict = schema.model_dump(exclude_none=True, by_alias=True)
+ upload_path = schema_dict["paths"]["/upload"]["post"]
+ content = upload_path["requestBody"]["content"]
+ assert "multipart/form-data" in content
+
+ # Resolve $ref to get the actual schema
+ ref = content["multipart/form-data"]["schema"]["$ref"]
+ schema_name = ref.split("/")[-1]
+ props = schema_dict["components"]["schemas"][schema_name]["properties"]
+ assert props["file_data"]["type"] == "string"
+ assert props["file_data"]["format"] == "binary"
+
+
+def test_multipart_missing_boundary(gw_event):
+ """Test that missing boundary in content-type raises ValueError."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[bytes, File()]):
+ return {"size": len(file_data)}
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = "multipart/form-data" # no boundary
+ gw_event["body"] = base64.b64encode(b"some data").decode()
+ gw_event["isBase64Encoded"] = True
+
+ with pytest.raises(ValueError, match="Missing boundary"):
+ app(gw_event, {})
+
+
+def test_multipart_quoted_boundary(gw_event):
+ """Test that boundary with quotes is parsed correctly."""
+ from aws_lambda_powertools.event_handler.openapi.params import File
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[bytes, File()]):
+ return {"size": len(file_data)}
+
+ boundary = "----TestBoundary"
+ body, _ = _build_multipart_body(
+ [
+ {"name": "file_data", "value": b"hello", "filename": "test.txt"},
+ ],
+ boundary=boundary,
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ # Use quoted boundary
+ gw_event["headers"]["content-type"] = f'multipart/form-data; boundary="{boundary}"'
+ gw_event["body"] = body
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"size": 5}
+
+
+def test_multipart_multiple_values_same_field(gw_event):
+ """Test multiple values for the same field name are collected as list."""
+ from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[List[UploadFile], File()]):
+ return {"count": len(file_data), "filenames": [f.filename for f in file_data]}
+
+ # Build body with two parts having the same field name
+ boundary = "----TestBoundary"
+ raw = (
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="a.txt"\r\n'
+ f"\r\n"
+ f"content a\r\n"
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="b.txt"\r\n'
+ f"\r\n"
+ f"content b\r\n"
+ f"--{boundary}--\r\n"
+ ).encode()
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}"
+ gw_event["body"] = base64.b64encode(raw).decode()
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ parsed = json.loads(result["body"])
+ assert parsed["count"] == 2
+ assert parsed["filenames"] == ["a.txt", "b.txt"]
+
+
+def test_multipart_three_values_same_field(gw_event):
+ """Test three or more values for same field name builds onto existing list."""
+ from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[List[UploadFile], File()]):
+ return {"count": len(file_data), "filenames": [f.filename for f in file_data]}
+
+ boundary = "----TestBoundary"
+ raw = (
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="a.txt"\r\n'
+ f"\r\n"
+ f"aaa\r\n"
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="b.txt"\r\n'
+ f"\r\n"
+ f"bbb\r\n"
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="c.txt"\r\n'
+ f"\r\n"
+ f"ccc\r\n"
+ f"--{boundary}--\r\n"
+ ).encode()
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}"
+ gw_event["body"] = base64.b64encode(raw).decode()
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ parsed = json.loads(result["body"])
+ assert parsed["count"] == 3
+ assert parsed["filenames"] == ["a.txt", "b.txt", "c.txt"]
+
+
+def test_multipart_part_without_headers_separator(gw_event):
+ """Test that a malformed part missing the header/body separator is skipped."""
+ from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[UploadFile, File()]):
+ return {"filename": file_data.filename}
+
+ # Build a body with one malformed part (no \r\n\r\n) and one valid part
+ boundary = "----TestBoundary"
+ raw = (
+ f"--{boundary}\r\n"
+ f"This part has no header separator at all\r\n"
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="good.txt"\r\n'
+ f"\r\n"
+ f"good content\r\n"
+ f"--{boundary}--\r\n"
+ ).encode()
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}"
+ gw_event["body"] = base64.b64encode(raw).decode()
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ parsed = json.loads(result["body"])
+ assert parsed["filename"] == "good.txt"
+
+
+def test_multipart_part_without_field_name(gw_event):
+ """Test that a part missing the name parameter in Content-Disposition is skipped."""
+ from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[UploadFile, File()]):
+ return {"filename": file_data.filename}
+
+ # Build a body with one part that has no name= param and one valid part
+ boundary = "----TestBoundary"
+ raw = (
+ f"--{boundary}\r\n"
+ f"Content-Disposition: form-data\r\n"
+ f"\r\n"
+ f"orphan content\r\n"
+ f"--{boundary}\r\n"
+ f'Content-Disposition: form-data; name="file_data"; filename="valid.txt"\r\n'
+ f"\r\n"
+ f"valid content\r\n"
+ f"--{boundary}--\r\n"
+ ).encode()
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}"
+ gw_event["body"] = base64.b64encode(raw).decode()
+ gw_event["isBase64Encoded"] = True
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ parsed = json.loads(result["body"])
+ assert parsed["filename"] == "valid.txt"
+
+
+def test_upload_file_validate_error():
+ """Test UploadFile._validate raises ValueError for non-UploadFile values."""
+ from aws_lambda_powertools.event_handler.openapi.params import UploadFile
+
+ with pytest.raises(ValueError, match="Expected UploadFile, got str"):
+ UploadFile._validate("not an upload file")
+
+ with pytest.raises(ValueError, match="Expected UploadFile, got int"):
+ UploadFile._validate(42)
+
+
+def test_multipart_unclosed_quote_in_header():
+ """Test that _extract_header_param returns None when quote is unclosed."""
+ from aws_lambda_powertools.event_handler.middlewares.openapi_validation import _extract_header_param
+
+ # name=" is present but closing quote is missing
+ result = _extract_header_param('Content-Disposition: form-data; name="broken', "name")
+ assert result is None
+
+
+def test_multipart_generic_parse_error(gw_event):
+ """Test that non-ValueError exceptions during multipart parsing produce 422."""
+ from unittest.mock import patch
+
+ from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.post("/upload")
+ def upload(file_data: Annotated[UploadFile, File()]):
+ return {"filename": file_data.filename}
+
+ body_b64, content_type = _build_multipart_body(
+ [{"name": "file_data", "value": b"data", "filename": "test.txt"}],
+ )
+
+ gw_event["httpMethod"] = "POST"
+ gw_event["path"] = "/upload"
+ gw_event["headers"]["content-type"] = content_type
+ gw_event["body"] = body_b64
+ gw_event["isBase64Encoded"] = True
+
+ # Patch _parse_multipart_body to raise a non-ValueError (e.g. TypeError)
+ with patch(
+ "aws_lambda_powertools.event_handler.middlewares.openapi_validation._parse_multipart_body",
+ side_effect=TypeError("unexpected type"),
+ ):
+ result = app(gw_event, {})
+ assert result["statusCode"] == 422
+ body = json.loads(result["body"])
+ assert body["detail"][0]["type"] == "multipart_invalid"
+
+
+# ---------- Cookie parameter tests ----------
+
+
+def test_cookie_param_basic(gw_event):
+ """Test basic cookie parameter extraction from REST API v1 (Cookie header)."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(session_id: Annotated[str, Cookie()]):
+ return {"session_id": session_id}
+
+ gw_event["path"] = "/me"
+ gw_event["headers"]["cookie"] = "session_id=abc123; theme=dark"
+ # Clear multiValueHeaders to avoid interference
+ gw_event.pop("multiValueHeaders", None)
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["session_id"] == "abc123"
+
+
+def test_cookie_param_missing_required(gw_event):
+ """Test that a missing required cookie returns 422."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(session_id: Annotated[str, Cookie()]):
+ return {"session_id": session_id}
+
+ gw_event["path"] = "/me"
+ gw_event["headers"]["cookie"] = "theme=dark"
+ gw_event.pop("multiValueHeaders", None)
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 422
+
+
+def test_cookie_param_with_default(gw_event):
+ """Test cookie parameter with a default value when cookie is absent."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(theme: Annotated[str, Cookie()] = "light"):
+ return {"theme": theme}
+
+ gw_event["path"] = "/me"
+ gw_event["headers"].pop("cookie", None)
+ gw_event.pop("multiValueHeaders", None)
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["theme"] == "light"
+
+
+def test_cookie_param_multiple_cookies(gw_event):
+ """Test extracting multiple cookie parameters."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(
+ session_id: Annotated[str, Cookie()],
+ theme: Annotated[str, Cookie()] = "light",
+ ):
+ return {"session_id": session_id, "theme": theme}
+
+ gw_event["path"] = "/me"
+ gw_event["headers"]["cookie"] = "session_id=abc123; theme=dark"
+ gw_event.pop("multiValueHeaders", None)
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["session_id"] == "abc123"
+ assert body["theme"] == "dark"
+
+
+def test_cookie_param_int_validation(gw_event):
+ """Test cookie parameter with int type validation."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(visits: Annotated[int, Cookie()]):
+ return {"visits": visits}
+
+ gw_event["path"] = "/me"
+ gw_event["headers"]["cookie"] = "visits=42"
+ gw_event.pop("multiValueHeaders", None)
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["visits"] == 42
+
+ # Invalid int
+ gw_event["headers"]["cookie"] = "visits=not_a_number"
+ result = app(gw_event, {})
+ assert result["statusCode"] == 422
+
+
+def test_cookie_param_http_api_v2(gw_event_http):
+ """Test cookie parameter with HTTP API v2 (dedicated cookies field)."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayHttpResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(session_id: Annotated[str, Cookie()]):
+ return {"session_id": session_id}
+
+ gw_event_http["rawPath"] = "/me"
+ gw_event_http["requestContext"]["http"]["method"] = "GET"
+ gw_event_http["cookies"] = ["session_id=xyz789", "theme=dark"]
+
+ result = app(gw_event_http, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["session_id"] == "xyz789"
+
+
+def test_cookie_param_lambda_function_url(gw_event_lambda_url):
+ """Test cookie parameter with Lambda Function URL (v2 format)."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = LambdaFunctionUrlResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(session_id: Annotated[str, Cookie()]):
+ return {"session_id": session_id}
+
+ gw_event_lambda_url["rawPath"] = "/me"
+ gw_event_lambda_url["requestContext"]["http"]["method"] = "GET"
+ gw_event_lambda_url["cookies"] = ["session_id=fn_url_abc"]
+
+ result = app(gw_event_lambda_url, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["session_id"] == "fn_url_abc"
+
+
+def test_cookie_param_alb(gw_event_alb):
+ """Test cookie parameter with ALB (Cookie header in multiValueHeaders)."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = ALBResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(session_id: Annotated[str, Cookie()]):
+ return {"session_id": session_id}
+
+ gw_event_alb["path"] = "/me"
+ gw_event_alb["httpMethod"] = "GET"
+ gw_event_alb["multiValueHeaders"]["cookie"] = ["session_id=alb_abc"]
+
+ result = app(gw_event_alb, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["session_id"] == "alb_abc"
+
+
+def test_cookie_param_openapi_schema():
+ """Test that Cookie() generates correct OpenAPI schema with in=cookie."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(
+ session_id: Annotated[str, Cookie(description="Session identifier")],
+ theme: Annotated[str, Cookie(description="UI theme")] = "light",
+ ):
+ return {"session_id": session_id}
+
+ schema = app.get_openapi_schema()
+ schema_dict = schema.model_dump(mode="json", by_alias=True, exclude_none=True)
+
+ path = schema_dict["paths"]["/me"]["get"]
+ params = path["parameters"]
+
+ cookie_params = [p for p in params if p["in"] == "cookie"]
+ assert len(cookie_params) == 2
+
+ session_param = next(p for p in cookie_params if p["name"] == "session_id")
+ assert session_param["required"] is True
+ assert session_param["description"] == "Session identifier"
+
+ theme_param = next(p for p in cookie_params if p["name"] == "theme")
+ assert theme_param.get("required") is not True
+ assert theme_param["description"] == "UI theme"
+
+
+def test_cookie_param_with_query_and_header(gw_event):
+ """Test that Cookie(), Query(), and Header() work together."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(
+ user_id: Annotated[str, Query()],
+ x_request_id: Annotated[str, Header()],
+ session_id: Annotated[str, Cookie()],
+ ):
+ return {
+ "user_id": user_id,
+ "x_request_id": x_request_id,
+ "session_id": session_id,
+ }
+
+ gw_event["path"] = "/me"
+ gw_event["queryStringParameters"] = {"user_id": "u123"}
+ gw_event["multiValueQueryStringParameters"] = {"user_id": ["u123"]}
+ gw_event["headers"]["x-request-id"] = "req-456"
+ gw_event["multiValueHeaders"] = {"x-request-id": ["req-456"], "cookie": ["session_id=sess-789"]}
+ gw_event["headers"]["cookie"] = "session_id=sess-789"
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["user_id"] == "u123"
+ assert body["x_request_id"] == "req-456"
+ assert body["session_id"] == "sess-789"
+
+
+def test_cookie_param_no_cookies_in_request(gw_event):
+ """Test that empty cookies dict is handled gracefully."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = APIGatewayRestResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(theme: Annotated[str, Cookie()] = "light"):
+ return {"theme": theme}
+
+ gw_event["path"] = "/me"
+ gw_event["headers"] = {}
+ gw_event.pop("multiValueHeaders", None)
+
+ result = app(gw_event, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["theme"] == "light"
+
+
+def test_cookie_param_vpc_lattice_v2(gw_event_vpc_lattice):
+ """Test cookie parameter with VPC Lattice v2 (headers are lists)."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = VPCLatticeV2Resolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(session_id: Annotated[str, Cookie()]):
+ return {"session_id": session_id}
+
+ gw_event_vpc_lattice["method"] = "GET"
+ gw_event_vpc_lattice["path"] = "/me"
+ gw_event_vpc_lattice["headers"]["cookie"] = ["session_id=lattice_abc"]
+
+ result = app(gw_event_vpc_lattice, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["session_id"] == "lattice_abc"
+
+
+def test_cookie_param_vpc_lattice_v1(gw_event_vpc_lattice_v1):
+ """Test cookie parameter with VPC Lattice v1 (comma-separated headers)."""
+ from aws_lambda_powertools.event_handler.openapi.params import Cookie
+
+ app = VPCLatticeResolver(enable_validation=True)
+
+ @app.get("/me")
+ def handler(session_id: Annotated[str, Cookie()]):
+ return {"session_id": session_id}
+
+ gw_event_vpc_lattice_v1["method"] = "GET"
+ gw_event_vpc_lattice_v1["raw_path"] = "/me"
+ gw_event_vpc_lattice_v1["headers"]["cookie"] = "session_id=lattice_v1_abc"
+
+ result = app(gw_event_vpc_lattice_v1, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["session_id"] == "lattice_v1_abc"
+
+
+def test_alb_response_none_body_with_validation(gw_event_alb):
+ # GIVEN an ALBResolver with validation enabled
+ app = ALBResolver(enable_validation=True)
+
+ gw_event_alb["path"] = "/no-content"
+ gw_event_alb["httpMethod"] = "DELETE"
+
+ # WHEN a handler returns Response with body=None and return type is None
+ @app.delete("/no-content")
+ def handler() -> None:
+ return Response(status_code=204, body=None)
+
+ # THEN the response should be 204 with empty body (not 422 validation error)
+ result = app(gw_event_alb, {})
+ assert result["statusCode"] == 204
+ assert result["body"] == ""
+
+
+def test_alb_response_typed_none_body_with_validation(gw_event_alb):
+ # GIVEN an ALBResolver with validation enabled
+ app = ALBResolver(enable_validation=True)
+
+ gw_event_alb["path"] = "/no-content"
+ gw_event_alb["httpMethod"] = "DELETE"
+
+ # WHEN a handler returns Response[None] with body=None
+ @app.delete("/no-content")
+ def handler() -> Response[None]:
+ return Response(status_code=204, body=None)
+
+ # THEN the response should be 204 with empty body (not 422 validation error)
+ result = app(gw_event_alb, {})
+ assert result["statusCode"] == 204
+ assert result["body"] == ""
diff --git a/tests/functional/event_handler/_pydantic/test_resolve_async_validation.py b/tests/functional/event_handler/_pydantic/test_resolve_async_validation.py
new file mode 100644
index 00000000000..92b414f72b5
--- /dev/null
+++ b/tests/functional/event_handler/_pydantic/test_resolve_async_validation.py
@@ -0,0 +1,55 @@
+import asyncio
+
+from aws_lambda_powertools.event_handler.api_gateway import (
+ APIGatewayHttpResolver,
+ BaseRouter,
+)
+from tests.functional.utils import load_event
+
+API_RESTV2_EVENT = load_event("apiGatewayProxyV2Event_GET.json")
+
+
+def _setup_app(app, event):
+ BaseRouter.current_event = app._to_proxy_event(event)
+ BaseRouter.lambda_context = {}
+
+
+class TestResolveAsyncValidation:
+ def test_validation_middleware_created_and_used(self):
+ # GIVEN a resolver with validation enabled and an async handler
+ app = APIGatewayHttpResolver(enable_validation=True)
+
+ @app.get("/my/path")
+ async def get_lambda() -> dict:
+ await asyncio.sleep(0)
+ return {"message": "validated"}
+
+ # WHEN calling _resolve_async
+ _setup_app(app, API_RESTV2_EVENT)
+ result = asyncio.run(app._resolve_async())
+
+ # THEN the validation middlewares are created and the response is valid
+ response = result.build(app.current_event, app._cors)
+ assert response["statusCode"] == 200
+ assert hasattr(app, "_request_validation_middleware")
+ assert hasattr(app, "_response_validation_middleware")
+
+ def test_validation_middleware_lazy_created_for_per_route_validation(self):
+ # GIVEN a resolver WITHOUT global validation, but a route WITH enable_validation=True
+ app = APIGatewayHttpResolver()
+ assert not hasattr(app, "_request_validation_middleware")
+
+ @app.get("/my/path", enable_validation=True)
+ async def get_lambda() -> dict:
+ await asyncio.sleep(0)
+ return {"message": "lazy validated"}
+
+ # WHEN calling _resolve_async (triggers lazy creation in Route.call_async)
+ _setup_app(app, API_RESTV2_EVENT)
+ result = asyncio.run(app._resolve_async())
+
+ # THEN validation middlewares are lazily created on the app
+ response = result.build(app.current_event, app._cors)
+ assert response["statusCode"] == 200
+ assert hasattr(app, "_request_validation_middleware")
+ assert hasattr(app, "_response_validation_middleware")
diff --git a/tests/functional/event_handler/required_dependencies/test_async_middleware_frame.py b/tests/functional/event_handler/required_dependencies/test_async_middleware_frame.py
new file mode 100644
index 00000000000..6154820454d
--- /dev/null
+++ b/tests/functional/event_handler/required_dependencies/test_async_middleware_frame.py
@@ -0,0 +1,89 @@
+import asyncio
+
+import pytest
+
+from aws_lambda_powertools.event_handler import content_types
+from aws_lambda_powertools.event_handler.api_gateway import (
+ ApiGatewayResolver,
+ ProxyEventType,
+ Response,
+)
+from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
+from aws_lambda_powertools.event_handler.middlewares.async_utils import AsyncMiddlewareFrame, wrap_middleware_async
+from tests.functional.utils import load_event
+
+API_REST_EVENT = load_event("apiGatewayProxyEvent.json")
+
+
+def _make_app() -> ApiGatewayResolver:
+ app = ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent)
+ app.current_event = app._to_proxy_event(API_REST_EVENT)
+ app.lambda_context = {}
+ return app
+
+
+def test_sync_middleware_raising_before_next_does_not_deadlock():
+ # GIVEN a sync middleware that raises before calling next()
+ # This previously caused a deadlock because middleware_called_next was never set
+ app = _make_app()
+
+ class AuthError(Exception):
+ pass
+
+ def failing_middleware(app: ApiGatewayResolver, next_middleware: NextMiddleware):
+ raise AuthError("denied")
+
+ async def next_handler(app: ApiGatewayResolver):
+ await asyncio.sleep(0)
+ return Response(200, content_types.TEXT_HTML, "should not reach")
+
+ frame = AsyncMiddlewareFrame(current_middleware=failing_middleware, next_middleware=next_handler)
+
+ # WHEN calling the frame
+ # THEN the exception propagates without deadlocking
+ with pytest.raises(AuthError, match="denied"):
+ asyncio.run(frame(app))
+
+
+def test_wrap_middleware_async_sync_raising_before_next_does_not_deadlock():
+ # GIVEN a sync middleware that raises before calling next(), using wrap_middleware_async
+ # This exercises _run_sync_middleware_in_thread directly
+ app = _make_app()
+
+ class AuthError(Exception):
+ pass
+
+ def failing_middleware(app, next_middleware):
+ raise AuthError("denied")
+
+ async def next_handler(app):
+ return Response(200, content_types.TEXT_HTML, "should not reach")
+
+ wrapped = wrap_middleware_async(failing_middleware, next_handler)
+
+ # WHEN calling the wrapped middleware
+ # THEN the exception propagates without deadlocking
+ with pytest.raises(AuthError, match="denied"):
+ asyncio.run(wrapped(app))
+
+
+def test_async_middleware_raising_before_next_propagates():
+ # GIVEN an async middleware that raises before calling next()
+ app = _make_app()
+
+ class ValidationError(Exception):
+ pass
+
+ async def failing_middleware(app: ApiGatewayResolver, next_middleware: NextMiddleware):
+ raise ValidationError("invalid request")
+
+ async def next_handler(app: ApiGatewayResolver):
+ await asyncio.sleep(0)
+ return Response(200, content_types.TEXT_HTML, "should not reach")
+
+ frame = AsyncMiddlewareFrame(current_middleware=failing_middleware, next_middleware=next_handler)
+
+ # WHEN calling the frame
+ # THEN the exception propagates
+ with pytest.raises(ValidationError, match="invalid request"):
+ asyncio.run(frame(app))
diff --git a/tests/functional/event_handler/required_dependencies/test_depends.py b/tests/functional/event_handler/required_dependencies/test_depends.py
new file mode 100644
index 00000000000..d5e49e07cdd
--- /dev/null
+++ b/tests/functional/event_handler/required_dependencies/test_depends.py
@@ -0,0 +1,509 @@
+"""Tests for the Depends() dependency injection feature using Annotated."""
+
+import json
+
+import pytest
+from typing_extensions import Annotated
+
+from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
+from aws_lambda_powertools.event_handler.depends import DependencyResolutionError, Depends
+from aws_lambda_powertools.event_handler.request import Request
+from tests.functional.utils import load_event
+
+API_GW_V2_EVENT = load_event("apiGatewayProxyV2Event.json")
+
+
+def test_depends_simple():
+ """A simple dependency is resolved and injected into the handler."""
+ app = APIGatewayHttpResolver()
+
+ def get_greeting() -> str:
+ return "hello"
+
+ @app.post("/my/path")
+ def handler(greeting: Annotated[str, Depends(get_greeting)]):
+ return {"greeting": greeting}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"greeting": "hello"}
+
+
+def test_depends_nested():
+ """Dependencies can depend on other dependencies."""
+ app = APIGatewayHttpResolver()
+
+ def get_prefix() -> str:
+ return "Hello"
+
+ def get_greeting(prefix: Annotated[str, Depends(get_prefix)]) -> str:
+ return f"{prefix}, world!"
+
+ @app.post("/my/path")
+ def handler(greeting: Annotated[str, Depends(get_greeting)]):
+ return {"greeting": greeting}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"greeting": "Hello, world!"}
+
+
+def test_depends_cache_per_invocation():
+ """Same dependency used twice in one invocation is only resolved once (use_cache=True)."""
+ app = APIGatewayHttpResolver()
+ call_count = 0
+
+ def get_config() -> dict:
+ nonlocal call_count
+ call_count += 1
+ return {"key": "value"}
+
+ def get_a(config: Annotated[dict, Depends(get_config)]) -> str:
+ return config["key"]
+
+ def get_b(config: Annotated[dict, Depends(get_config)]) -> str:
+ return config["key"]
+
+ @app.post("/my/path")
+ def handler(a: Annotated[str, Depends(get_a)], b: Annotated[str, Depends(get_b)]):
+ return {"a": a, "b": b}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert call_count == 1 # get_config called once despite being used by both get_a and get_b
+
+
+def test_depends_no_cache():
+ """use_cache=False resolves every time."""
+ app = APIGatewayHttpResolver()
+ call_count = 0
+
+ def get_value() -> int:
+ nonlocal call_count
+ call_count += 1
+ return call_count
+
+ @app.post("/my/path")
+ def handler(
+ a: Annotated[int, Depends(get_value, use_cache=False)],
+ b: Annotated[int, Depends(get_value, use_cache=False)],
+ ):
+ return {"a": a, "b": b}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert call_count == 2
+
+
+def test_depends_with_request():
+ """A dependency can receive the Request object."""
+ app = APIGatewayHttpResolver()
+
+ def get_method(request: Request) -> str:
+ return request.method
+
+ @app.post("/my/path")
+ def handler(method: Annotated[str, Depends(get_method)]):
+ return {"method": method}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"method": "POST"}
+
+
+def test_depends_override():
+ """dependency_overrides replaces a dependency callable for testing."""
+ app = APIGatewayHttpResolver()
+
+ def get_tenant() -> str:
+ return "real-tenant"
+
+ @app.post("/my/path")
+ def handler(tenant: Annotated[str, Depends(get_tenant)]):
+ return {"tenant": tenant}
+
+ app.dependency_overrides[get_tenant] = lambda: "test-tenant"
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"tenant": "test-tenant"}
+
+ app.dependency_overrides.clear()
+
+
+def test_depends_override_nested():
+ """dependency_overrides works for nested dependencies too."""
+ app = APIGatewayHttpResolver()
+
+ def get_db_client():
+ return "real-db"
+
+ def get_table(db: Annotated[str, Depends(get_db_client)]) -> str:
+ return f"table-from-{db}"
+
+ @app.post("/my/path")
+ def handler(table: Annotated[str, Depends(get_table)]):
+ return {"table": table}
+
+ app.dependency_overrides[get_db_client] = lambda: "mock-db"
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"table": "table-from-mock-db"}
+
+ app.dependency_overrides.clear()
+
+
+def test_depends_multiple_handlers():
+ """Dependencies work across different route handlers."""
+ app = APIGatewayHttpResolver()
+
+ def get_user() -> str:
+ return "user-123"
+
+ @app.get("/my/path")
+ def get_handler(user: Annotated[str, Depends(get_user)]):
+ return {"user": user, "action": "get"}
+
+ @app.post("/my/path")
+ def post_handler(user: Annotated[str, Depends(get_user)]):
+ return {"user": user, "action": "post"}
+
+ # Test POST (matches the event)
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"user": "user-123", "action": "post"}
+
+
+def test_depends_reusable_type_alias():
+ """Annotated type aliases can be reused across handlers."""
+ app = APIGatewayHttpResolver()
+
+ def get_tenant() -> str:
+ return "tenant-abc"
+
+ TenantId = Annotated[str, Depends(get_tenant)]
+
+ @app.post("/my/path")
+ def handler(tenant: TenantId):
+ return {"tenant": tenant}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"tenant": "tenant-abc"}
+
+
+def test_handler_without_depends_works_normally():
+ """A plain handler with no Depends() params is not affected by DI."""
+ app = APIGatewayHttpResolver()
+
+ @app.post("/my/path")
+ def handler():
+ return {"ok": True}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"ok": True}
+
+
+def test_depends_not_cached_across_invocations():
+ """Each app() call resolves dependencies fresh — no cross-request leakage."""
+ app = APIGatewayHttpResolver()
+ call_count = 0
+
+ def get_counter() -> int:
+ nonlocal call_count
+ call_count += 1
+ return call_count
+
+ @app.post("/my/path")
+ def handler(c: Annotated[int, Depends(get_counter)]):
+ return {"c": c}
+
+ result1 = app(API_GW_V2_EVENT, {})
+ result2 = app(API_GW_V2_EVENT, {})
+
+ assert json.loads(result1["body"]) == {"c": 1}
+ assert json.loads(result2["body"]) == {"c": 2}
+ assert call_count == 2
+
+
+def test_depends_deeply_nested():
+ """Three-level dependency chain resolves correctly."""
+ app = APIGatewayHttpResolver()
+
+ def get_url() -> str:
+ return "postgres://localhost"
+
+ def get_conn(url: Annotated[str, Depends(get_url)]) -> str:
+ return f"conn({url})"
+
+ def get_session(conn: Annotated[str, Depends(get_conn)]) -> str:
+ return f"session({conn})"
+
+ @app.post("/my/path")
+ def handler(session: Annotated[str, Depends(get_session)]):
+ return {"session": session}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"session": "session(conn(postgres://localhost))"}
+
+
+def test_depends_with_request_reads_headers():
+ """A dependency using Request can read actual request headers."""
+ app = APIGatewayHttpResolver()
+
+ def get_user_agent(request: Request) -> str:
+ return request.headers.get("user-agent", "unknown")
+
+ @app.post("/my/path")
+ def handler(ua: Annotated[str, Depends(get_user_agent)]):
+ return {"ua": ua}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert isinstance(json.loads(result["body"])["ua"], str)
+
+
+def test_depends_returning_none():
+ """A dependency can return None without breaking."""
+ app = APIGatewayHttpResolver()
+
+ def get_nothing() -> None:
+ return None
+
+ @app.post("/my/path")
+ def handler(val: Annotated[None, Depends(get_nothing)]):
+ return {"val": val}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"val": None}
+
+
+def test_depends_exception_raises_dependency_resolution_error():
+ """If a dependency raises, a DependencyResolutionError wraps the original exception."""
+ app = APIGatewayHttpResolver()
+
+ def broken() -> str:
+ raise ValueError("boom")
+
+ @app.post("/my/path")
+ def handler(val: Annotated[str, Depends(broken)]):
+ return {"val": val}
+
+ with pytest.raises(DependencyResolutionError, match="broken.*boom"):
+ app(API_GW_V2_EVENT, {})
+
+
+def test_depends_non_callable_raises_dependency_resolution_error():
+ """Passing a non-callable to Depends() raises DependencyResolutionError immediately."""
+ with pytest.raises(DependencyResolutionError, match="requires a callable"):
+ Depends("not_a_function") # type: ignore
+
+ with pytest.raises(DependencyResolutionError, match="requires a callable"):
+ Depends(42) # type: ignore
+
+ with pytest.raises(DependencyResolutionError, match="requires a callable"):
+ Depends(None) # type: ignore
+
+
+def test_depends_accepts_lambda():
+ """Depends() works with a lambda as the dependency."""
+ app = APIGatewayHttpResolver()
+
+ @app.post("/my/path")
+ def handler(val: Annotated[str, Depends(lambda: "from-lambda")]):
+ return {"val": val}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"val": "from-lambda"}
+
+
+def test_depends_accepts_class_with_call():
+ """Depends() works with a class that implements __call__."""
+ app = APIGatewayHttpResolver()
+
+ class TenantProvider:
+ def __call__(self) -> str:
+ return "tenant-from-class"
+
+ @app.post("/my/path")
+ def handler(tenant: Annotated[str, Depends(TenantProvider())]):
+ return {"tenant": tenant}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"tenant": "tenant-from-class"}
+
+
+def test_depends_accepts_class_as_factory():
+ """Depends() works with a class itself (constructor as callable)."""
+ app = APIGatewayHttpResolver()
+
+ class Config:
+ def __init__(self):
+ self.region = "us-east-1"
+
+ @app.post("/my/path")
+ def handler(config: Annotated[Config, Depends(Config)]):
+ return {"region": config.region}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"region": "us-east-1"}
+
+
+def test_depends_with_unresolvable_annotations_is_ignored():
+ """A handler whose annotations cannot be resolved by get_type_hints is treated as having no deps."""
+ app = APIGatewayHttpResolver()
+
+ # Build a function with broken annotations that get_type_hints cannot resolve.
+ # The param has a default so the handler can still be called without it.
+ def make_handler():
+ def handler(x: "CompletelyBogusType" = None): # noqa: F821
+ return {"ok": True}
+
+ return handler
+
+ app.post("/my/path")(make_handler())
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"ok": True}
+
+
+def test_depends_without_request_does_not_inject():
+ """A dependency that does NOT declare Request still works when request is available."""
+ app = APIGatewayHttpResolver()
+
+ def get_static() -> str:
+ return "no-request-needed"
+
+ @app.post("/my/path")
+ def handler(val: Annotated[str, Depends(get_static)]):
+ return {"val": val}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"val": "no-request-needed"}
+
+
+def test_depends_with_broken_type_hints_on_dependency():
+ """A dependency callable with broken annotations still resolves (get_type_hints fails gracefully)."""
+ app = APIGatewayHttpResolver()
+
+ # Create a callable whose annotations reference a nonexistent type
+ # so get_type_hints() will raise inside solve_dependencies
+ broken_dep = type(
+ "BrokenDep",
+ (),
+ {
+ "__call__": lambda self: "it-works",
+ "__annotations__": {"x": "NonExistentType"},
+ "__module__": __name__,
+ },
+ )()
+
+ @app.post("/my/path")
+ def handler(val: Annotated[str, Depends(broken_dep)]):
+ return {"val": val}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"val": "it-works"}
+
+
+# ---------------------------------------------------------------------------
+# request.context — bridge between middleware and Depends()
+# ---------------------------------------------------------------------------
+
+
+def test_depends_request_context_writable():
+ """Dependencies can write to request.context and handlers can read it."""
+ app = APIGatewayHttpResolver()
+
+ def set_tenant(request: Request) -> str:
+ tenant = request.headers.get("x-tenant-id", "default")
+ request.context["tenant"] = tenant
+ return tenant
+
+ @app.post("/my/path")
+ def handler(tenant: Annotated[str, Depends(set_tenant)], request: Request):
+ return {"tenant": tenant, "from_context": request.context.get("tenant")}
+
+ event = {**API_GW_V2_EVENT, "headers": {**API_GW_V2_EVENT.get("headers", {}), "x-tenant-id": "acme-corp"}}
+ result = app(event, {})
+
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["tenant"] == "acme-corp"
+ assert body["from_context"] == "acme-corp"
+
+
+def test_depends_request_context_bridges_middleware():
+ """Middleware writes to app.context, Depends() reads via request.context."""
+ app = APIGatewayHttpResolver()
+
+ def auth_middleware(app, next_middleware):
+ app.append_context(user="admin-user")
+ return next_middleware(app)
+
+ app.use(middlewares=[auth_middleware])
+
+ def get_current_user(request: Request) -> str:
+ return request.context["user"]
+
+ @app.post("/my/path")
+ def handler(user: Annotated[str, Depends(get_current_user)]):
+ return {"user": user}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"user": "admin-user"}
+
+
+def test_depends_request_context_with_router():
+ """request.context works when routes come from an included Router."""
+ from aws_lambda_powertools.event_handler.api_gateway import Router
+
+ app = APIGatewayHttpResolver()
+ router = Router()
+
+ def mw(app, next_middleware):
+ app.append_context(role="admin")
+ return next_middleware(app)
+
+ app.use(middlewares=[mw])
+
+ def get_role(request: Request) -> str:
+ return request.context["role"]
+
+ @router.post("/my/path")
+ def handler(role: Annotated[str, Depends(get_role)]):
+ return {"role": role}
+
+ app.include_router(router)
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ assert json.loads(result["body"]) == {"role": "admin"}
+
+
+def test_depends_request_resolved_event():
+ """Dependencies can access the full event via request.resolved_event."""
+ app = APIGatewayHttpResolver()
+
+ def get_path(request: Request) -> str:
+ return request.resolved_event.path
+
+ @app.post("/my/path")
+ def handler(path: Annotated[str, Depends(get_path)]):
+ return {"path": path}
+
+ result = app(API_GW_V2_EVENT, {})
+ assert result["statusCode"] == 200
+ body = json.loads(result["body"])
+ assert body["path"] == "/my/path"
diff --git a/tests/functional/event_handler/required_dependencies/test_http_resolver.py b/tests/functional/event_handler/required_dependencies/test_http_resolver.py
index 40fb3d20c64..4665812e64a 100644
--- a/tests/functional/event_handler/required_dependencies/test_http_resolver.py
+++ b/tests/functional/event_handler/required_dependencies/test_http_resolver.py
@@ -1,4 +1,4 @@
-"""Tests for HttpResolverLocal - ASGI-compatible HTTP resolver for local development."""
+"""Tests for HttpResolverLocal - ASGI-compatible HTTP resolver."""
from __future__ import annotations
@@ -11,10 +11,6 @@
from aws_lambda_powertools.event_handler import HttpResolverLocal, Response
from aws_lambda_powertools.event_handler.http_resolver import MockLambdaContext
-# Suppress warning for all tests
-pytestmark = pytest.mark.filterwarnings("ignore:HttpResolverLocal is intended for local development")
-
-
# =============================================================================
# ASGI Test Helpers
# =============================================================================
@@ -1242,3 +1238,219 @@ def hello():
# THEN it returns 404 (method mismatch is treated as not found)
assert captured["status_code"] == 404
+
+
+# =============================================================================
+# CORS Tests (issue #8267)
+# =============================================================================
+
+
+@pytest.mark.asyncio
+async def test_cors_options_preflight_returns_204():
+ # GIVEN an app with CORSConfig and a POST route
+ from aws_lambda_powertools.event_handler.api_gateway import CORSConfig
+
+ app = HttpResolverLocal(cors=CORSConfig(allow_origin="*"))
+
+ @app.post("/items")
+ def create_item():
+ return {"ok": True}
+
+ # WHEN a browser sends a CORS preflight OPTIONS request
+ scope = {
+ "type": "http",
+ "method": "OPTIONS",
+ "path": "/items",
+ "query_string": b"",
+ "headers": [
+ (b"origin", b"http://localhost:3000"),
+ (b"access-control-request-method", b"POST"),
+ ],
+ }
+
+ receive = make_asgi_receive()
+ captured: dict[str, Any] = {"status_code": None, "headers": []}
+
+ async def send(message: dict[str, Any]) -> None:
+ await asyncio.sleep(0)
+ if message["type"] == "http.response.start":
+ captured["status_code"] = message["status"]
+ captured["headers"].extend(message.get("headers", []))
+
+ await app(scope, receive, send)
+
+ # THEN it returns 204 with CORS headers (not 500 or 404)
+ assert captured["status_code"] == 204
+
+ header_names = [name.lower() for name, _ in captured["headers"]]
+ assert b"access-control-allow-origin" in header_names
+ assert b"access-control-allow-methods" in header_names
+
+
+@pytest.mark.asyncio
+async def test_cors_options_preflight_with_exception_handler_does_not_return_500():
+ # GIVEN an app with CORSConfig and a generic exception handler that returns 500
+ import json
+
+ from aws_lambda_powertools.event_handler.api_gateway import CORSConfig
+
+ app = HttpResolverLocal(cors=CORSConfig(allow_origin="*"))
+
+ @app.post("/items")
+ def create_item():
+ return {"ok": True}
+
+ @app.exception_handler(Exception)
+ def handle_server_error(ex: Exception):
+ return Response(
+ status_code=500,
+ content_type="application/json",
+ body=json.dumps({"error": "internal"}),
+ )
+
+ # WHEN a browser sends a CORS preflight OPTIONS request
+ scope = {
+ "type": "http",
+ "method": "OPTIONS",
+ "path": "/items",
+ "query_string": b"",
+ "headers": [
+ (b"origin", b"http://localhost:3000"),
+ (b"access-control-request-method", b"POST"),
+ ],
+ }
+
+ receive = make_asgi_receive()
+ captured: dict[str, Any] = {"status_code": None, "headers": []}
+
+ async def send(message: dict[str, Any]) -> None:
+ await asyncio.sleep(0)
+ if message["type"] == "http.response.start":
+ captured["status_code"] = message["status"]
+ captured["headers"].extend(message.get("headers", []))
+
+ await app(scope, receive, send)
+
+ # THEN the OPTIONS request returns 204, not 500
+ assert captured["status_code"] == 204
+ header_names = [name.lower() for name, _ in captured["headers"]]
+ assert b"access-control-allow-origin" in header_names
+
+
+@pytest.mark.asyncio
+async def test_no_cors_options_returns_404():
+ # GIVEN an app WITHOUT CORSConfig
+ app = HttpResolverLocal()
+
+ @app.post("/items")
+ def create_item():
+ return {"ok": True}
+
+ # WHEN a browser sends an OPTIONS request (no CORS configured)
+ scope = {
+ "type": "http",
+ "method": "OPTIONS",
+ "path": "/items",
+ "query_string": b"",
+ "headers": [],
+ }
+
+ receive = make_asgi_receive()
+ send, captured = make_asgi_send()
+
+ await app(scope, receive, send)
+
+ # THEN it returns 404 (no CORS config, no special handling)
+ assert captured["status_code"] == 404
+
+
+@pytest.mark.asyncio
+async def test_cors_options_includes_allowed_methods_header():
+ # GIVEN an app with CORSConfig and multiple routes
+ from aws_lambda_powertools.event_handler.api_gateway import CORSConfig
+
+ app = HttpResolverLocal(cors=CORSConfig(allow_origin="https://example.com"))
+
+ @app.get("/resource")
+ def get_resource():
+ return {"method": "GET"}
+
+ @app.post("/resource")
+ def post_resource():
+ return {"method": "POST"}
+
+ # WHEN an OPTIONS preflight is sent
+ scope = {
+ "type": "http",
+ "method": "OPTIONS",
+ "path": "/resource",
+ "query_string": b"",
+ "headers": [
+ (b"origin", b"https://example.com"),
+ (b"access-control-request-method", b"GET"),
+ ],
+ }
+
+ receive = make_asgi_receive()
+ captured: dict[str, Any] = {"status_code": None, "headers": []}
+
+ async def send(message: dict[str, Any]) -> None:
+ await asyncio.sleep(0)
+ if message["type"] == "http.response.start":
+ captured["status_code"] = message["status"]
+ captured["headers"].extend(message.get("headers", []))
+
+ await app(scope, receive, send)
+
+ # THEN 204 is returned with Access-Control-Allow-Methods header
+ assert captured["status_code"] == 204
+ allow_methods_headers = [v for name, v in captured["headers"] if name.lower() == b"access-control-allow-methods"]
+ assert len(allow_methods_headers) == 1
+
+
+@pytest.mark.asyncio
+async def test_cors_disallowed_header_not_in_allow_headers():
+ # GIVEN an app with CORSConfig that only allows specific headers
+ from aws_lambda_powertools.event_handler.api_gateway import CORSConfig
+
+ app = HttpResolverLocal(cors=CORSConfig(allow_origin="*", allow_headers=["X-Custom-Allowed"]))
+
+ @app.post("/items")
+ def create_item():
+ return {"ok": True}
+
+ # WHEN a preflight requests an unlisted header
+ scope = {
+ "type": "http",
+ "method": "OPTIONS",
+ "path": "/items",
+ "query_string": b"",
+ "headers": [
+ (b"origin", b"http://localhost:3000"),
+ (b"access-control-request-method", b"POST"),
+ (b"access-control-request-headers", b"X-Not-Allowed"),
+ ],
+ }
+
+ receive = make_asgi_receive()
+ captured: dict[str, Any] = {"status_code": None, "headers": []}
+
+ async def send(message: dict[str, Any]) -> None:
+ await asyncio.sleep(0)
+ if message["type"] == "http.response.start":
+ captured["status_code"] = message["status"]
+ captured["headers"].extend(message.get("headers", []))
+
+ await app(scope, receive, send)
+
+ # THEN the server still returns 204 (browser enforces the rejection, not the server)
+ assert captured["status_code"] == 204
+
+ # AND the unlisted header is absent from Access-Control-Allow-Headers
+ allow_headers_value = next(
+ (v.decode() for name, v in captured["headers"] if name.lower() == b"access-control-allow-headers"),
+ "",
+ )
+ assert "X-Not-Allowed" not in allow_headers_value
+ # AND the explicitly allowed header IS present
+ assert "X-Custom-Allowed" in allow_headers_value
diff --git a/tests/functional/event_handler/required_dependencies/test_registered_api_adapter_async.py b/tests/functional/event_handler/required_dependencies/test_registered_api_adapter_async.py
new file mode 100644
index 00000000000..10d5b4602f0
--- /dev/null
+++ b/tests/functional/event_handler/required_dependencies/test_registered_api_adapter_async.py
@@ -0,0 +1,335 @@
+import asyncio
+import re
+from typing import cast
+
+import pytest
+from typing_extensions import Annotated
+
+from aws_lambda_powertools.event_handler import content_types
+from aws_lambda_powertools.event_handler.api_gateway import (
+ APIGatewayHttpResolver,
+ ApiGatewayResolver,
+ APIGatewayRestResolver,
+ BaseRouter,
+ ProxyEventType,
+ Response,
+ Route,
+)
+from aws_lambda_powertools.event_handler.depends import Depends
+from aws_lambda_powertools.event_handler.middlewares.async_utils import _registered_api_adapter_async
+from aws_lambda_powertools.event_handler.request import Request
+from tests.functional.utils import load_event
+
+API_REST_EVENT = load_event("apiGatewayProxyEvent.json")
+API_RESTV2_EVENT = load_event("apiGatewayProxyV2Event_GET.json")
+
+
+def _setup_resolver_context(app: ApiGatewayResolver, event: dict) -> None:
+ """Populate the resolver context the same way resolve() does, without calling the full chain."""
+ BaseRouter.current_event = app._to_proxy_event(cast(dict, event))
+ BaseRouter.lambda_context = {}
+
+
+@pytest.mark.parametrize(
+ "app, event",
+ [
+ (ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent), API_REST_EVENT),
+ (APIGatewayRestResolver(), API_REST_EVENT),
+ (APIGatewayHttpResolver(), API_RESTV2_EVENT),
+ ],
+)
+def test_sync_handler_returns_response(app: ApiGatewayResolver, event):
+ # GIVEN a sync route handler
+ @app.get("/my/path")
+ def get_lambda():
+ return Response(200, content_types.TEXT_HTML, "sync response")
+
+ # WHEN resolving the event through the normal chain
+ result = app(event, {})
+
+ # THEN the sync handler is called and returns correctly
+ assert result["statusCode"] == 200
+ assert result["body"] == "sync response"
+
+
+@pytest.mark.parametrize(
+ "app, event",
+ [
+ (ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent), API_REST_EVENT),
+ (APIGatewayRestResolver(), API_REST_EVENT),
+ (APIGatewayHttpResolver(), API_RESTV2_EVENT),
+ ],
+)
+def test_async_handler_is_awaited(app: ApiGatewayResolver, event):
+ # GIVEN an async route handler registered on the resolver
+ @app.get("/my/path")
+ async def get_lambda():
+ return Response(200, content_types.TEXT_HTML, "async response")
+
+ # WHEN populating context and calling the async adapter directly
+ _setup_resolver_context(app, event)
+ app.append_context(_route_args={})
+
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN the async handler is awaited and returns correctly
+ assert result.status_code == 200
+ assert result.body == "async response"
+
+
+@pytest.mark.parametrize(
+ "app, event",
+ [
+ (ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent), API_REST_EVENT),
+ (APIGatewayRestResolver(), API_REST_EVENT),
+ (APIGatewayHttpResolver(), API_RESTV2_EVENT),
+ ],
+)
+def test_sync_handler_through_adapter(app: ApiGatewayResolver, event):
+ # GIVEN a sync route handler
+ @app.get("/my/path")
+ def get_lambda():
+ return Response(200, content_types.TEXT_HTML, "sync via adapter")
+
+ # WHEN calling _registered_api_adapter_async with a sync handler
+ _setup_resolver_context(app, event)
+ app.append_context(_route_args={})
+
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN sync handler works through the async adapter without issue
+ assert result.status_code == 200
+ assert result.body == "sync via adapter"
+
+
+@pytest.mark.parametrize(
+ "app, event",
+ [
+ (ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent), API_REST_EVENT),
+ (APIGatewayRestResolver(), API_REST_EVENT),
+ (APIGatewayHttpResolver(), API_RESTV2_EVENT),
+ ],
+)
+def test_adapter_passes_route_args_to_async_handler(app: ApiGatewayResolver, event):
+ # GIVEN an async handler that expects route arguments
+ async def get_lambda(name: str):
+ return Response(200, content_types.TEXT_HTML, name)
+
+ # WHEN route_args are set in the context
+ _setup_resolver_context(app, event)
+ app.append_context(_route_args={"name": "powertools"})
+
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN the route args are passed to the handler
+ assert result.status_code == 200
+ assert result.body == "powertools"
+
+
+@pytest.mark.parametrize(
+ "app, event",
+ [
+ (ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent), API_REST_EVENT),
+ (APIGatewayRestResolver(), API_REST_EVENT),
+ (APIGatewayHttpResolver(), API_RESTV2_EVENT),
+ ],
+)
+def test_adapter_passes_route_args_to_sync_handler(app: ApiGatewayResolver, event):
+ # GIVEN a sync handler that expects route arguments
+ def get_lambda(name: str):
+ return Response(200, content_types.TEXT_HTML, name)
+
+ # WHEN route_args are set in the context
+ _setup_resolver_context(app, event)
+ app.append_context(_route_args={"name": "powertools"})
+
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN the route args are passed to the sync handler
+ assert result.status_code == 200
+ assert result.body == "powertools"
+
+
+def test_adapter_converts_dict_response_from_async_handler():
+ # GIVEN an async handler that returns a dict (not a Response object)
+ app = ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent)
+
+ async def get_lambda():
+ return {"message": "hello"}
+
+ # WHEN calling through the async adapter
+ _setup_resolver_context(app, API_REST_EVENT)
+ app.append_context(_route_args={})
+
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN _to_response normalizes the dict into a Response object
+ assert result.status_code == 200
+ assert result.body is not None
+
+
+def test_adapter_converts_tuple_response_from_async_handler():
+ # GIVEN an async handler that returns a (dict, status_code) tuple
+ app = ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent)
+
+ async def get_lambda():
+ return {"created": True}, 201
+
+ # WHEN calling through the async adapter
+ _setup_resolver_context(app, API_REST_EVENT)
+ app.append_context(_route_args={})
+
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN _to_response normalizes the tuple into a Response object
+ assert result.status_code == 201
+
+
+def test_adapter_with_no_route_in_context():
+ # GIVEN a handler and no _route in context
+ app = ApiGatewayResolver(proxy_type=ProxyEventType.APIGatewayProxyEvent)
+
+ async def get_lambda():
+ return Response(200, content_types.TEXT_HTML, "no route")
+
+ # WHEN _route is None in context (default)
+ _setup_resolver_context(app, API_REST_EVENT)
+ app.append_context(_route_args={})
+
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN the adapter skips request injection and dependency resolution
+ assert result.status_code == 200
+ assert result.body == "no route"
+
+
+def test_adapter_injects_request_param():
+ # GIVEN an async handler that declares a Request parameter
+ app = APIGatewayHttpResolver()
+
+ async def get_lambda(request: Request):
+ return Response(200, content_types.TEXT_HTML, request.method)
+
+ # WHEN a Route is present in context with request_param_name not yet checked
+ _setup_resolver_context(app, API_RESTV2_EVENT)
+ route = Route(
+ method="GET",
+ path="/my/path",
+ rule=re.compile(r"^/my/path$"),
+ func=get_lambda,
+ cors=False,
+ compress=False,
+ )
+ app.append_context(_route=route, _route_args={})
+
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN the Request object is injected and request_param_name is cached
+ assert result.status_code == 200
+ assert route.request_param_name_checked is True
+ assert route.request_param_name == "request"
+
+
+def test_adapter_uses_cached_request_param_name():
+ # GIVEN a Route where request_param_name was already resolved
+ app = APIGatewayHttpResolver()
+
+ async def get_lambda(req: Request):
+ return Response(200, content_types.TEXT_HTML, req.method)
+
+ _setup_resolver_context(app, API_RESTV2_EVENT)
+ route = Route(
+ method="GET",
+ path="/my/path",
+ rule=re.compile(r"^/my/path$"),
+ func=get_lambda,
+ cors=False,
+ compress=False,
+ )
+ route.request_param_name = "req"
+ route.request_param_name_checked = True
+ app.append_context(_route=route, _route_args={})
+
+ # WHEN calling the adapter a second time (cache hit)
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN it still injects the Request using the cached param name
+ assert result.status_code == 200
+
+
+def test_adapter_resolves_dependencies():
+ # GIVEN an async handler with Depends() parameters
+ app = APIGatewayHttpResolver()
+
+ def get_greeting() -> str:
+ return "hello"
+
+ async def get_lambda(greeting: Annotated[str, Depends(get_greeting)]):
+ return {"greeting": greeting}
+
+ _setup_resolver_context(app, API_RESTV2_EVENT)
+ route = Route(
+ method="GET",
+ path="/my/path",
+ rule=re.compile(r"^/my/path$"),
+ func=get_lambda,
+ cors=False,
+ compress=False,
+ )
+ app.append_context(_route=route, _route_args={})
+
+ # WHEN calling the adapter
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN dependencies are resolved and injected
+ assert result.status_code == 200
+
+
+def test_adapter_resolves_dependencies_with_sync_handler():
+ # GIVEN a sync handler with Depends() parameters
+ app = APIGatewayHttpResolver()
+
+ def get_greeting() -> str:
+ return "hello"
+
+ def get_lambda(greeting: Annotated[str, Depends(get_greeting)]):
+ return {"greeting": greeting}
+
+ _setup_resolver_context(app, API_RESTV2_EVENT)
+ route = Route(
+ method="GET",
+ path="/my/path",
+ rule=re.compile(r"^/my/path$"),
+ func=get_lambda,
+ cors=False,
+ compress=False,
+ )
+ app.append_context(_route=route, _route_args={})
+
+ # WHEN calling the adapter with a sync handler that has dependencies
+ result = asyncio.run(
+ _registered_api_adapter_async(app, get_lambda),
+ )
+
+ # THEN dependencies are resolved and injected for sync handler too
+ assert result.status_code == 200
diff --git a/tests/functional/event_handler/required_dependencies/test_request.py b/tests/functional/event_handler/required_dependencies/test_request.py
new file mode 100644
index 00000000000..b00ae6659ba
--- /dev/null
+++ b/tests/functional/event_handler/required_dependencies/test_request.py
@@ -0,0 +1,669 @@
+"""Tests for the Request object feature (GH #7992).
+
+Covers:
+- ``app.request`` availability in global and route-level middleware
+- ``Request`` type-annotation injection in route handlers
+- ``Request`` properties: route, path_parameters, method, headers, query_parameters, body
+- ``RuntimeError`` when ``app.request`` is accessed outside of resolution
+- Backward compatibility: routes without ``Request`` continue to work unchanged
+- ``APIGatewayHttpResolver`` and ``ALBResolver`` variants
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from aws_lambda_powertools.event_handler import (
+ ALBResolver,
+ APIGatewayHttpResolver,
+ APIGatewayRestResolver,
+ Request,
+ Response,
+)
+from tests.functional.utils import load_event
+
+if TYPE_CHECKING:
+ from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
+
+# ---------------------------------------------------------------------------
+# Shared test events
+# ---------------------------------------------------------------------------
+
+API_REST_EVENT = load_event("apiGatewayProxyEvent.json") # GET /my/path
+API_RESTV2_EVENT = load_event("apiGatewayProxyV2Event_GET.json")
+
+
+def _make_rest_event(path: str, method: str = "GET", path_parameters: dict | None = None, body: str | None = None):
+ """Build a minimal API Gateway REST (v1) proxy event."""
+ return {
+ "httpMethod": method,
+ "path": path,
+ "pathParameters": path_parameters,
+ "queryStringParameters": None,
+ "multiValueQueryStringParameters": None,
+ "headers": {"Content-Type": "application/json", "user-agent": "pytest"},
+ "multiValueHeaders": {},
+ "body": body,
+ "isBase64Encoded": False,
+ "requestContext": {"httpMethod": method, "resourcePath": path},
+ "resource": path,
+ "stageVariables": None,
+ }
+
+
+# ---------------------------------------------------------------------------
+# app.request in global middleware
+# ---------------------------------------------------------------------------
+
+
+def test_request_available_in_global_middleware():
+ app = APIGatewayRestResolver()
+ captured: list[Request] = []
+
+ def capture_middleware(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
+ captured.append(app.request)
+ return next_middleware(app)
+
+ app.use(middlewares=[capture_middleware])
+
+ @app.get("/my/path")
+ def handler():
+ return {}
+
+ app(API_REST_EVENT, {})
+
+ assert len(captured) == 1
+ req = captured[0]
+ assert isinstance(req, Request)
+ assert req.route == "/my/path"
+ assert req.method == "GET"
+
+
+def test_request_route_pattern_uses_openapi_format():
+ """route property should use {param} OpenAPI notation, not Powertools notation."""
+ app = APIGatewayRestResolver()
+ captured: list[Request] = []
+
+ def mw(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
+ captured.append(app.request)
+ return next_middleware(app)
+
+ app.use(middlewares=[mw])
+
+ @app.get("/applications/