Merge main to develop

integration/combination/test_custom_http_api_domains_test.py

@@ -1,7 +1,11 @@
 from unittest.case import skipIf
 
 from integration.config.service_names import CUSTOM_DOMAIN
-from integration.helpers.base_internal_test import CUSTOM_DOMAIN_TOP_LEVEL, BaseInternalTest
+from integration.helpers.base_internal_test import (
+    CUSTOM_DOMAIN_TOP_LEVEL,
+    FEATURE_TOGGLE_CUSTOM_DOMAIN_TOP_LEVEL,
+    BaseInternalTest,
+)
 from integration.helpers.base_test import nonblocking
 from integration.helpers.resource import current_region_not_included
 
@@ -24,7 +28,9 @@ def test_custom_http_api_domains_regional(self):
         result = api_gateway_client.get_domain_name(DomainName=domain_name_id)
 
         if "FeatureToggle" in self.pipeline_prefix:
-            self.assertEqual("httpapi.ftl.sam-gamma-regional.com", result["DomainName"])
+            self.assertEqual(
+                f"httpapi-sam-gamma-regional.{FEATURE_TOGGLE_CUSTOM_DOMAIN_TOP_LEVEL}", result["DomainName"]
+            )
         else:
             self.assertEqual(f"httpapi-sam-gamma-regional.{CUSTOM_DOMAIN_TOP_LEVEL}", result["DomainName"])
 

integration/combination/test_custom_rest_api_domains.py

@@ -1,7 +1,11 @@
 from unittest.case import skipIf
 
 from integration.config.service_names import CUSTOM_DOMAIN
-from integration.helpers.base_internal_test import CUSTOM_DOMAIN_TOP_LEVEL, BaseInternalTest
+from integration.helpers.base_internal_test import (
+    CUSTOM_DOMAIN_TOP_LEVEL,
+    FEATURE_TOGGLE_CUSTOM_DOMAIN_TOP_LEVEL,
+    BaseInternalTest,
+)
 from integration.helpers.base_test import nonblocking
 from integration.helpers.resource import current_region_not_included
 
@@ -23,7 +27,7 @@ def test_custom_rest_api_domains_edge(self):
         result = api_gateway_client.get_domain_name(domainName=domain_name_id)
 
         if "FeatureToggle" in self.pipeline_prefix:
-            self.assertEqual("ftl.sam-gamma-edge.com", result["domainName"])
+            self.assertEqual(f"sam-gamma-edge.{FEATURE_TOGGLE_CUSTOM_DOMAIN_TOP_LEVEL}", result["domainName"])
         else:
             self.assertEqual(f"sam-gamma-edge.{CUSTOM_DOMAIN_TOP_LEVEL}", result["domainName"])
 
@@ -44,7 +48,7 @@ def test_custom_rest_api_domains_regional(self):
         result = api_gateway_client.get_domain_name(domainName=domain_name_id)
 
         if "FeatureToggle" in self.pipeline_prefix:
-            self.assertEqual("ftl.sam-gamma-regional.com", result["domainName"])
+            self.assertEqual(f"sam-gamma-regional.{FEATURE_TOGGLE_CUSTOM_DOMAIN_TOP_LEVEL}", result["domainName"])
         else:
             self.assertEqual(f"sam-gamma-regional.{CUSTOM_DOMAIN_TOP_LEVEL}", result["domainName"])
 

integration/combination/test_websocket_api_basic.py

@@ -0,0 +1,58 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import WEBSOCKET_API
+from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
+
+
+@skipIf(current_region_does_not_support([WEBSOCKET_API]), "WebSocketApi is not supported in this region")
+class TestWebSocketApiBasic(BaseTest):
+
+    def test_websocket_api_basic(self):
+        """
+        Creates a Basic WebSocket API
+        """
+        # Validates list of generated resources is same as expected
+        self.create_and_verify_stack("combination/websocket_api_basic")
+
+        stages = self.get_api_v2_stack_stages()
+
+        self.assertEqual(len(stages), 1)
+        self.assertEqual(stages[0]["StageName"], "default")
+
+        api_id = self.get_stack_outputs()["ApiId"]
+        api_2_client = self.client_provider.api_v2_client
+
+        routes = api_2_client.get_routes(ApiId=api_id)["Items"]
+        self.assertEqual(len(routes), 1)
+        self.assertEqual(routes[0]["RouteKey"], "$default")
+
+        integrations = api_2_client.get_integrations(ApiId=api_id)["Items"]
+        self.assertEqual(len(integrations), 1)
+
+    def test_websocket_api_basic_config(self):
+        """
+        Checks API config parameters that are strings
+        """
+        self.create_and_verify_stack("combination/websocket_api_basic_config")
+        websocket_api_list = self.get_stack_resources("AWS::ApiGatewayV2::Api")
+        self.assertEqual(len(websocket_api_list), 1)
+
+        websocket_resource = websocket_api_list[0]
+        websocket_api_id = websocket_resource["PhysicalResourceId"]
+        api_v2_client = self.client_provider.api_v2_client
+        api_configuration = api_v2_client.get_api(ApiId=websocket_api_id)
+        properties_to_check = {
+            "ApiKeySelectionExpression": "$request.header.x-api-key",
+            "Description": "Toy API",
+            "DisableExecuteApiEndpoint": False,
+            "Name": "MyApiName",
+            "ProtocolType": "WEBSOCKET",
+            "RouteSelectionExpression": "$request.body.action",
+        }
+        for key, value in properties_to_check.items():
+            self.assertEqual(api_configuration[key], value)
+        # assert custom tags are present
+        # CFN and SAM tags also exist, so we can't check equality of the full list
+        self.assertEqual(api_configuration["Tags"].get("t1"), "v1")
+        self.assertEqual(api_configuration["Tags"].get("t2"), "v2")

integration/combination/test_websocket_api_custom_domains.py

@@ -0,0 +1,47 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import CUSTOM_DOMAIN, WEBSOCKET_API
+from integration.helpers.base_internal_test import (
+    CUSTOM_DOMAIN_TOP_LEVEL,
+    FEATURE_TOGGLE_CUSTOM_DOMAIN_TOP_LEVEL,
+    BaseInternalTest,
+)
+from integration.helpers.base_test import nonblocking
+from integration.helpers.resource import current_region_does_not_support, current_region_not_included
+
+
+# Custom domain tests require pre-created infrastructure (hosted zones, certificates)
+# that only exists in us-east-1.
+@skipIf(
+    current_region_does_not_support([WEBSOCKET_API]) or current_region_not_included([CUSTOM_DOMAIN]),
+    "WebSocketApi is not supported or CustomDomain infrastructure is not available in this testing region",
+)
+@nonblocking
+class TestWebSocketApiCustomDomains(BaseInternalTest):
+    def test_websocket_custom_api_domains_regional(self):
+        self.create_and_verify_stack("combination/websocket_api_custom_domains_regional")
+
+        domain_name_list = self.get_stack_resources("AWS::ApiGatewayV2::DomainName")
+        self.assertEqual(1, len(domain_name_list))
+
+        domain_name_id = self.get_physical_id_by_type("AWS::ApiGatewayV2::DomainName")
+
+        api_gateway_client = self.client_provider.api_v2_client
+        result = api_gateway_client.get_domain_name(DomainName=domain_name_id)
+
+        if "FeatureToggle" in self.pipeline_prefix:
+            self.assertEqual(
+                f"websocket-sam-gamma-regional.{FEATURE_TOGGLE_CUSTOM_DOMAIN_TOP_LEVEL}", result["DomainName"]
+            )
+        else:
+            self.assertEqual(f"websocket-sam-gamma-regional.{CUSTOM_DOMAIN_TOP_LEVEL}", result["DomainName"])
+
+        domain_name_configs = result["DomainNameConfigurations"]
+        self.assertEqual(1, len(domain_name_configs))
+        domain_name_config = domain_name_configs[0]
+
+        self.assertEqual("REGIONAL", domain_name_config["EndpointType"])
+        self.assertEqual("TLS_1_2", domain_name_config["SecurityPolicy"])
+
+        domain_name_configs = result["DomainNameConfigurations"]
+        self.assertEqual(1, len(domain_name_configs))

integration/combination/test_websocket_api_multiple.py

@@ -0,0 +1,15 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import WEBSOCKET_API
+from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
+
+
+@skipIf(current_region_does_not_support([WEBSOCKET_API]), "WebSocketApi is not supported in this region")
+class TestWebSocketApiMultiple(BaseTest):
+
+    def test_websocket_multi_api(self):
+        self.create_and_verify_stack("combination/websocket_api_multiple_api")
+
+    def test_websocket_multi_route(self):
+        self.create_and_verify_stack("combination/websocket_api_multiple_routes")

integration/combination/test_websocket_api_route_config.py

@@ -0,0 +1,35 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import WEBSOCKET_API
+from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
+
+
+@skipIf(current_region_does_not_support([WEBSOCKET_API]), "WebSocketApi is not supported in this region")
+class TestWebSocketApiRouteConfig(BaseTest):
+
+    def test_websocket_api_route_config(self):
+        """
+        Checks API config parameters that are strings
+        """
+        self.create_and_verify_stack("combination/websocket_api_route_config")
+        websocket_route_list = self.get_stack_resources("AWS::ApiGatewayV2::Route")
+        self.assertEqual(len(websocket_route_list), 1)
+        websocket_route_integ_list = self.get_stack_resources("AWS::ApiGatewayV2::Integration")
+        self.assertEqual(len(websocket_route_integ_list), 1)
+        websocket_route_perm_list = self.get_stack_resources("AWS::Lambda::Permission")
+        self.assertEqual(len(websocket_route_perm_list), 1)
+        api_id = self.get_stack_outputs()["ApiId"]
+        api_2_client = self.client_provider.api_v2_client
+        routes = api_2_client.get_routes(ApiId=api_id)["Items"]
+        self.assertEqual(len(routes), 1)
+        route = routes[0]
+        self.assertEqual(route["RouteKey"], "$connect")
+        self.assertEqual(route["ModelSelectionExpression"], "$request.body.modelType")
+        self.assertEqual(route["OperationName"], "connect")
+        self.assertIsNotNone(route["RequestParameters"].get("route.request.querystring.p1"))
+        self.assertEqual(route["RouteResponseSelectionExpression"], "$default")
+        integrations = api_2_client.get_integrations(ApiId=api_id)
+        self.assertEqual(len(integrations["Items"]), 1)
+        integration = integrations["Items"][0]
+        self.assertEqual(integration["TimeoutInMillis"], 15000)

integration/combination/test_websocket_api_route_settings.py

@@ -0,0 +1,20 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import WEBSOCKET_API
+from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
+
+
+@skipIf(current_region_does_not_support([WEBSOCKET_API]), "WebSocketApi is not supported in this region")
+class TestWebSocketApiRouteSettings(BaseTest):
+    def test_websocket_api_route_settings(self):
+        """
+        Verifies that RouteSettings with specific route keys deploys successfully.
+        This test ensures Stage has DependsOn for routes.
+        """
+        self.create_and_verify_stack("combination/websocket_api_route_settings")
+
+        stages = self.get_api_v2_stack_stages()
+        self.assertEqual(len(stages), 1)
+        stage = stages[0]
+        self.assertIsNotNone(stage.get("RouteSettings"))

integration/combination/test_websocket_api_stage_config.py

@@ -0,0 +1,22 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import WEBSOCKET_API
+from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
+
+
+@skipIf(current_region_does_not_support([WEBSOCKET_API]), "WebSocketApi is not supported in this region")
+class TestWebSocketApiStageConfig(BaseTest):
+
+    def test_websocket_api_stage_config(self):
+        """
+        Checks API config parameters that are strings
+        """
+        self.create_and_verify_stack("combination/websocket_api_stage_config")
+        stages = self.get_api_v2_stack_stages()
+        self.assertEqual(len(stages), 1)
+        stage = stages[0]
+        self.assertEqual(stage["StageName"], "Prod")
+        self.assertEqual(stage["StageVariables"], {"var1": "val1", "var2": "val2"})
+        self.assertIsNotNone(stage["Tags"].get("t1"))
+        self.assertIsNotNone(stage["Tags"].get("t2"))

integration/combination/test_websocket_api_with_auth.py

@@ -0,0 +1,76 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import WEBSOCKET_API
+from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
+
+
+@skipIf(current_region_does_not_support([WEBSOCKET_API]), "WebSocketApi is not supported in this region")
+class TestWebSocketApiWithAuth(BaseTest):
+
+    def test_websocket_api_with_iam_auth(self):
+        """
+        Creates a WebSocket API with an IAM authorizer
+        """
+        self.create_and_verify_stack("combination/websocket_api_with_iam_auth")
+
+        websocket_api_list = self.get_stack_resources("AWS::ApiGatewayV2::Api")
+        self.assertEqual(len(websocket_api_list), 1)
+
+        stages = self.get_api_v2_stack_stages()
+
+        self.assertEqual(len(stages), 1)
+        self.assertEqual(stages[0]["StageName"], "default")
+
+        websocket_resource = websocket_api_list[0]
+        websocket_api_id = websocket_resource["PhysicalResourceId"]
+        api_v2_client = self.client_provider.api_v2_client
+        routes_list = api_v2_client.get_routes(ApiId=websocket_api_id)["Items"]
+        route = routes_list[0]
+        self.assertEqual(route["AuthorizationType"], "AWS_IAM")
+
+    def test_none_auth(self):
+        self.create_and_verify_stack("combination/websocket_api_with_none_auth")
+
+        websocket_api_list = self.get_stack_resources("AWS::ApiGatewayV2::Api")
+        self.assertEqual(len(websocket_api_list), 1)
+
+        websocket_resource = websocket_api_list[0]
+        websocket_api_id = websocket_resource["PhysicalResourceId"]
+        api_v2_client = self.client_provider.api_v2_client
+        routes_list = api_v2_client.get_routes(ApiId=websocket_api_id)["Items"]
+        route = routes_list[0]
+        self.assertEqual(route["AuthorizationType"], "NONE")
+
+    def test_websocket_api_with_lambda_auth_config(self):
+        """
+        Creates a WebSocket API with a Lambda authorizer
+        """
+        self.create_and_verify_stack("combination/websocket_api_with_lambda_auth")
+
+        websocket_api_list = self.get_stack_resources("AWS::ApiGatewayV2::Api")
+        self.assertEqual(len(websocket_api_list), 1)
+
+        websocket_resource = websocket_api_list[0]
+        websocket_api_id = websocket_resource["PhysicalResourceId"]
+        api_v2_client = self.client_provider.api_v2_client
+
+        route_list = api_v2_client.get_routes(ApiId=websocket_api_id)["Items"]
+        self.assertEqual(len(route_list), 1)
+        route = route_list[0]
+        self.assertEqual(route["AuthorizationType"], "CUSTOM")
+        self.assertIsNotNone(route["AuthorizerId"])
+
+        authorizer_list = api_v2_client.get_authorizers(ApiId=websocket_api_id)["Items"]
+        self.assertEqual(len(authorizer_list), 1)
+        lambda_auth = authorizer_list[0]
+        # Not sure this is returning properly either
+        self.assertEqual(lambda_auth["AuthorizerType"], "REQUEST")
+        # Verify authorizer URI contains expected components
+        authorizer_uri = lambda_auth["AuthorizerUri"]
+        self.assertIn("lambda:path/2015-03-31/functions", authorizer_uri)
+        self.assertIn("MyAuthFn", authorizer_uri)
+        self.assertIn("/invocations", authorizer_uri)
+
+        # Same authorizer coming from the route and from the authorizers
+        self.assertEqual(route["AuthorizerId"], lambda_auth["AuthorizerId"])

integration/combination/test_websocket_api_with_disable_execute_api_endpoint.py

@@ -0,0 +1,27 @@
+from unittest.case import skipIf
+
+from parameterized import parameterized
+
+from integration.config.service_names import WEBSOCKET_API
+from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
+
+
+@skipIf(current_region_does_not_support([WEBSOCKET_API]), "WebSocketApi is not supported in this region")
+class TestWebSocketApiWithDisableExecuteApiEndpoint(BaseTest):
+    @parameterized.expand([True, False])
+    def test_disable_execute_api_endpoint_true(self, is_disable):
+        parameters = [
+            {
+                "ParameterKey": "DisableExecuteApiEndpointValue",
+                "ParameterValue": "true" if is_disable else "false",
+                "UsePreviousValue": False,
+                "ResolvedValue": "string",
+            }
+        ]
+
+        self.create_and_verify_stack("combination/websocket_api_with_disable_execute_api_endpoint", parameters)
+        api_2_client = self.client_provider.api_v2_client
+        api_id = self.get_stack_outputs()["ApiId"]
+        api_result = api_2_client.get_api(ApiId=api_id)
+        self.assertEqual(api_result["DisableExecuteApiEndpoint"], is_disable)

integration/combination/test_websocket_api_with_disable_schema_validation.py

@@ -0,0 +1,27 @@
+from unittest.case import skipIf
+
+from parameterized import parameterized
+
+from integration.config.service_names import WEBSOCKET_API
+from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
+
+
+@skipIf(current_region_does_not_support([WEBSOCKET_API]), "WebSocketApi is not supported in this region")
+class TestWebSocketApiDisableSchemaValidation(BaseTest):
+    @parameterized.expand([True, False])
+    def test_disable_schema_validation(self, is_disable):
+        parameters = [
+            {
+                "ParameterKey": "DisableSchemaValidationValue",
+                "ParameterValue": "true" if is_disable else "false",
+                "UsePreviousValue": False,
+                "ResolvedValue": "string",
+            }
+        ]
+        self.create_and_verify_stack("combination/websocket_api_with_disable_schema_validation", parameters)
+        websocket_api_list = self.get_stack_resources("AWS::ApiGatewayV2::Api")
+        self.assertEqual(len(websocket_api_list), 1)
+        # API Gateway SDK for some reason doesn't return the disableSchemaValidation
+        # property when getting the API, so we just check that there were no errors
+        # during the creation, but we can't check that it was properly applied