refactor(data_classes): simplify nested data classes (#6289)

* refactor(data_classes): simplify nested data classes

Make each nested data class only know its own properties instead of
having to reference from the root of the data. This hopefully simplifies
the code and avoids DRY. Also simplify test cases.

* refactor(data_classes): fix tests

* refactor(data_classes): push another small commit to see if GH works

---------

Co-authored-by: Leandro Damascena <[email protected]>

Author: ericbn

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -18,7 +18,7 @@ class ALBEventRequestContext(DictWrapper):
     @property
     def elb_target_group_arn(self) -> str:
         """Target group arn for your Lambda function"""
-        return self["requestContext"]["elb"]["targetGroupArn"]
+        return self["elb"]["targetGroupArn"]
 
 
 class ALBEvent(BaseProxyEvent):
@@ -32,7 +32,7 @@ class ALBEvent(BaseProxyEvent):
 
     @property
     def request_context(self) -> ALBEventRequestContext:
-        return ALBEventRequestContext(self._data)
+        return ALBEventRequestContext(self["requestContext"])
 
     @property
     def multi_value_query_string_parameters(self) -> dict[str, list[str]]:

aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py

@@ -183,7 +183,7 @@ def stage_variables(self) -> dict[str, str]:
 
     @property
     def request_context(self) -> BaseRequestContext:
-        return BaseRequestContext(self._data)
+        return BaseRequestContext(self["requestContext"])
 
     @overload
     def get_header_value(
@@ -306,7 +306,7 @@ def query_string_parameters(self) -> dict[str, str]:
 
     @property
     def request_context(self) -> BaseRequestContextV2:
-        return BaseRequestContextV2(self._data)
+        return BaseRequestContextV2(self["requestContext"])
 
     @property
     def path_parameters(self) -> dict[str, str]:

aws_lambda_powertools/utilities/data_classes/api_gateway_proxy_event.py

@@ -61,42 +61,41 @@ class APIGatewayEventRequestContext(BaseRequestContext):
     @property
     def connected_at(self) -> int | None:
         """The Epoch-formatted connection time. (WebSocket API)"""
-        return self["requestContext"].get("connectedAt")
+        return self.get("connectedAt")
 
     @property
     def connection_id(self) -> str | None:
         """A unique ID for the connection that can be used to make a callback to the client. (WebSocket API)"""
-        return self["requestContext"].get("connectionId")
+        return self.get("connectionId")
 
     @property
     def event_type(self) -> str | None:
         """The event type: `CONNECT`, `MESSAGE`, or `DISCONNECT`. (WebSocket API)"""
-        return self["requestContext"].get("eventType")
+        return self.get("eventType")
 
     @property
     def message_direction(self) -> str | None:
         """Message direction (WebSocket API)"""
-        return self["requestContext"].get("messageDirection")
+        return self.get("messageDirection")
 
     @property
     def message_id(self) -> str | None:
         """A unique server-side ID for a message. Available only when the `eventType` is `MESSAGE`."""
-        return self["requestContext"].get("messageId")
+        return self.get("messageId")
 
     @property
     def operation_name(self) -> str | None:
         """The name of the operation being performed"""
-        return self["requestContext"].get("operationName")
+        return self.get("operationName")
 
     @property
     def route_key(self) -> str | None:
         """The selected route key."""
-        return self["requestContext"].get("routeKey")
+        return self.get("routeKey")
 
     @property
     def authorizer(self) -> APIGatewayEventAuthorizer:
-        authz_data = self._data.get("requestContext", {}).get("authorizer", {})
-        return APIGatewayEventAuthorizer(authz_data)
+        return APIGatewayEventAuthorizer(self.get("authorizer") or {})
 
 
 class APIGatewayProxyEvent(BaseProxyEvent):
@@ -136,7 +135,7 @@ def resolved_headers_field(self) -> dict[str, Any]:
 
     @property
     def request_context(self) -> APIGatewayEventRequestContext:
-        return APIGatewayEventRequestContext(self._data)
+        return APIGatewayEventRequestContext(self["requestContext"])
 
     @property
     def path_parameters(self) -> dict[str, str]:
@@ -248,8 +247,7 @@ def iam(self) -> RequestContextV2AuthorizerIam:
 class RequestContextV2(BaseRequestContextV2):
     @property
     def authorizer(self) -> RequestContextV2Authorizer:
-        ctx = self.get("requestContext") or {}  # key might exist but can be `null`
-        return RequestContextV2Authorizer(ctx.get("authorizer", {}))
+        return RequestContextV2Authorizer(self.get("authorizer") or {})
 
 
 class APIGatewayProxyEventV2(BaseProxyEvent):
@@ -291,7 +289,7 @@ def cookies(self) -> list[str]:
 
     @property
     def request_context(self) -> RequestContextV2:
-        return RequestContextV2(self._data)
+        return RequestContextV2(self["requestContext"])
 
     @property
     def path_parameters(self) -> dict[str, str]:

aws_lambda_powertools/utilities/data_classes/appsync_authorizer_event.py

@@ -11,32 +11,32 @@ class AppSyncAuthorizerEventRequestContext(DictWrapper):
     @property
     def api_id(self) -> str:
         """AppSync API ID"""
-        return self["requestContext"]["apiId"]
+        return self["apiId"]
 
     @property
     def account_id(self) -> str:
         """AWS Account ID"""
-        return self["requestContext"]["accountId"]
+        return self["accountId"]
 
     @property
     def request_id(self) -> str:
         """Requestt ID"""
-        return self["requestContext"]["requestId"]
+        return self["requestId"]
 
     @property
     def query_string(self) -> str:
         """GraphQL query string"""
-        return self["requestContext"]["queryString"]
+        return self["queryString"]
 
     @property
     def operation_name(self) -> str | None:
         """GraphQL operation name, optional"""
-        return self["requestContext"].get("operationName")
+        return self.get("operationName")
 
     @property
     def variables(self) -> dict:
         """GraphQL variables"""
-        return self["requestContext"]["variables"]
+        return self["variables"]
 
 
 class AppSyncAuthorizerEvent(DictWrapper):
@@ -57,7 +57,7 @@ def authorization_token(self) -> str:
     @property
     def request_context(self) -> AppSyncAuthorizerEventRequestContext:
         """Request context"""
-        return AppSyncAuthorizerEventRequestContext(self._data)
+        return AppSyncAuthorizerEventRequestContext(self["requestContext"])
 
 
 class AppSyncAuthorizerResponse:

aws_lambda_powertools/utilities/data_classes/cognito_user_pool_event.py

@@ -205,7 +205,6 @@ def json_body(self) -> Any:
         """Parses the submitted body as json"""
         if self.decoded_body:
             return self._json_deserializer(self.decoded_body)
-
         return None
 
     @cached_property
@@ -370,235 +369,235 @@ def validity_not_before(self) -> str:
 class APIGatewayEventIdentity(DictWrapper):
     @property
     def access_key(self) -> str | None:
-        return self["requestContext"]["identity"].get("accessKey")
+        return self.get("accessKey")
 
     @property
     def account_id(self) -> str | None:
         """The AWS account ID associated with the request."""
-        return self["requestContext"]["identity"].get("accountId")
+        return self.get("accountId")
 
     @property
     def api_key(self) -> str | None:
         """For API methods that require an API key, this variable is the API key associated with the method request.
         For methods that don't require an API key, this variable is null."""
-        return self["requestContext"]["identity"].get("apiKey")
+        return self.get("apiKey")
 
     @property
     def api_key_id(self) -> str | None:
         """The API key ID associated with an API request that requires an API key."""
-        return self["requestContext"]["identity"].get("apiKeyId")
+        return self.get("apiKeyId")
 
     @property
     def caller(self) -> str | None:
         """The principal identifier of the caller making the request."""
-        return self["requestContext"]["identity"].get("caller")
+        return self.get("caller")
 
     @property
     def cognito_authentication_provider(self) -> str | None:
         """A comma-separated list of the Amazon Cognito authentication providers used by the caller
         making the request. Available only if the request was signed with Amazon Cognito credentials."""
-        return self["requestContext"]["identity"].get("cognitoAuthenticationProvider")
+        return self.get("cognitoAuthenticationProvider")
 
     @property
     def cognito_authentication_type(self) -> str | None:
         """The Amazon Cognito authentication type of the caller making the request.
         Available only if the request was signed with Amazon Cognito credentials."""
-        return self["requestContext"]["identity"].get("cognitoAuthenticationType")
+        return self.get("cognitoAuthenticationType")
 
     @property
     def cognito_identity_id(self) -> str | None:
         """The Amazon Cognito identity ID of the caller making the request.
         Available only if the request was signed with Amazon Cognito credentials."""
-        return self["requestContext"]["identity"].get("cognitoIdentityId")
+        return self.get("cognitoIdentityId")
 
     @property
     def cognito_identity_pool_id(self) -> str | None:
         """The Amazon Cognito identity pool ID of the caller making the request.
         Available only if the request was signed with Amazon Cognito credentials."""
-        return self["requestContext"]["identity"].get("cognitoIdentityPoolId")
+        return self.get("cognitoIdentityPoolId")
 
     @property
     def principal_org_id(self) -> str | None:
         """The AWS organization ID."""
-        return self["requestContext"]["identity"].get("principalOrgId")
+        return self.get("principalOrgId")
 
     @property
     def source_ip(self) -> str:
         """The source IP address of the TCP connection making the request to API Gateway."""
-        return self["requestContext"]["identity"]["sourceIp"]
+        return self["sourceIp"]
 
     @property
     def user(self) -> str | None:
         """The principal identifier of the user making the request."""
-        return self["requestContext"]["identity"].get("user")
+        return self.get("user")
 
     @property
     def user_agent(self) -> str | None:
         """The User Agent of the API caller."""
-        return self["requestContext"]["identity"].get("userAgent")
+        return self.get("userAgent")
 
     @property
     def user_arn(self) -> str | None:
         """The Amazon Resource Name (ARN) of the effective user identified after authentication."""
-        return self["requestContext"]["identity"].get("userArn")
+        return self.get("userArn")
 
     @property
     def client_cert(self) -> RequestContextClientCert | None:
-        client_cert = self["requestContext"]["identity"].get("clientCert")
+        client_cert = self.get("clientCert")
         return None if client_cert is None else RequestContextClientCert(client_cert)
 
 
 class BaseRequestContext(DictWrapper):
     @property
     def account_id(self) -> str:
         """The AWS account ID associated with the request."""
-        return self["requestContext"]["accountId"]
+        return self["accountId"]
 
     @property
     def api_id(self) -> str:
         """The identifier API Gateway assigns to your API."""
-        return self["requestContext"]["apiId"]
+        return self["apiId"]
 
     @property
     def domain_name(self) -> str | None:
         """A domain name"""
-        return self["requestContext"].get("domainName")
+        return self.get("domainName")
 
     @property
     def domain_prefix(self) -> str | None:
-        return self["requestContext"].get("domainPrefix")
+        return self.get("domainPrefix")
 
     @property
     def extended_request_id(self) -> str | None:
         """An automatically generated ID for the API call, which contains more useful information
         for debugging/troubleshooting."""
-        return self["requestContext"].get("extendedRequestId")
+        return self.get("extendedRequestId")
 
     @property
     def protocol(self) -> str:
         """The request protocol, for example, HTTP/1.1."""
-        return self["requestContext"]["protocol"]
+        return self["protocol"]
 
     @property
     def http_method(self) -> str:
         """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
-        return self["requestContext"]["httpMethod"]
+        return self["httpMethod"]
 
     @property
     def identity(self) -> APIGatewayEventIdentity:
-        return APIGatewayEventIdentity(self._data)
+        return APIGatewayEventIdentity(self["identity"])
 
     @property
     def path(self) -> str:
-        return self["requestContext"]["path"]
+        return self["path"]
 
     @property
     def stage(self) -> str:
         """The deployment stage of the API request"""
-        return self["requestContext"]["stage"]
+        return self["stage"]
 
     @property
     def request_id(self) -> str:
         """The ID that API Gateway assigns to the API request."""
-        return self["requestContext"]["requestId"]
+        return self["requestId"]
 
     @property
     def request_time(self) -> str | None:
         """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)"""
-        return self["requestContext"].get("requestTime")
+        return self.get("requestTime")
 
     @property
     def request_time_epoch(self) -> int:
         """The Epoch-formatted request time."""
-        return self["requestContext"]["requestTimeEpoch"]
+        return self["requestTimeEpoch"]
 
     @property
     def resource_id(self) -> str:
-        return self["requestContext"]["resourceId"]
+        return self["resourceId"]
 
     @property
     def resource_path(self) -> str:
-        return self["requestContext"]["resourcePath"]
+        return self["resourcePath"]
 
 
 class RequestContextV2Http(DictWrapper):
     @property
     def method(self) -> str:
-        return self["requestContext"]["http"]["method"]
+        return self["method"]
 
     @property
     def path(self) -> str:
-        return self["requestContext"]["http"]["path"]
+        return self["path"]
 
     @property
     def protocol(self) -> str:
         """The request protocol, for example, HTTP/1.1."""
-        return self["requestContext"]["http"]["protocol"]
+        return self["protocol"]
 
     @property
     def source_ip(self) -> str:
         """The source IP address of the TCP connection making the request to API Gateway."""
-        return self["requestContext"]["http"]["sourceIp"]
+        return self["sourceIp"]
 
     @property
     def user_agent(self) -> str:
         """The User Agent of the API caller."""
-        return self["requestContext"]["http"]["userAgent"]
+        return self["userAgent"]
 
 
 class BaseRequestContextV2(DictWrapper):
     @property
     def account_id(self) -> str:
         """The AWS account ID associated with the request."""
-        return self["requestContext"]["accountId"]
+        return self["accountId"]
 
     @property
     def api_id(self) -> str:
         """The identifier API Gateway assigns to your API."""
-        return self["requestContext"]["apiId"]
+        return self["apiId"]
 
     @property
     def domain_name(self) -> str:
         """A domain name"""
-        return self["requestContext"]["domainName"]
+        return self["domainName"]
 
     @property
     def domain_prefix(self) -> str:
-        return self["requestContext"]["domainPrefix"]
+        return self["domainPrefix"]
 
     @property
     def http(self) -> RequestContextV2Http:
-        return RequestContextV2Http(self._data)
+        return RequestContextV2Http(self["http"])
 
     @property
     def request_id(self) -> str:
         """The ID that API Gateway assigns to the API request."""
-        return self["requestContext"]["requestId"]
+        return self["requestId"]
 
     @property
     def route_key(self) -> str:
         """The selected route key."""
-        return self["requestContext"]["routeKey"]
+        return self["routeKey"]
 
     @property
     def stage(self) -> str:
         """The deployment stage of the API request"""
-        return self["requestContext"]["stage"]
+        return self["stage"]
 
     @property
     def time(self) -> str:
         """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)."""
-        return self["requestContext"]["time"]
+        return self["time"]
 
     @property
     def time_epoch(self) -> int:
         """The Epoch-formatted request time."""
-        return self["requestContext"]["timeEpoch"]
+        return self["timeEpoch"]
 
     @property
     def authentication(self) -> RequestContextClientCert | None:
         """Optional when using mutual TLS authentication"""
         # FunctionURL might have NONE as AuthZ
-        authentication = self["requestContext"].get("authentication") or {}
+        authentication = self.get("authentication") or {}
         client_cert = authentication.get("clientCert")
         return None if client_cert is None else RequestContextClientCert(client_cert)

aws_lambda_powertools/utilities/data_classes/common.py

@@ -177,38 +177,33 @@ def asdict(self) -> dict:
 
 
 class KinesisFirehoseRecordMetadata(DictWrapper):
-    @property
-    def _metadata(self) -> dict:
-        """Optional: metadata associated with this record; present only when Kinesis Stream is source"""
-        return self["kinesisRecordMetadata"]  # could raise KeyError
-
     @property
     def shard_id(self) -> str:
         """Kinesis stream shard ID; present only when Kinesis Stream is source"""
-        return self._metadata["shardId"]
+        return self["shardId"]
 
     @property
     def partition_key(self) -> str:
         """Kinesis stream partition key; present only when Kinesis Stream is source"""
-        return self._metadata["partitionKey"]
+        return self["partitionKey"]
 
     @property
     def approximate_arrival_timestamp(self) -> int:
         """Kinesis stream approximate arrival ISO timestamp; present only when Kinesis Stream is source"""
-        return self._metadata["approximateArrivalTimestamp"]
+        return self["approximateArrivalTimestamp"]
 
     @property
     def sequence_number(self) -> str:
         """Kinesis stream sequence number; present only when Kinesis Stream is source"""
-        return self._metadata["sequenceNumber"]
+        return self["sequenceNumber"]
 
     @property
     def subsequence_number(self) -> int:
         """Kinesis stream sub-sequence number; present only when Kinesis Stream is source
 
         Note: this will only be present for Kinesis streams using record aggregation
         """
-        return self._metadata["subsequenceNumber"]
+        return self["subsequenceNumber"]
 
 
 class KinesisFirehoseRecord(DictWrapper):
@@ -230,7 +225,8 @@ def data(self) -> str:
     @property
     def metadata(self) -> KinesisFirehoseRecordMetadata | None:
         """Optional: metadata associated with this record; present only when Kinesis Stream is source"""
-        return KinesisFirehoseRecordMetadata(self._data) if self.get("kinesisRecordMetadata") else None
+        metadata = self.get("kinesisRecordMetadata")
+        return KinesisFirehoseRecordMetadata(metadata) if metadata else None
 
     @property
     def data_as_bytes(self) -> bytes:

aws_lambda_powertools/utilities/data_classes/kinesis_firehose_event.py

@@ -15,27 +15,27 @@ class KinesisStreamRecordPayload(DictWrapper):
     @property
     def approximate_arrival_timestamp(self) -> float:
         """The approximate time that the record was inserted into the stream"""
-        return float(self["kinesis"]["approximateArrivalTimestamp"])
+        return float(self["approximateArrivalTimestamp"])
 
     @property
     def data(self) -> str:
         """The data blob"""
-        return self["kinesis"]["data"]
+        return self["data"]
 
     @property
     def kinesis_schema_version(self) -> str:
         """Schema version for the record"""
-        return self["kinesis"]["kinesisSchemaVersion"]
+        return self["kinesisSchemaVersion"]
 
     @property
     def partition_key(self) -> str:
         """Identifies which shard in the stream the data record is assigned to"""
-        return self["kinesis"]["partitionKey"]
+        return self["partitionKey"]
 
     @property
     def sequence_number(self) -> str:
         """The unique identifier of the record within its shard"""
-        return self["kinesis"]["sequenceNumber"]
+        return self["sequenceNumber"]
 
     def data_as_bytes(self) -> bytes:
         """Decode binary encoded data as bytes"""
@@ -94,7 +94,7 @@ def invoke_identity_arn(self) -> str:
     @property
     def kinesis(self) -> KinesisStreamRecordPayload:
         """Underlying Kinesis record associated with the event"""
-        return KinesisStreamRecordPayload(self._data)
+        return KinesisStreamRecordPayload(self["kinesis"])
 
 
 class KinesisStreamEvent(DictWrapper):

aws_lambda_powertools/utilities/data_classes/kinesis_stream_event.py

@@ -18,7 +18,7 @@ def principal_id(self) -> str:
 class S3RequestParameters(DictWrapper):
     @property
     def source_ip_address(self) -> str:
-        return self["requestParameters"]["sourceIPAddress"]
+        return self["sourceIPAddress"]
 
 
 class S3EventNotificationEventBridgeBucket(DictWrapper):
@@ -40,8 +40,8 @@ def size(self) -> int | None:
 
     @property
     def etag(self) -> str:
-        """Object etag. Object deletion event doesn't contain etag; we default to empty string"""
-        return self.get("etag", "")  # type: ignore[return-value]  # false positive
+        """Object eTag. Object deletion event doesn't contain eTag; we default to empty string"""
+        return self.get("etag") or ""
 
     @property
     def version_id(self) -> str:
@@ -156,77 +156,77 @@ def detail(self) -> S3EventBridgeNotificationDetail:  # type: ignore[override]
 class S3Bucket(DictWrapper):
     @property
     def name(self) -> str:
-        return self["s3"]["bucket"]["name"]
+        return self["name"]
 
     @property
     def owner_identity(self) -> S3Identity:
-        return S3Identity(self["s3"]["bucket"]["ownerIdentity"])
+        return S3Identity(self["ownerIdentity"])
 
     @property
     def arn(self) -> str:
-        return self["s3"]["bucket"]["arn"]
+        return self["arn"]
 
 
 class S3Object(DictWrapper):
     @property
     def key(self) -> str:
         """Object key"""
-        return self["s3"]["object"]["key"]
+        return self["key"]
 
     @property
     def size(self) -> int:
         """Object byte size"""
-        return int(self["s3"]["object"]["size"])
+        return int(self["size"])
 
     @property
     def etag(self) -> str:
         """Object eTag. Object deletion event doesn't contain eTag; we default to empty string"""
-        return self["s3"]["object"].get("eTag", "")
+        return self.get("eTag") or ""
 
     @property
     def version_id(self) -> str | None:
         """Object version if bucket is versioning-enabled, otherwise null"""
-        return self["s3"]["object"].get("versionId")
+        return self.get("versionId")
 
     @property
     def sequencer(self) -> str:
         """A string representation of a hexadecimal value used to determine event sequence,
         only used with PUTs and DELETEs
         """
-        return self["s3"]["object"]["sequencer"]
+        return self["sequencer"]
 
 
 class S3Message(DictWrapper):
     @property
     def s3_schema_version(self) -> str:
-        return self["s3"]["s3SchemaVersion"]
+        return self["s3SchemaVersion"]
 
     @property
     def configuration_id(self) -> str:
         """ID found in the bucket notification configuration"""
-        return self["s3"]["configurationId"]
+        return self["configurationId"]
 
     @property
     def bucket(self) -> S3Bucket:
-        return S3Bucket(self._data)
+        return S3Bucket(self["bucket"])
 
     @property
     def get_object(self) -> S3Object:
         """Get the `object` property as an S3Object"""
         # Note: this name conflicts with existing python builtins
-        return S3Object(self._data)
+        return S3Object(self["object"])
 
 
 class S3EventRecordGlacierRestoreEventData(DictWrapper):
     @property
     def lifecycle_restoration_expiry_time(self) -> str:
         """Time when the object restoration will be expired."""
-        return self["restoreEventData"]["lifecycleRestorationExpiryTime"]
+        return self["lifecycleRestorationExpiryTime"]
 
     @property
     def lifecycle_restore_storage_class(self) -> str:
         """Source storage class for restore"""
-        return self["restoreEventData"]["lifecycleRestoreStorageClass"]
+        return self["lifecycleRestoreStorageClass"]
 
 
 class S3EventRecordGlacierEventData(DictWrapper):
@@ -236,7 +236,7 @@ def restore_event_data(self) -> S3EventRecordGlacierRestoreEventData:
 
         The glacierEventData key is only visible for s3:ObjectRestore:Completed events
         """
-        return S3EventRecordGlacierRestoreEventData(self._data)
+        return S3EventRecordGlacierRestoreEventData(self["restoreEventData"])
 
 
 class S3EventRecord(DictWrapper):
@@ -272,7 +272,7 @@ def user_identity(self) -> S3Identity:
 
     @property
     def request_parameters(self) -> S3RequestParameters:
-        return S3RequestParameters(self._data)
+        return S3RequestParameters(self["requestParameters"])
 
     @property
     def response_elements(self) -> dict[str, str]:
@@ -286,7 +286,7 @@ def response_elements(self) -> dict[str, str]:
 
     @property
     def s3(self) -> S3Message:
-        return S3Message(self._data)
+        return S3Message(self["s3"])
 
     @property
     def glacier_event_data(self) -> S3EventRecordGlacierEventData | None:

aws_lambda_powertools/utilities/data_classes/s3_event.py

@@ -212,11 +212,11 @@ def action(self) -> SESReceiptAction:
 class SESMessage(DictWrapper):
     @property
     def mail(self) -> SESMail:
-        return SESMail(self["ses"]["mail"])
+        return SESMail(self["mail"])
 
     @property
     def receipt(self) -> SESReceipt:
-        return SESReceipt(self["ses"]["receipt"])
+        return SESReceipt(self["receipt"])
 
 
 class SESEventRecord(DictWrapper):
@@ -232,7 +232,7 @@ def event_version(self) -> str:
 
     @property
     def ses(self) -> SESMessage:
-        return SESMessage(self._data)
+        return SESMessage(self["ses"])
 
 
 class SESEvent(DictWrapper):

aws_lambda_powertools/utilities/data_classes/ses_event.py

@@ -103,11 +103,11 @@ def test_cognito_custom_message_trigger_event():
     assert parsed_event.request.client_metadata == {}
 
     parsed_event.response.sms_message = "sms"
-    assert parsed_event.response.sms_message == parsed_event["response"]["smsMessage"]
+    assert parsed_event.response.sms_message == raw_event["response"]["smsMessage"]
     parsed_event.response.email_message = "email"
-    assert parsed_event.response.email_message == parsed_event["response"]["emailMessage"]
+    assert parsed_event.response.email_message == raw_event["response"]["emailMessage"]
     parsed_event.response.email_subject = "subject"
-    assert parsed_event.response.email_subject == parsed_event["response"]["emailSubject"]
+    assert parsed_event.response.email_subject == raw_event["response"]["emailSubject"]
 
 
 def test_cognito_custom_email_sender_trigger_event():
@@ -141,7 +141,7 @@ def test_cognito_pre_authentication_trigger_event():
     assert parsed_event.trigger_source == raw_event["triggerSource"]
 
     assert parsed_event.request.user_not_found is None
-    parsed_event["request"]["userNotFound"] = True
+    raw_event["request"]["userNotFound"] = True
     assert parsed_event.request.user_not_found is True
     assert parsed_event.request.user_attributes.get("email") == raw_event["request"]["userAttributes"]["email"]
     assert parsed_event.request.validation_data == {}
@@ -171,57 +171,42 @@ def test_cognito_pre_token_generation_trigger_event():
     assert parsed_event.request.user_attributes.get("email") == raw_event["request"]["userAttributes"]["email"]
     assert parsed_event.request.client_metadata == {}
 
-    parsed_event["request"]["groupConfiguration"]["preferredRole"] = "temp"
+    raw_event["request"]["groupConfiguration"]["preferredRole"] = "temp"
     group_configuration = parsed_event.request.group_configuration
     assert group_configuration.preferred_role == "temp"
 
-    assert parsed_event["response"].get("claimsOverrideDetails") is None
     claims_override_details = parsed_event.response.claims_override_details
-    assert parsed_event["response"]["claimsOverrideDetails"] == {}
-
     assert claims_override_details.claims_to_add_or_override == {}
     assert claims_override_details.claims_to_suppress == []
     assert claims_override_details.group_configuration is None
 
     claims_override_details.group_configuration = {}
     assert claims_override_details.group_configuration._data == {}
-    assert parsed_event["response"]["claimsOverrideDetails"]["groupOverrideDetails"] == {}
 
     expected_claims = {"test": "value"}
     claims_override_details.claims_to_add_or_override = expected_claims
     assert claims_override_details.claims_to_add_or_override["test"] == "value"
-    assert parsed_event["response"]["claimsOverrideDetails"]["claimsToAddOrOverride"] == expected_claims
 
     claims_override_details.claims_to_suppress = ["email"]
     assert claims_override_details.claims_to_suppress[0] == "email"
-    assert parsed_event["response"]["claimsOverrideDetails"]["claimsToSuppress"] == ["email"]
 
     expected_groups = ["group-A", "group-B"]
     claims_override_details.set_group_configuration_groups_to_override(expected_groups)
     assert claims_override_details.group_configuration.groups_to_override == expected_groups
-    assert (
-        parsed_event["response"]["claimsOverrideDetails"]["groupOverrideDetails"]["groupsToOverride"] == expected_groups
-    )
-    claims_override_details = parsed_event.response.claims_override_details
-    assert claims_override_details["groupOverrideDetails"]["groupsToOverride"] == expected_groups
 
     claims_override_details.set_group_configuration_iam_roles_to_override(["role"])
     assert claims_override_details.group_configuration.iam_roles_to_override == ["role"]
-    assert parsed_event["response"]["claimsOverrideDetails"]["groupOverrideDetails"]["iamRolesToOverride"] == ["role"]
 
     claims_override_details.set_group_configuration_preferred_role("role_name")
     assert claims_override_details.group_configuration.preferred_role == "role_name"
-    assert parsed_event["response"]["claimsOverrideDetails"]["groupOverrideDetails"]["preferredRole"] == "role_name"
 
     # Ensure that even if "claimsOverrideDetails" was explicitly set to None
     # accessing `event.response.claims_override_details` would set it to `{}`
-    parsed_event["response"]["claimsOverrideDetails"] = None
+    raw_event["response"]["claimsOverrideDetails"] = None
     claims_override_details = parsed_event.response.claims_override_details
     assert claims_override_details._data == {}
-    assert parsed_event["response"]["claimsOverrideDetails"] == {}
     claims_override_details.claims_to_suppress = ["email"]
     assert claims_override_details.claims_to_suppress[0] == "email"
-    assert parsed_event["response"]["claimsOverrideDetails"]["claimsToSuppress"] == ["email"]
 
 
 def test_cognito_pre_token_v2_generation_trigger_event():
@@ -236,15 +221,12 @@ def test_cognito_pre_token_v2_generation_trigger_event():
     assert parsed_event.request.user_attributes.get("email") == raw_event["request"]["userAttributes"]["email"]
     assert parsed_event.request.client_metadata == {}
 
-    parsed_event["request"]["groupConfiguration"]["preferredRole"] = "temp"
+    raw_event["request"]["groupConfiguration"]["preferredRole"] = "temp"
     group_configuration = parsed_event.request.group_configuration
     assert group_configuration.preferred_role == "temp"
     assert parsed_event.request.scopes == raw_event["request"]["scopes"]
 
-    assert parsed_event["response"].get("claimsAndScopeOverrideDetails") is None
     claims_scope_override_details = parsed_event.response.claims_scope_override_details
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"] == {}
-
     claims_scope_override_details.id_token_generation = claims_scope_override_details.access_token_generation = {}
     assert claims_scope_override_details.id_token_generation.claims_to_add_or_override == {}
     assert claims_scope_override_details.id_token_generation.claims_to_suppress == []
@@ -258,45 +240,24 @@ def test_cognito_pre_token_v2_generation_trigger_event():
 
     claims_scope_override_details.group_configuration = {}
     assert claims_scope_override_details.group_configuration._data == {}
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["groupOverrideDetails"] == {}
 
     expected_claims = {"test": "value"}
     claims_scope_override_details.id_token_generation.claims_to_add_or_override = expected_claims
     claims_scope_override_details.access_token_generation.claims_to_add_or_override = expected_claims
     assert claims_scope_override_details.id_token_generation.claims_to_add_or_override["test"] == "value"
     assert claims_scope_override_details.access_token_generation.claims_to_add_or_override["test"] == "value"
-    assert (
-        parsed_event["response"]["claimsAndScopeOverrideDetails"]["idTokenGeneration"]["claimsToAddOrOverride"]
-        == expected_claims
-    )
-    assert (
-        parsed_event["response"]["claimsAndScopeOverrideDetails"]["accessTokenGeneration"]["claimsToAddOrOverride"]
-        == expected_claims
-    )
 
     claims_scope_override_details.id_token_generation.claims_to_suppress = (
         claims_scope_override_details.access_token_generation.claims_to_suppress
     ) = ["email"]
     assert claims_scope_override_details.id_token_generation.claims_to_suppress[0] == "email"
     assert claims_scope_override_details.access_token_generation.claims_to_suppress[0] == "email"
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["idTokenGeneration"]["claimsToSuppress"] == [
-        "email",
-    ]
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["accessTokenGeneration"]["claimsToSuppress"] == [
-        "email",
-    ]
 
     claims_scope_override_details.id_token_generation.scopes_to_suppress = (
         claims_scope_override_details.access_token_generation.scopes_to_suppress
     ) = ["email"]
     assert claims_scope_override_details.id_token_generation.scopes_to_suppress[0] == "email"
     assert claims_scope_override_details.access_token_generation.scopes_to_suppress[0] == "email"
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["idTokenGeneration"]["scopesToSuppress"] == [
-        "email",
-    ]
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["accessTokenGeneration"]["scopesToSuppress"] == [
-        "email",
-    ]
 
     claims_scope_override_details.id_token_generation.scopes_to_add = (
         claims_scope_override_details.access_token_generation.scopes_to_add
@@ -305,47 +266,27 @@ def test_cognito_pre_token_v2_generation_trigger_event():
         claims_scope_override_details.id_token_generation.scopes_to_add[0] == "email"
         and claims_scope_override_details.access_token_generation.scopes_to_add[0] == "email"
     )
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["idTokenGeneration"]["scopesToAdd"] == ["email"]
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["accessTokenGeneration"]["scopesToAdd"] == [
-        "email",
-    ]
 
     expected_groups = ["group-A", "group-B"]
     claims_scope_override_details.set_group_configuration_groups_to_override(expected_groups)
     assert claims_scope_override_details.group_configuration.groups_to_override == expected_groups
-    assert (
-        parsed_event["response"]["claimsAndScopeOverrideDetails"]["groupOverrideDetails"]["groupsToOverride"]
-        == expected_groups
-    )
     claims_scope_override_details = parsed_event.response.claims_scope_override_details
-    assert claims_scope_override_details["groupOverrideDetails"]["groupsToOverride"] == expected_groups
 
     claims_scope_override_details.set_group_configuration_iam_roles_to_override(["role"])
     assert claims_scope_override_details.group_configuration.iam_roles_to_override == ["role"]
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["groupOverrideDetails"]["iamRolesToOverride"] == [
-        "role",
-    ]
 
     claims_scope_override_details.set_group_configuration_preferred_role("role_name")
     assert claims_scope_override_details.group_configuration.preferred_role == "role_name"
-    assert (
-        parsed_event["response"]["claimsAndScopeOverrideDetails"]["groupOverrideDetails"]["preferredRole"]
-        == "role_name"
-    )
 
     # Ensure that even if "claimsAndScopeOverrideDetails" was explicitly set to None
     # accessing `event.response.claims_scope_override_details` would set it to `{}`
-    parsed_event["response"]["claimsAndScopeOverrideDetails"] = None
+    raw_event["response"]["claimsAndScopeOverrideDetails"] = None
     claims_scope_override_details = parsed_event.response.claims_scope_override_details
     assert claims_scope_override_details._data == {}
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"] == {}
 
     claims_scope_override_details.id_token_generation = {}
     claims_scope_override_details.id_token_generation.claims_to_suppress = ["email"]
     assert claims_scope_override_details.id_token_generation.claims_to_suppress[0] == "email"
-    assert parsed_event["response"]["claimsAndScopeOverrideDetails"]["idTokenGeneration"]["claimsToSuppress"] == [
-        "email",
-    ]
 
 
 def test_cognito_define_auth_challenge_trigger_event():
@@ -367,14 +308,14 @@ def test_cognito_define_auth_challenge_trigger_event():
 
     # Verify setters
     parsed_event.response.challenge_name = "CUSTOM_CHALLENGE"
-    assert parsed_event.response.challenge_name == parsed_event["response"]["challengeName"]
+    assert parsed_event.response.challenge_name == raw_event["response"]["challengeName"]
     assert parsed_event.response.challenge_name == "CUSTOM_CHALLENGE"
     parsed_event.response.fail_authentication = True
     assert parsed_event.response.fail_authentication
-    assert parsed_event.response.fail_authentication == parsed_event["response"]["failAuthentication"]
+    assert parsed_event.response.fail_authentication == raw_event["response"]["failAuthentication"]
     parsed_event.response.issue_tokens = True
     assert parsed_event.response.issue_tokens
-    assert parsed_event.response.issue_tokens == parsed_event["response"]["issueTokens"]
+    assert parsed_event.response.issue_tokens == raw_event["response"]["issueTokens"]
 
 
 def test_create_auth_challenge_trigger_event():
@@ -395,13 +336,13 @@ def test_create_auth_challenge_trigger_event():
 
     # Verify setters
     parsed_event.response.public_challenge_parameters = {"test": "value"}
-    assert parsed_event.response.public_challenge_parameters == parsed_event["response"]["publicChallengeParameters"]
+    assert parsed_event.response.public_challenge_parameters == raw_event["response"]["publicChallengeParameters"]
     assert parsed_event.response.public_challenge_parameters["test"] == "value"
     parsed_event.response.private_challenge_parameters = {"private": "value"}
-    assert parsed_event.response.private_challenge_parameters == parsed_event["response"]["privateChallengeParameters"]
+    assert parsed_event.response.private_challenge_parameters == raw_event["response"]["privateChallengeParameters"]
     assert parsed_event.response.private_challenge_parameters["private"] == "value"
     parsed_event.response.challenge_metadata = "meta"
-    assert parsed_event.response.challenge_metadata == parsed_event["response"]["challengeMetadata"]
+    assert parsed_event.response.challenge_metadata == raw_event["response"]["challengeMetadata"]
     assert parsed_event.response.challenge_metadata == "meta"
 
 
@@ -423,5 +364,5 @@ def test_verify_auth_challenge_response_trigger_event():
 
     # Verify setters
     parsed_event.response.answer_correct = True
-    assert parsed_event.response.answer_correct == parsed_event["response"]["answerCorrect"]
+    assert parsed_event.response.answer_correct == raw_event["response"]["answerCorrect"]
     assert parsed_event.response.answer_correct

tests/unit/data_classes/required_dependencies/test_cognito_user_pool_event.py

@@ -29,7 +29,6 @@ def test_kinesis_stream_event():
 
     kinesis = record.kinesis
     kinesis_raw = raw_event["Records"][0]["kinesis"]
-    assert kinesis._data["kinesis"] == kinesis_raw
 
     assert kinesis.approximate_arrival_timestamp == kinesis_raw["approximateArrivalTimestamp"]
     assert kinesis.data == kinesis_raw["data"]