Add positive functional tests for SqsS3EventNotificationModel

Since the SqsS3EventNotificationModel is only responsible for chaining
(rather than redefining) the existing SqsModel and S3Model models, let's
re-use both the existing stubs and their corresponding assertions.

Author: theipster

tests/functional/parser/test_s3.py

@@ -6,8 +6,7 @@
 from tests.functional.utils import load_event
 
 
-@event_parser(model=S3Model)
-def handle_s3(event: S3Model, _: LambdaContext):
+def assert_s3(event: S3Model):
     records = list(event.Records)
     assert len(records) == 1
     record: S3RecordModel = records[0]
@@ -41,6 +40,11 @@ def handle_s3(event: S3Model, _: LambdaContext):
     assert record.glacierEventData is None
 
 
+@event_parser(model=S3Model)
+def handle_s3(event: S3Model, _: LambdaContext):
+    assert_s3(event)
+
+
 @event_parser(model=S3Model)
 def handle_s3_glacier(event: S3Model, _: LambdaContext):
     records = list(event.Records)

tests/functional/parser/test_sqs_s3_event_notification.py

@@ -0,0 +1,32 @@
+from aws_lambda_powertools.utilities.parser import event_parser
+from aws_lambda_powertools.utilities.parser.models import (
+    SqsS3EventNotificationModel,
+    S3Model,
+)
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from tests.functional.parser.schemas import MyAdvancedSqsBusiness
+from tests.functional.utils import json_serialize, load_event
+
+@event_parser(model=SqsS3EventNotificationModel)
+def handle_sqs_json_body_containing_s3_notifications(event: SqsS3EventNotificationModel, _: LambdaContext):
+    return event
+
+@event_parser(model=MyAdvancedSqsBusiness)
+def generate_sqs_event() -> MyAdvancedSqsBusiness:
+    return load_event("sqsEvent.json")
+
+@event_parser(model=S3Model)
+def generate_s3_event_notification() -> S3Model:
+    return load_event("s3Event.json")
+
+def test_handle_sqs_json_body_containing_s3_notifications():
+    sqs_event = generate_sqs_event()
+    s3_event_notification = generate_s3_event_notification()
+    for record in sqs_event.Records:
+        record.body = json_serialize(s3_event_notification)
+
+    parsed_event: SqsS3EventNotificationModel = handle_sqs_json_body_containing_s3_notifications(sqs_event, LambdaContext())
+
+    assert len(parsed_event.Records) == 2
+    for parsed_sqs_record in parsed_event.Records:
+        assert_s3(parsed_sqs_record.body)