ISSUE-1503: add StreamRecord tests

aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py

@@ -182,8 +182,10 @@ def approximate_creation_date_time(self) -> Optional[int]:
         item = self.get("ApproximateCreationDateTime")
         return None if item is None else int(item)
 
+    # This override breaks the Mapping protocol of DictWrapper, it's left here for backwards compatibility with
+    # a 'type: ignore' comment. This is currently the only subclass of DictWrapper that breaks this protocol.
     @property
-    def keys(self) -> Optional[Dict[str, AttributeValue]]:
+    def keys(self) -> Optional[Dict[str, AttributeValue]]:  # type: ignore
         """The primary key attribute(s) for the DynamoDB item that was modified."""
         return _attribute_value_dict(self._data, "Keys")
 

tests/functional/test_data_classes.py

@@ -74,6 +74,7 @@
     AttributeValueType,
     DynamoDBRecordEventName,
     DynamoDBStreamEvent,
+    StreamRecord,
     StreamViewType,
 )
 from aws_lambda_powertools.utilities.data_classes.event_source import event_source
@@ -630,6 +631,23 @@ def test_dynamo_attribute_value_type_error():
         print(attribute_value.get_type)
 
 
+def test_stream_record_keys_with_valid_keys():
+    attribute_value = {"Foo": "Bar"}
+    sr = StreamRecord({"Keys": {"Key1": attribute_value}})
+    assert sr.keys == {"Key1": AttributeValue(attribute_value)}
+
+
+def test_stream_record_keys_with_no_keys():
+    sr = StreamRecord({})
+    assert sr.keys is None
+
+
+def test_stream_record_keys_overrides_dict_wrapper_keys():
+    data = {"Keys": {"key1": {"attr1": "value1"}}}
+    sr = StreamRecord(data)
+    assert sr.keys != data.keys()
+
+
 def test_event_bridge_event():
     event = EventBridgeEvent(load_event("eventBridgeEvent.json"))