@@ -2683,9 +2683,12 @@ class DummyEvent:
26832683 class DummyApp :
26842684 current_event = DummyEvent ()
26852685 # Correct monkeypatch: replace parse_qs with a function that raises Exception
2686+ def _raise (* a , ** kw ):
2687+ raise Exception ("fail" )
2688+
26862689 monkeypatch .setattr (
26872690 "aws_lambda_powertools.event_handler.middlewares.openapi_validation.parse_qs" ,
2688- lambda * a , ** kw : ( _ for _ in ()). throw ( Exception ( "fail" ))
2691+ _raise ,
26892692 )
26902693 middleware = OpenAPIRequestValidationMiddleware ()
26912694 with pytest .raises (Exception ) as excinfo :
@@ -2697,7 +2700,150 @@ class DummyApp:
26972700def test_get_body_field_location_alias_omitted ():
26982701 """Test _get_body_field_location with field_alias_omitted True and False"""
26992702 from aws_lambda_powertools .event_handler .middlewares .openapi_validation import _get_body_field_location
2703+ from typing import cast
2704+ from aws_lambda_powertools .event_handler .openapi .compat import ModelField
2705+
27002706 class DummyField :
27012707 alias = "field_alias"
2702- assert _get_body_field_location (DummyField (), True ) == ("body" ,)
2703- assert _get_body_field_location (DummyField (), False ) == ("body" , "field_alias" )
2708+
2709+ # Cast the dummy to ModelField to satisfy static typing analyzers
2710+ dummy = cast (ModelField , DummyField ())
2711+
2712+ assert _get_body_field_location (dummy , True ) == ("body" ,)
2713+ assert _get_body_field_location (dummy , False ) == ("body" , "field_alias" )
2714+
2715+
2716+ def test_decode_request_body_fallback_on_invalid_base64 ():
2717+ """When is_base64_encoded is True but decoding fails, the raw body should be returned as bytes."""
2718+ from types import SimpleNamespace
2719+
2720+ current_event = SimpleNamespace (body = "not-base64!!" , is_base64_encoded = True )
2721+ app = SimpleNamespace (current_event = current_event )
2722+
2723+ result = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ]).OpenAPIRequestValidationMiddleware ()._decode_request_body (app )
2724+ assert isinstance (result , (bytes , bytearray ))
2725+ assert result == b"not-base64!!"
2726+
2727+
2728+ def test_normalize_field_value_uploadfile_with_annotated_bytes ():
2729+ """If a field_info.annotation is Annotated[bytes,...], UploadFile should be converted to bytes."""
2730+ from typing import Annotated
2731+ from aws_lambda_powertools .event_handler .openapi .params import UploadFile
2732+
2733+ file_content = b"hello"
2734+ upload = UploadFile (file = file_content , filename = "f" , content_type = "text/plain" )
2735+
2736+ class DummyFieldInfo :
2737+ annotation = Annotated [bytes , "meta" ]
2738+
2739+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2740+ result = ov ._normalize_field_value (upload , DummyFieldInfo ())
2741+ assert result == file_content
2742+
2743+
2744+ def test_convert_value_type_uploadfile_to_bytes_and_back ():
2745+ from aws_lambda_powertools .event_handler .openapi .params import UploadFile
2746+
2747+ file_content = b"data"
2748+ upload = UploadFile (file = file_content , filename = "f" , content_type = "text/plain" )
2749+
2750+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2751+ # Convert UploadFile to bytes when field_type is bytes
2752+ converted = ov ._convert_value_type (upload , bytes )
2753+ assert converted == file_content
2754+
2755+ # Convert bytes to UploadFile when field_type is UploadFile
2756+ converted_back = ov ._convert_value_type (file_content , UploadFile )
2757+ assert isinstance (converted_back , UploadFile )
2758+ assert converted_back .file == file_content
2759+
2760+
2761+ def test_get_embed_body_wraps_when_alias_omitted ():
2762+ from types import SimpleNamespace
2763+
2764+ field = SimpleNamespace (field_info = SimpleNamespace (embed = None ), alias = "my_field" )
2765+ required_params = [field ]
2766+ received_body = {"a" : 1 }
2767+
2768+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2769+ new_body , alias_omitted = ov ._get_embed_body (field = field , required_params = required_params , received_body = received_body )
2770+ assert alias_omitted is True
2771+ assert new_body == {"my_field" : received_body }
2772+
2773+
2774+ def test_split_section_headers_and_content_handles_crlf_and_lf ():
2775+ # CRLF version
2776+ section_crlf = b"Content-Disposition: form-data; name=\" f\" \r \n Content-Type: text/plain\r \n \r \n hello"
2777+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2778+ middleware = ov .OpenAPIRequestValidationMiddleware ()
2779+ headers_part , content = middleware ._split_section_headers_and_content (section_crlf )
2780+ assert "Content-Disposition" in headers_part
2781+ assert content == b"hello"
2782+
2783+ # LF version
2784+ section_lf = b"Content-Disposition: form-data; name=\" f\" \n \n world"
2785+ headers_part2 , content2 = middleware ._split_section_headers_and_content (section_lf )
2786+ assert "Content-Disposition" in headers_part2
2787+ assert content2 == b"world"
2788+
2789+
2790+ def test_parse_multipart_sections_creates_uploadfile_and_field ():
2791+ from aws_lambda_powertools .event_handler .openapi .params import UploadFile
2792+
2793+ boundary = b"--boundary"
2794+ # Build sections: empty preamble, a file part, a field part, and closing
2795+ file_section = (
2796+ b"\r \n " + boundary + b"\r \n "
2797+ + b'Content-Disposition: form-data; name="file"; filename="t.txt"\r \n '
2798+ + b"Content-Type: text/plain\r \n \r \n "
2799+ + b"file-content\r \n "
2800+ )
2801+ field_section = (
2802+ boundary + b"\r \n "
2803+ + b'Content-Disposition: form-data; name="field"\r \n \r \n '
2804+ + b"value\r \n "
2805+ )
2806+ closing = boundary + b"--\r \n "
2807+ decoded = file_section + field_section + closing
2808+
2809+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2810+ middleware = ov .OpenAPIRequestValidationMiddleware ()
2811+ parsed = middleware ._parse_multipart_sections (decoded , boundary )
2812+ assert "file" in parsed
2813+ assert isinstance (parsed ["file" ], UploadFile )
2814+ assert parsed ["file" ].file .strip () == b"file-content"
2815+ assert parsed ["field" ] == "value"
2816+
2817+
2818+ def test_decode_form_field_content_falls_back_to_bytes_on_decode_error ():
2819+ content = b"\xff "
2820+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2821+ middleware = ov .OpenAPIRequestValidationMiddleware ()
2822+ result = middleware ._decode_form_field_content (content )
2823+ assert isinstance (result , (bytes , bytearray ))
2824+
2825+
2826+ def test_extract_field_value_from_body_handles_attribute_error ():
2827+ from types import SimpleNamespace
2828+ errors = []
2829+ field = SimpleNamespace (alias = "x" )
2830+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2831+ val = ov ._extract_field_value_from_body (field , 123 , ("body" , "x" ), errors )
2832+ assert val is None
2833+ assert len (errors ) == 1
2834+
2835+
2836+ def test_resolve_field_type_returns_first_non_none_in_union ():
2837+ from typing import Union
2838+
2839+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2840+ resolved = ov ._resolve_field_type (Union [int , None ])
2841+ assert resolved == int
2842+
2843+
2844+ def test_normalize_field_value_list_to_single_when_not_sequence ():
2845+ class DummyFieldInfo :
2846+ annotation = int
2847+
2848+ ov = __import__ ("aws_lambda_powertools.event_handler.middlewares.openapi_validation" , fromlist = ["*" ])
2849+ assert ov ._normalize_field_value ([1 , 2 , 3 ], DummyFieldInfo ()) == 1
0 commit comments