Feat allow dict values in generate event cmd (#7938)

* feat(sam): allow generation of dict values in generate-event cmd

* feat: support dictionary values in generate-event command + add tests

* fix: run black, remove unused import + add test

* fix: pr remark + add test for nested query string params

---------

Co-authored-by: vicheey <181402101+vicheey@users.noreply.github.com>
Co-authored-by: seshubaws <116689586+seshubaws@users.noreply.github.com>

Author: 3 people

samcli/lib/generated_sample_events/event-mapping.json

@@ -241,6 +241,10 @@
           "type": "string",
           "default": "path/to/resource"
         },
+        "querystringparameters": {
+          "type": "dict",
+          "default": "{\"foo\": \"bar\"}"
+        },
         "account-id": {
           "default": "123456789012"
         },

samcli/lib/generated_sample_events/events.py

@@ -6,7 +6,7 @@
 import json
 import os
 import warnings
-from typing import Dict
+from typing import Any, Dict, Optional
 from urllib.parse import quote as url_quote
 
 with warnings.catch_warnings():
@@ -42,7 +42,7 @@ def __init__(self):
         with open(file_name) as f:
             self.event_mapping = json.load(f)
 
-    def transform(self, tags, values_to_sub):
+    def transform(self, tags: Dict[str, Any], values_to_sub: Dict[str, Any]) -> Dict[str, Any]:
         """
         transform (if needed) values_to_sub with given tags
 
@@ -68,21 +68,25 @@ def transform(self, tags, values_to_sub):
                     values_to_sub[child_tag] = child_val
         return values_to_sub
 
-    def transform_val(self, properties, val):
+    def transform_val(self, properties: Dict[str, Any], val: Optional[str]) -> str:
         """
         transform (if needed) given val with given properties
 
         Parameters
         ----------
         properties: dict
             set of properties to be used for transformation
-        val: string
+        val: string or None
             the value to undergo transformation
         Returns
         -------
         transformed
-            the transformed value
+            the transformed value (always a string for template rendering)
         """
+        # Handle None values by using the default
+        if val is None:
+            val = properties.get("default", "")
+
         transformed = val
 
         # encode if needed
@@ -95,6 +99,9 @@ def transform_val(self, properties, val):
         if hashing is not None:
             transformed = self.hash(hashing, transformed)
 
+        # Note: dict type is handled separately in generate_event() using placeholders
+        # to avoid JSON escaping issues during template rendering
+
         return transformed
 
     @staticmethod
@@ -166,6 +173,15 @@ def generate_event(self, service_name: str, event_type: str, values_to_sub: Dict
 
         # set variables for easy calling
         tags = self.event_mapping[service_name][event_type]["tags"]
+
+        # Store original dict-type values before transformation
+        dict_type_values = {}
+        for tag_name, tag_properties in tags.items():
+            if tag_properties.get("type") == "dict" and tag_name in values_to_sub:
+                dict_type_values[tag_name] = values_to_sub[tag_name]
+                # Replace with a placeholder for safe template rendering
+                values_to_sub[tag_name] = f"__DICT_PLACEHOLDER_{tag_name}__"
+
         values_to_sub = self.transform(tags, values_to_sub)
 
         # construct the path to the Events json file
@@ -180,4 +196,24 @@ def generate_event(self, service_name: str, event_type: str, values_to_sub: Dict
         data = json.dumps(data, indent=2)
 
         # return the substituted file (A string containing the rendered template.)
-        return renderer.render(data, values_to_sub)
+        rendered = renderer.render(data, values_to_sub)
+
+        # Parse the rendered result
+        rendered_json = json.loads(rendered)
+
+        # Replace placeholders with actual dict values
+        key_map = {k.lower(): k for k in rendered_json.keys()}
+        for tag_name, original_value in dict_type_values.items():
+            original_key = key_map.get(tag_name.lower())
+            if original_key:
+                placeholder = f"__DICT_PLACEHOLDER_{tag_name}__"
+                if isinstance(rendered_json[original_key], str) and rendered_json[original_key] == placeholder:
+                    try:
+                        # Parse the original JSON string value to a dict
+                        rendered_json[original_key] = json.loads(original_value)
+                    except (json.JSONDecodeError, TypeError):
+                        # If parsing fails, keep the original string value
+                        rendered_json[original_key] = original_value
+
+        # Return the final JSON with proper dictionary values
+        return json.dumps(rendered_json, indent=2)

samcli/lib/generated_sample_events/events/apigateway/AwsProxy.json

@@ -4,9 +4,7 @@
   "path": "/{{{path}}}",
   "httpMethod": "{{{method}}}",
   "isBase64Encoded": true,
-  "queryStringParameters": {
-    "foo": "bar"
-  },
+  "queryStringParameters": "{{{querystringparameters}}}",
   "multiValueQueryStringParameters": {
     "foo": [
       "bar"

tests/integration/local/generate_event/test_cli_integ.py

@@ -1,6 +1,7 @@
 from unittest import TestCase
-from subprocess import Popen
+from subprocess import Popen, PIPE
 import os
+import json
 
 from tests.testing_utils import get_sam_command
 
@@ -10,3 +11,78 @@ def test_generate_event_substitution(self):
         process = Popen([get_sam_command(), "local", "generate-event", "s3", "put"])
         process.communicate()
         self.assertEqual(process.returncode, 0)
+
+    def test_generate_event_with_dict_value(self):
+        """Test that dictionary values are properly handled in generate-event"""
+        process = Popen(
+            [
+                get_sam_command(),
+                "local",
+                "generate-event",
+                "apigateway",
+                "aws-proxy",
+                "--method",
+                "GET",
+                "--path",
+                "document",
+                "--body",
+                "",
+                "--querystringparameters",
+                '{"documentId": "1044", "versionId": "v_1"}',
+            ],
+            stdout=PIPE,
+            stderr=PIPE,
+        )
+        stdout, stderr = process.communicate()
+        self.assertEqual(process.returncode, 0)
+
+        # Parse the output JSON
+        result = json.loads(stdout.decode("utf-8"))
+
+        # Verify that queryStringParameters is a dict, not a string
+        self.assertIsInstance(result["queryStringParameters"], dict)
+        self.assertEqual(result["queryStringParameters"]["documentId"], "1044")
+        self.assertEqual(result["queryStringParameters"]["versionId"], "v_1")
+
+        # Verify other fields are still properly substituted
+        self.assertEqual(result["httpMethod"], "GET")
+        self.assertEqual(result["path"], "/document")
+
+    def test_generate_event_with_multiple_query_params(self):
+        """Test that multiple query string parameters are properly handled"""
+        process = Popen(
+            [
+                get_sam_command(),
+                "local",
+                "generate-event",
+                "apigateway",
+                "aws-proxy",
+                "--method",
+                "POST",
+                "--path",
+                "api/search",
+                "--body",
+                '{"query": "test"}',
+                "--querystringparameters",
+                '{"filter": "active", "sort": "desc", "limit": "10", "offset": "0"}',
+            ],
+            stdout=PIPE,
+            stderr=PIPE,
+        )
+        stdout, stderr = process.communicate()
+        self.assertEqual(process.returncode, 0)
+
+        # Parse the output JSON
+        result = json.loads(stdout.decode("utf-8"))
+
+        # Verify that queryStringParameters is a dict with all parameters
+        self.assertIsInstance(result["queryStringParameters"], dict)
+        self.assertEqual(len(result["queryStringParameters"]), 4)
+        self.assertEqual(result["queryStringParameters"]["filter"], "active")
+        self.assertEqual(result["queryStringParameters"]["sort"], "desc")
+        self.assertEqual(result["queryStringParameters"]["limit"], "10")
+        self.assertEqual(result["queryStringParameters"]["offset"], "0")
+
+        # Verify other fields
+        self.assertEqual(result["httpMethod"], "POST")
+        self.assertEqual(result["path"], "/api/search")

tests/unit/commands/local/generate_event/test_event_generation.py

@@ -1,3 +1,4 @@
+import json
 import os
 
 from unittest import TestCase
@@ -73,6 +74,78 @@ def test_transform(self):
         }
         self.assertEqual(result, expected)
 
+    def test_transform_val_dict_type_kept_as_string(self):
+        """Test that dict type values are kept as strings during transform"""
+        properties = {"type": "dict"}
+        val = '{"key1": "value1", "key2": "value2"}'
+        result = events.Events().transform_val(properties, val)
+        # Should remain a string for template rendering
+        self.assertEqual(result, val)
+        self.assertIsInstance(result, str)
+
+    def test_generate_event_with_dict_value(self):
+        """Test that generate_event properly handles dictionary values"""
+        events_lib = events.Events()
+        # Test with apigateway aws-proxy which has querystringparameters as dict type
+        # Pass as JSON string (as it would come from CLI)
+        result = events_lib.generate_event(
+            "apigateway",
+            "aws-proxy",
+            {
+                "method": "GET",
+                "path": "test",
+                "body": "",
+                "querystringparameters": '{"documentId": "1044", "versionId": "v_1"}',
+            },
+        )
+
+        result_json = json.loads(result)
+        # Verify that queryStringParameters is a dict, not a string
+        self.assertIsInstance(result_json["queryStringParameters"], dict)
+        self.assertEqual(result_json["queryStringParameters"]["documentId"], "1044")
+        self.assertEqual(result_json["queryStringParameters"]["versionId"], "v_1")
+
+    def test_generate_event_with_complex_dict_value(self):
+        """Test that generate_event handles complex dict values with special characters"""
+        events_lib = events.Events()
+        # Test with complex JSON including special characters
+        result = events_lib.generate_event(
+            "apigateway",
+            "aws-proxy",
+            {
+                "method": "POST",
+                "path": "api/search",
+                "body": "",
+                "querystringparameters": '{"filter": "status=active", "sort": "desc", "limit": "10"}',
+            },
+        )
+
+        result_json = json.loads(result)
+        # Verify complex dict is properly parsed
+        self.assertIsInstance(result_json["queryStringParameters"], dict)
+        self.assertEqual(result_json["queryStringParameters"]["filter"], "status=active")
+        self.assertEqual(result_json["queryStringParameters"]["sort"], "desc")
+        self.assertEqual(result_json["queryStringParameters"]["limit"], "10")
+
+    def test_generate_event_with_nested_dict_value(self):
+        """Test that generate_event preserves nested objects inside dict-type values"""
+        events_lib = events.Events()
+        result = events_lib.generate_event(
+            "apigateway",
+            "aws-proxy",
+            {
+                "method": "GET",
+                "path": "test",
+                "body": "",
+                "querystringparameters": '{"outer": {"inner": "x"}, "items": ["a", "b"]}',
+            },
+        )
+
+        result_json = json.loads(result)
+        self.assertIsInstance(result_json["queryStringParameters"], dict)
+        self.assertEqual(result_json["queryStringParameters"]["outer"]["inner"], "x")
+        self.assertEqual(result_json["queryStringParameters"]["items"], ["a", "b"])
+
 
 class TestServiceCommand(TestCase):
     def setUp(self):