fix ResponseMetaDataTransformer

localstack_snapshot/snapshots/transformer.py

@@ -102,8 +102,14 @@ def transform(self, input_data: dict, *, ctx: TransformContext) -> dict:
             if k == "ResponseMetadata":
                 metadata = v
                 http_headers = metadata.get("HTTPHeaders")
+                if not isinstance(http_headers, dict):
+                    continue
+
                 # TODO "x-amz-bucket-region"
                 # TestS3.test_region_header_exists -> verifies bucket-region
+
+                # FIXME: proper value is `content-type` with no underscore in lowercase, but this will necessitate a
+                #  refresh of all snapshots
                 headers_to_collect = ["content_type"]
                 simplified_headers = {}
                 for h in headers_to_collect:

tests/test_transformer.py

@@ -5,6 +5,7 @@
 
 from localstack_snapshot.snapshots.transformer import (
     JsonStringTransformer,
+    ResponseMetaDataTransformer,
     SortingTransformer,
     TimestampTransformer,
     TransformContext,
@@ -405,3 +406,81 @@ def test_generic_timestamp_transformer(self):
         ctx = TransformContext()
         output = transformer.transform(input, ctx=ctx)
         assert output == expected
+
+
+class TestResponseMetaDataTransformer:
+    def test_with_headers(self):
+        input_data = {"ResponseMetadata": {"HTTPHeaders": {"header1": "value1"}}}
+
+        metadata_transformer = ResponseMetaDataTransformer()
+
+        expected_key_value = {"ResponseMetadata": {"HTTPHeaders": {}}}
+
+        copied = copy.deepcopy(input_data)
+        ctx = TransformContext()
+        assert metadata_transformer.transform(copied, ctx=ctx) == expected_key_value
+        assert ctx.serialized_replacements == []
+
+    def test_with_headers_and_status_code(self):
+        input_data = {
+            "ResponseMetadata": {"HTTPHeaders": {"header1": "value1"}, "HTTPStatusCode": 500}
+        }
+
+        metadata_transformer = ResponseMetaDataTransformer()
+
+        expected_key_value = {"ResponseMetadata": {"HTTPHeaders": {}, "HTTPStatusCode": 500}}
+
+        copied = copy.deepcopy(input_data)
+        ctx = TransformContext()
+        assert metadata_transformer.transform(copied, ctx=ctx) == expected_key_value
+        assert ctx.serialized_replacements == []
+
+    def test_with_status_code_only(self):
+        input_data = {"ResponseMetadata": {"HTTPStatusCode": 500, "RandomData": "random"}}
+
+        metadata_transformer = ResponseMetaDataTransformer()
+
+        expected_key_value = {"ResponseMetadata": {"HTTPStatusCode": 500, "RandomData": "random"}}
+
+        copied = copy.deepcopy(input_data)
+        ctx = TransformContext()
+        assert metadata_transformer.transform(copied, ctx=ctx) == expected_key_value
+        assert ctx.serialized_replacements == []
+
+    def test_with_empty_response_metadata(self):
+        input_data = {"ResponseMetadata": {"NotHeaders": "data"}}
+
+        metadata_transformer = ResponseMetaDataTransformer()
+
+        expected_key_value = {"ResponseMetadata": {"NotHeaders": "data"}}
+
+        copied = copy.deepcopy(input_data)
+        ctx = TransformContext()
+        assert metadata_transformer.transform(copied, ctx=ctx) == expected_key_value
+        assert ctx.serialized_replacements == []
+
+    def test_with_headers_wrong_type(self):
+        input_data = {"ResponseMetadata": {"HTTPHeaders": "data"}}
+
+        metadata_transformer = ResponseMetaDataTransformer()
+
+        expected_key_value = {"ResponseMetadata": {"HTTPHeaders": "data"}}
+
+        copied = copy.deepcopy(input_data)
+        ctx = TransformContext()
+        assert metadata_transformer.transform(copied, ctx=ctx) == expected_key_value
+        assert ctx.serialized_replacements == []
+
+    def test_headers_filtering(self):
+        input_data = {
+            "ResponseMetadata": {"HTTPHeaders": {"content_type": "value1", "header1": "value1"}}
+        }
+
+        metadata_transformer = ResponseMetaDataTransformer()
+
+        expected_key_value = {"ResponseMetadata": {"HTTPHeaders": {"content_type": "value1"}}}
+
+        copied = copy.deepcopy(input_data)
+        ctx = TransformContext()
+        assert metadata_transformer.transform(copied, ctx=ctx) == expected_key_value
+        assert ctx.serialized_replacements == []