@@ -343,3 +343,95 @@ def test_schema_dict_model_dump_handling(self):
343343 # This should call model_dump and process the result
344344 fix_upload_file_schema_references (mock_schema )
345345 mock_schema .model_dump .assert_called_once_with (by_alias = True )
346+
347+ def test_openapi_validation_webkit_boundary_extraction (self ):
348+ """Test WebKit boundary extraction in multipart parsing."""
349+ from aws_lambda_powertools .event_handler .middlewares .openapi_validation import (
350+ OpenAPIRequestValidationMiddleware ,
351+ )
352+
353+ middleware = OpenAPIRequestValidationMiddleware ()
354+
355+ # Test WebKit boundary format
356+ webkit_content_type = "multipart/form-data; WebKitFormBoundary123ABC"
357+ boundary_bytes = middleware ._extract_boundary_bytes (webkit_content_type )
358+ assert b"WebKitFormBoundary123ABC" in boundary_bytes
359+
360+ # Test missing boundary entirely
361+ try :
362+ middleware ._extract_boundary_bytes ("multipart/form-data" )
363+ raise AssertionError ("Should have raised ValueError" )
364+ except ValueError as e :
365+ assert "No boundary found" in str (e )
366+
367+ def test_openapi_validation_base64_decoding_error (self ):
368+ """Test base64 decoding error handling in body processing."""
369+ from aws_lambda_powertools .event_handler import APIGatewayRestResolver
370+ from aws_lambda_powertools .event_handler .middlewares .openapi_validation import (
371+ OpenAPIRequestValidationMiddleware ,
372+ )
373+
374+ middleware = OpenAPIRequestValidationMiddleware ()
375+ app = APIGatewayRestResolver ()
376+
377+ # Mock an event with invalid base64 content
378+ mock_event = Mock ()
379+ mock_event .body = "invalid_base64_content"
380+ mock_event .is_base64_encoded = True
381+ app .current_event = mock_event
382+
383+ # Should handle base64 decode error gracefully
384+ result = middleware ._decode_request_body (app )
385+ assert isinstance (result , (bytes , str ))
386+
387+ def test_openapi_validation_multipart_name_extraction (self ):
388+ """Test name extraction from multipart sections."""
389+ from aws_lambda_powertools .event_handler .middlewares .openapi_validation import (
390+ OpenAPIRequestValidationMiddleware ,
391+ )
392+
393+ middleware = OpenAPIRequestValidationMiddleware ()
394+
395+ # Test section without name parameter
396+ section_without_name = b"Content-Disposition: form-data\r \n \r \n test_content"
397+ field_name , content = middleware ._parse_multipart_section (section_without_name )
398+ assert field_name is None
399+ assert content == b""
400+
401+ # Test section with name parameter
402+ section_with_name = b'Content-Disposition: form-data; name="test_field"\r \n \r \n test_content'
403+ field_name , content = middleware ._parse_multipart_section (section_with_name )
404+ assert field_name == "test_field"
405+ assert content == "test_content" # Method returns string, not bytes
406+
407+ def test_openapi_validation_attribute_error_handling (self ):
408+ """Test AttributeError handling in field value extraction."""
409+ from aws_lambda_powertools .event_handler .openapi .compat import get_missing_field_error
410+
411+ # Create a mock object that raises AttributeError on get()
412+ class MockBodyWithAttributeError :
413+ def get (self , key ):
414+ raise AttributeError ("Mock AttributeError" )
415+
416+ # Create a mock field
417+ mock_field = Mock ()
418+ mock_field .alias = "test_field"
419+ mock_field .required = True
420+
421+ # Test with the problematic body object
422+ mock_body = MockBodyWithAttributeError ()
423+
424+ # This should trigger the AttributeError handling path
425+ errors = []
426+ loc = ("body" , "test_field" )
427+
428+ # Test the specific condition that triggers AttributeError handling
429+ value = None
430+ if mock_body is not None and value is None :
431+ try :
432+ mock_body .get (mock_field .alias )
433+ except AttributeError :
434+ errors .append (get_missing_field_error (loc ))
435+
436+ assert len (errors ) == 1
437+ assert "test_field" in str (errors [0 ])
0 commit comments