localstack · bentsku · Nov 29, 2024 · Nov 29, 2024 · Nov 29, 2024 · Nov 29, 2024
diff --git a/localstack-core/localstack/services/apigateway/next_gen/execute_api/integrations/mock.py b/localstack-core/localstack/services/apigateway/next_gen/execute_api/integrations/mock.py
@@ -1,5 +1,6 @@
 import json
 import logging
+import re
 from json import JSONDecodeError
 
 from werkzeug.datastructures import Headers
@@ -39,20 +40,73 @@ def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:
 
         return EndpointResponse(status_code=status_code, body=b"", headers=Headers())
 
-    @staticmethod
-    def get_status_code(integration_req: IntegrationRequest) -> int | None:
+    def get_status_code(self, integration_req: IntegrationRequest) -> int | None:
         try:
-            body = json.loads(to_str(integration_req["body"]))
+            body = json.loads(integration_req["body"])
         except JSONDecodeError as e:
             LOG.debug(
-                "Exception while parsing integration request body: %s",
+                "Exception while JSON parsing integration request body: %s"
+                "Falling back to custom parser",
                 e,
                 exc_info=LOG.isEnabledFor(logging.DEBUG),
             )
-            return
+            body = self.parse_invalid_json(to_str(integration_req["body"]))
 
         status_code = body.get("statusCode")
         if not isinstance(status_code, int):
             return
 
         return status_code
+
+    def parse_invalid_json(self, body: str) -> dict:
+        """This is a quick fix to unblock cdk users setting cors policy for rest apis.
+        CDK creates a MOCK OPTIONS route with in valid json. `{statusCode: 200}`
+        Aws probably has a custom token parser. We can implement one
+        at some point if we have user requests for it"""
+        try:
+            statuscode = ""
+            matched = re.match(r"^\s*{(.+)}\s*$", body).group(1)
+            splits = [m.strip() for m in matched.split(",")]
+            # TODO this is not right, but nested object would otherwise break the parsing
+            kvs = [s.split(":", maxsplit=1) for s in splits]
+            for kv in kvs:
+                assert len(kv) == 2
+                k, v = kv
+                k = k.strip()
+                v = v.strip()
+
+                assert k
+                assert v
+
+                if k == "statusCode":
+                    statuscode = int(v)
+                    continue
+
+                if (first_char := k[0]) in "[{":
+                    raise Exception
+                if first_char in "'\"":
+                    assert len(k) > 2
+                    assert k[-1] == first_char
+                    k = k[1:-1]
+
+                if (v_first_char := v[0]) in "[{'\"":
+                    assert len(v) > 2
+                    if v_first_char == "{":
+                        # TODO reparse objects
+                        assert v[-1] == "}"
+                    elif v_first_char == "[":
+                        # TODO validate arrays
+                        assert v[-1] == "]"
+                    else:
+                        assert v[-1] == v_first_char
+                        v = v[1:-1]
+
+                if k == "statusCode":
+                    statuscode = int(v)
+
+            return {"statusCode": statuscode}
+        except Exception as e:
+            LOG.debug(
+                "Error Parsing an invalid json, %s", e, exc_info=LOG.isEnabledFor(logging.DEBUG)
+            )
+            return {"statusCode": ""}
diff --git a/tests/aws/services/apigateway/test_apigateway_integrations.py b/tests/aws/services/apigateway/test_apigateway_integrations.py
@@ -548,10 +548,11 @@ def test_put_integration_validation(
 
 @markers.aws.validated
 @pytest.mark.skipif(
-    condition=not is_next_gen_api(),
+    condition=not is_next_gen_api() and not is_aws_cloud(),
     reason="Behavior is properly implemented in Legacy, it returns the MOCK response",
 )
-def test_integration_mock_with_path_param(create_rest_apigw, aws_client):
+def test_integration_mock_with_path_param(create_rest_apigw, aws_client, snapshot):
+    snapshot.add_transformer(snapshot.transform.key_value("cacheNamespace"))
     api_id, _, root = create_rest_apigw(
         name=f"test-api-{short_uid()}",
         description="this is my api",
@@ -580,7 +581,7 @@ def test_integration_mock_with_path_param(create_rest_apigw, aws_client):
 
     # you don't have to pass URI for Mock integration as it's not used anyway
     # when exporting an API in AWS, apparently you can get integration path parameters even if not used
-    aws_client.apigateway.put_integration(
+    integration = aws_client.apigateway.put_integration(
         restApiId=api_id,
         resourceId=resource_id,
         httpMethod="GET",
@@ -589,8 +590,11 @@ def test_integration_mock_with_path_param(create_rest_apigw, aws_client):
         requestParameters={
             "integration.request.path.integrationPath": "method.request.path.testPath",
         },
-        requestTemplates={"application/json": '{"statusCode": 200}'},
+        # This template was modified to validate a cdk issue where it creates this template part
+        # of some L2 construct for CORS handling. This isn't valid JSON but accepted by aws.
+        requestTemplates={"application/json": "{statusCode: 200}"},
     )
+    snapshot.match("integration", integration)
 
     aws_client.apigateway.put_integration_response(
         restApiId=api_id,

diff --git a/tests/aws/services/apigateway/test_apigateway_integrations.snapshot.json b/tests/aws/services/apigateway/test_apigateway_integrations.snapshot.json
@@ -1056,5 +1056,27 @@
         "response": "this is the else clause"
       }
     }
+  },
+  "tests/aws/services/apigateway/test_apigateway_integrations.py::test_integration_mock_with_path_param": {
+    "recorded-date": "29-11-2024, 19:27:54",
+    "recorded-content": {
+      "integration": {
+        "cacheKeyParameters": [],
+        "cacheNamespace": "<cache-namespace:1>",
+        "passthroughBehavior": "WHEN_NO_MATCH",
+        "requestParameters": {
+          "integration.request.path.integrationPath": "method.request.path.testPath"
+        },
+        "requestTemplates": {
+          "application/json": "{statusCode: 200}"
+        },
+        "timeoutInMillis": 29000,
+        "type": "MOCK",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 201
+        }
+      }
+    }
   }
 }
diff --git a/tests/aws/services/apigateway/test_apigateway_integrations.validation.json b/tests/aws/services/apigateway/test_apigateway_integrations.validation.json
@@ -15,7 +15,7 @@
     "last_validated_date": "2024-04-15T23:07:07+00:00"
   },
   "tests/aws/services/apigateway/test_apigateway_integrations.py::test_integration_mock_with_path_param": {
-    "last_validated_date": "2024-11-05T12:55:51+00:00"
+    "last_validated_date": "2024-11-29T19:27:54+00:00"
   },
   "tests/aws/services/apigateway/test_apigateway_integrations.py::test_integration_mock_with_request_overrides_in_response_template": {
     "last_validated_date": "2024-11-06T23:09:04+00:00"

diff --git a/tests/unit/services/apigateway/test_mock_integration.py b/tests/unit/services/apigateway/test_mock_integration.py
@@ -51,3 +51,31 @@ def test_mock_integration(self, create_default_context):
         with pytest.raises(InternalServerError) as exc_info:
             mock_integration.invoke(ctx)
         assert exc_info.match("Internal server error")
+
+    def test_custom_parser(self, create_default_context):
+        mock_integration = RestApiMockIntegration()
+
+        valid_templates = [
+            "{statusCode: 200,super{ f}oo: [ba r]}",
+            "{statusCode: 200, \"value\": 'goog'}",
+            "{statusCode: 200, foo}: [ba r]}",
+            "{statusCode: 200, foo'}: [ba r]}",
+            "{statusCode: 200, }foo: [ba r]}",
+        ]
+        invalid_templates = [
+            "statusCode: 200",
+            "{statusCode: 200, {foo: [ba r]}",
+            # This test fails as we do not support nested objects
+            # "{statusCode: 200, what:{}foo: [ba r]}}"
+        ]
+
+        for valid_template in valid_templates:
+            ctx = create_default_context(body=valid_template)
+            response = mock_integration.invoke(ctx)
+            assert response["status_code"] == 200
+
+        for invalid_template in invalid_templates:
+            ctx = create_default_context(body=invalid_template)
+            with pytest.raises(InternalServerError) as exc_info:
+                mock_integration.invoke(ctx)
+            assert exc_info.match("Internal server error")