chore: sync foss changes (#2359)

Author: mingkun2020

integration/combination/test_custom_http_api_domains_test.py

@@ -0,0 +1,48 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import CUSTOM_DOMAIN
+from integration.helpers.base_internal_test import BaseInternalTest
+from integration.helpers.file_resources import FILE_TO_S3_URI_MAP
+from integration.helpers.resource import current_region_not_included
+
+
+@skipIf(
+    current_region_not_included([CUSTOM_DOMAIN]),
+    "CustomDomain is not supported in this testing region",
+)
+class TestCustomHttpApiDomains(BaseInternalTest):
+    def test_custom_http_api_domains_regional(self):
+        self.create_and_verify_stack("combination/http_api_with_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)
+
+        self.assertEqual("httpapi.sam-gamma-regional.com", result["DomainName"])
+
+        mtls_auth_config = result["MutualTlsAuthentication"]
+        self.assertEqual(FILE_TO_S3_URI_MAP["MTLSCert.pem"]["uri"], mtls_auth_config["TruststoreUri"])
+
+        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"])
+
+    def test_custom_http_api_domains_regional_ownership_verification(self):
+        self.create_and_verify_stack("combination/http_api_with_custom_domains_regional_ownership_verification")
+
+        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)
+
+        domain_name_configs = result["DomainNameConfigurations"]
+        self.assertEqual(1, len(domain_name_configs))
+        domain_name_config = domain_name_configs[0]
+
+        self.assertIsNotNone(domain_name_config.get("OwnershipVerificationCertificateArn"))

integration/combination/test_custom_rest_api_domains.py

@@ -0,0 +1,59 @@
+from unittest.case import skipIf
+
+from integration.config.service_names import CUSTOM_DOMAIN
+from integration.helpers.base_internal_test import BaseInternalTest
+from integration.helpers.file_resources import FILE_TO_S3_URI_MAP
+from integration.helpers.resource import current_region_not_included
+
+
+@skipIf(
+    current_region_not_included([CUSTOM_DOMAIN]),
+    "CustomDomain is not supported in this testing region",
+)
+class TestCustomRestApiDomains(BaseInternalTest):
+    def test_custom_rest_api_domains_edge(self):
+        self.create_and_verify_stack("combination/api_with_custom_domains_edge")
+        domain_name_list = self.get_stack_resources("AWS::ApiGateway::DomainName")
+        self.assertEqual(1, len(domain_name_list))
+
+        domain_name_id = self.get_physical_id_by_type("AWS::ApiGateway::DomainName")
+        api_gateway_client = self.client_provider.api_client
+        result = api_gateway_client.get_domain_name(domainName=domain_name_id)
+
+        self.assertEqual("sam-gamma-edge.com", result["domainName"])
+
+        end_point_config = result["endpointConfiguration"]
+        end_point_types = end_point_config["types"]
+        self.assertEqual(1, len(end_point_types))
+        self.assertEqual("EDGE", end_point_types[0])
+
+    def test_custom_rest_api_domains_regional(self):
+        self.create_and_verify_stack("combination/api_with_custom_domains_regional")
+
+        domain_name_list = self.get_stack_resources("AWS::ApiGateway::DomainName")
+        self.assertEqual(1, len(domain_name_list))
+
+        domain_name_id = self.get_physical_id_by_type("AWS::ApiGateway::DomainName")
+
+        api_gateway_client = self.client_provider.api_client
+        result = api_gateway_client.get_domain_name(domainName=domain_name_id)
+
+        self.assertEqual("sam-gamma-regional.com", result["domainName"])
+        self.assertEqual("TLS_1_2", result["securityPolicy"])
+
+        end_point_config = result["endpointConfiguration"]
+        end_point_types = end_point_config["types"]
+        self.assertEqual(1, len(end_point_types))
+        self.assertEqual("REGIONAL", end_point_types[0])
+
+        mtls_auth_config = result["mutualTlsAuthentication"]
+        self.assertEqual(FILE_TO_S3_URI_MAP["MTLSCert.pem"]["uri"], mtls_auth_config["truststoreUri"])
+
+    def test_custom_rest_api_domains_regional_ownership_verification(self):
+        self.create_and_verify_stack("combination/api_with_custom_domains_regional_ownership_verification")
+
+        domain_name_id = self.get_physical_id_by_type("AWS::ApiGateway::DomainName")
+        api_gateway_client = self.client_provider.api_client
+        result = api_gateway_client.get_domain_name(domainName=domain_name_id)
+
+        self.assertIsNotNone(result.get("ownershipVerificationCertificateArn"))

integration/combination/test_function_with_alias.py

@@ -18,8 +18,7 @@ def test_updating_version_by_changing_property_value(self):
 
         # Changing CodeUri should create a new version, and leave the existing version in tact
         self.set_template_resource_property("MyLambdaFunction", "CodeUri", self.file_to_s3_uri_map["code2.zip"]["uri"])
-        self.transform_template()
-        self.deploy_stack()
+        self.update_stack()
 
         version_ids = self.get_function_version_by_name(function_name)
         self.assertEqual(["1", "2"], version_ids)
@@ -43,8 +42,7 @@ def test_alias_deletion_must_retain_version(self):
         # Check that the DeletionPolicy on Lambda Version holds good
         # Remove alias, update stack, and verify the version still exists by calling Lambda APIs
         self.remove_template_resource_property("MyLambdaFunction", "AutoPublishAlias")
-        self.transform_template()
-        self.deploy_stack()
+        self.update_stack()
 
         # Make sure both Lambda version & alias resource does not exist in stack
         alias = self.get_stack_resources("AWS::Lambda::Alias")
@@ -71,13 +69,13 @@ def test_function_with_alias_with_intrinsics(self):
         # Let's change Key by updating the template parameter, but keep template same
         # This should create a new version and leave existing version intact
         parameters[1]["ParameterValue"] = "code2.zip"
-        # self.deploy_stack(parameters)
-        self.update_stack("combination/function_with_alias_intrinsics", parameters)
+
+        self.update_stack(parameters)
         version_ids = get_function_versions(function_name, self.client_provider.lambda_client)
+        self.assertEqual(["1", "2"], version_ids)
 
-        self.assertEqual(["1"], version_ids)
         alias = self.get_alias(function_name, alias_name)
-        self.assertEqual("1", alias["FunctionVersion"])
+        self.assertEqual("2", alias["FunctionVersion"])
 
     def test_alias_in_globals_with_overrides(self):
         # It is good enough if we can create a stack. Globals are pre-processed on the SAM template and don't
@@ -111,8 +109,7 @@ def test_alias_with_event_sources_get_correct_permissions(self):
 
         # Remove the alias, deploy the stack, and verify that *all* permission entities transfer to the function
         self.remove_template_resource_property("MyAwesomeFunction", "AutoPublishAlias")
-        self.transform_template()
-        self.deploy_stack()
+        self.update_stack()
 
         # Get the policies on both function & alias
         # Alias should have *no* policies

integration/combination/test_function_with_deployment_preference.py

@@ -41,8 +41,7 @@ def test_flip_from_disable_to_enable(self):
         pref["Enabled"] = "True"
         self.set_template_resource_property("MyLambdaFunction", "DeploymentPreference", pref)
 
-        self.transform_template()
-        self.deploy_stack(self.get_default_test_template_parameters())
+        self.update_stack(self.get_default_test_template_parameters())
 
         self._verify_no_deployment_then_update_and_verify_deployment(self.get_default_test_template_parameters())
 
@@ -77,8 +76,7 @@ def _verify_no_deployment_then_update_and_verify_deployment(self, parameters=Non
         self.set_template_resource_property(
             LAMBDA_FUNCTION_NAME, "CodeUri", self.file_to_s3_uri_map["code2.zip"]["uri"]
         )
-        self.transform_template()
-        self.deploy_stack(parameters)
+        self.update_stack(parameters)
 
         for deployment_group in deployment_groups:
             deployments = self._get_deployments(application_name, deployment_group)

integration/combination/test_function_with_policy_templates.py

@@ -0,0 +1,70 @@
+from integration.helpers.base_test import BaseTest
+from integration.helpers.common_api import get_policy_statements
+
+
+class TestFunctionWithPolicyTemplates(BaseTest):
+    def test_with_policy_templates(self):
+        self.create_and_verify_stack("combination/function_with_policy_templates")
+        role_name = self.get_physical_id_by_type("AWS::IAM::Role")
+
+        # There should be three policies created. Each policy has the name <resource-logicalid>Policy<index>
+
+        # Verify the contents of first policy
+        sqs_poller_policy = get_policy_statements(role_name, "MyFunctionRolePolicy0", self.client_provider.iam_client)
+        self.assertEqual(len(sqs_poller_policy), 1, "Only one statement must be in SQS Poller policy")
+
+        sqs_policy_statement = sqs_poller_policy[0]
+        self.assertTrue(type(sqs_policy_statement["Resource"]) != list)
+
+        queue_url = self.get_physical_id_by_type("AWS::SQS::Queue")
+        parts = queue_url.split("/")
+        expected_queue_name = parts[-1]
+        actual_queue_arn = sqs_policy_statement["Resource"]
+        self.assertTrue(
+            actual_queue_arn.endswith(expected_queue_name),
+            "Queue Arn " + actual_queue_arn + " must end with suffix " + expected_queue_name,
+        )
+
+        # Verify the contents of second policy
+        lambda_invoke_policy = get_policy_statements(
+            role_name, "MyFunctionRolePolicy1", self.client_provider.iam_client
+        )
+        self.assertEqual(len(lambda_invoke_policy), 1, "One policies statements should be present")
+
+        lambda_policy_statement = lambda_invoke_policy[0]
+        self.assertTrue(type(lambda_policy_statement["Resource"]) != list)
+
+        #  NOTE: The resource ARN has "*" suffix to allow for any Lambda function version as well
+        expected_function_suffix = "function:somename*"
+        actual_function_arn = lambda_policy_statement["Resource"]
+        self.assertTrue(
+            actual_function_arn.endswith(expected_function_suffix),
+            "Function ARN " + actual_function_arn + " must end with suffix " + expected_function_suffix,
+        )
+
+        # Verify the contents of third policy
+        cloud_watch_put_metric_policy = get_policy_statements(
+            role_name, "MyFunctionRolePolicy2", self.client_provider.iam_client
+        )
+        self.assertEqual(
+            len(cloud_watch_put_metric_policy), 1, "Only one statement must be in CloudWatchPutMetricPolicy"
+        )
+
+        cloud_watch_put_metric_statement = cloud_watch_put_metric_policy[0]
+        self.assertEqual(cloud_watch_put_metric_statement.get("Resource"), "*")
+
+    def test_all_policy_templates(self):
+        # template too large, upload it to s3
+        self.create_and_verify_stack("combination/all_policy_templates", s3_uploader=self.s3_uploader)
+
+        iam_roles = self.get_stack_resources("AWS::IAM::Role")
+        actual_num_polices = 0
+
+        for iam_role in iam_roles:
+            role_name = iam_role.get("PhysicalResourceId")
+            result = self.client_provider.iam_client.list_role_policies(RoleName=role_name)
+            policy_names = result.get("PolicyNames")
+            actual_num_polices += len(policy_names)
+
+        expected_num_polices = 69
+        self.assertEqual(actual_num_polices, expected_num_polices)

integration/combination/test_function_with_signing_profile.py

@@ -5,7 +5,7 @@
 from integration.config.service_names import CODE_SIGN
 
 
-class TestDependsOn(BaseTest):
+class TestFunctionWithSigningProfile(BaseTest):
     @skipIf(current_region_does_not_support([CODE_SIGN]), "CodeSign is not supported in this testing region")
     def test_function_with_signing_profile(self):
         self.create_and_verify_stack("combination/function_with_signing_profile")

integration/combination/test_http_api_with_auth.py

@@ -55,7 +55,7 @@ def test_function_with_user_pool_event(self):
         self.assertEqual(oauth_2_auth["IdentitySource"][0], "$request.querystring.param")
 
         # Test updating stack
-        self.update_stack("combination/http_api_with_auth_updated")
+        self.update_stack(file_path="combination/http_api_with_auth_updated")
 
         http_api_list_updated = self.get_stack_resources("AWS::ApiGatewayV2::Api")
         self.assertEqual(len(http_api_list_updated), 1)

integration/combination/test_http_api_with_disable_execute_api_endpoint.py

@@ -3,11 +3,11 @@
 from parameterized import parameterized
 
 from integration.helpers.base_test import BaseTest
-from integration.helpers.resource import current_region_does_not_support
+from integration.helpers.resource import current_region_not_included
 from integration.config.service_names import CUSTOM_DOMAIN
 
 
-@skipIf(current_region_does_not_support([CUSTOM_DOMAIN]), "CustomDomain is not supported in this testing region")
+@skipIf(current_region_not_included([CUSTOM_DOMAIN]), "CustomDomain is not supported in this testing region")
 class TestHttpApiWithDisableExecuteApiEndpoint(BaseTest):
     @parameterized.expand(
         [