From 206110d14b9b1aa4ea458d1623f6d70b36802ca3 Mon Sep 17 00:00:00 2001
From: Benjamin Simon <benjh.simon@gmail.com>
Date: Fri, 20 Jun 2025 01:47:15 +0200
Subject: [PATCH] fix ResponseMetaDataTransformer

---
 localstack_snapshot/snapshots/transformer.py |  6 ++
 tests/test_transformer.py                    | 79 ++++++++++++++++++++
 2 files changed, 85 insertions(+)

diff --git a/localstack_snapshot/snapshots/transformer.py b/localstack_snapshot/snapshots/transformer.py
index f0a6384..63238a4 100644
--- a/localstack_snapshot/snapshots/transformer.py
+++ b/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:
diff --git a/tests/test_transformer.py b/tests/test_transformer.py
index 98a50d4..0468f91 100644
--- a/tests/test_transformer.py
+++ b/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 == []