test: Add service exclusion by region

INTEGRATION_TESTS.md

@@ -80,6 +80,12 @@ pytest --no-cov integration/single/test_basic_api.py::TestBasicApi::test_basic_a
 4. Write and add your python test code to the `integration` single or combination folder.
 5. Run it!
 
+## Skip tests for a specific service in a region
+
+1. Add the service you want to skip to the `integration/config/region_service_exclusion.yaml` under the region
+2. Add the @skipIf decorator to the test with the service name, take 'XRay' for example:
+```@skipIf(current_region_does_not_support('XRay'), 'XRay is not supported in this testing region')```
+
 ## Directory structure
 
 ### Helpers

integration/config/region_service_exclusion.yaml

@@ -0,0 +1,86 @@
+regions:
+  af-south-1:
+    - ServerlessRepo
+    - Cognito
+    - KMS
+    - CodeDeploy
+    - XRay
+    - IoT
+    - GatewayResponses
+    - HttpApi
+  ap-east-1:
+    - Cognito
+    - IoT
+    - ServerlessRepo
+    - HttpApi
+  ap-northeast-2:
+    - HttpApi
+  ap-northeast-3:
+    - Cognito
+    - IoT
+    - ServerlessRepo
+    - XRay
+    - CodeDeploy
+    - HttpApi
+  ap-south-1:
+    - HttpApi
+  ap-southeast-1:
+    - HttpApi
+  ca-central-1:
+    - Cognito
+    - IoT
+    - HttpApi
+  cn-north-1:
+    - ServerlessRepo
+    - Cognito
+    - KMS
+    - CodeDeploy
+    - XRay
+    - IoT
+    - GatewayResponses
+    - HttpApi
+  cn-northwest-1:
+    - ServerlessRepo
+    - Cognito
+    - KMS
+    - CodeDeploy
+    - XRay
+    - IoT
+    - GatewayResponses
+    - HttpApi
+  eu-north-1:
+    - ServerlessRepo
+    - Cognito
+    - IoT
+    - HttpApi
+    - Layers
+  eu-south-1:
+    - ServerlessRepo
+    - Cognito
+    - KMS
+    - CodeDeploy
+    - XRay
+    - IoT
+    - GatewayResponses
+    - HttpApi
+  eu-west-2:
+    - HttpApi
+  eu-west-3:
+    - Cognito
+    - IoT
+    - XRay
+    - HttpApi
+  me-south-1:
+    - ServerlessRepo
+    - Cognito
+    - IoT
+    - HttpApi
+  sa-east-1:
+    - IoT
+    - Cognito
+    - HttpApi
+  us-east-2:
+    - HttpApi
+  us-west-1:
+    - Cognito
+    - IoT

integration/helpers/base_test.py

@@ -3,6 +3,7 @@
 
 from integration.helpers.client_provider import ClientProvider
 from integration.helpers.resource import generate_suffix, create_bucket, verify_stack_resources
+from integration.helpers.yaml_utils import dump_yaml, load_yaml
 
 try:
     from pathlib import Path
@@ -294,7 +295,7 @@ def _fill_template(self, file_name):
             data = data.replace("${{{}}}".format(key), self.get_code_key_s3_uri(key))
         yaml_doc = yaml.load(data, Loader=yaml.FullLoader)
 
-        self._dump_yaml(updated_template_path, yaml_doc)
+        dump_yaml(updated_template_path, yaml_doc)
 
         self.sub_input_file_path = updated_template_path
 
@@ -311,12 +312,12 @@ def set_template_resource_property(self, resource_name, property_name, value):
         value
             value
         """
-        yaml_doc = self._load_yaml(self.sub_input_file_path)
+        yaml_doc = load_yaml(self.sub_input_file_path)
         yaml_doc["Resources"][resource_name]["Properties"][property_name] = value
-        self._dump_yaml(self.sub_input_file_path, yaml_doc)
+        dump_yaml(self.sub_input_file_path, yaml_doc)
 
     def get_template_resource_property(self, resource_name, property_name):
-        yaml_doc = self._load_yaml(self.sub_input_file_path)
+        yaml_doc = load_yaml(self.sub_input_file_path)
         return yaml_doc["Resources"][resource_name]["Properties"][property_name]
 
     def deploy_stack(self, parameters=None):
@@ -350,35 +351,3 @@ def verify_stack(self):
         error = verify_stack_resources(self.expected_resource_path, self.stack_resources)
         if error:
             self.fail(error)
-
-    def _load_yaml(self, file_path):
-        """
-        Loads a yaml file
-
-        Parameters
-        ----------
-        file_path : Path
-            File path
-
-        Returns
-        -------
-        Object
-            Yaml object
-        """
-        with open(file_path) as f:
-            data = f.read()
-        return yaml.load(data, Loader=yaml.FullLoader)
-
-    def _dump_yaml(self, file_path, yaml_doc):
-        """
-        Writes a yaml object to a file
-
-        Parameters
-        ----------
-        file_path : Path
-            File path
-        yaml_doc : Object
-            Yaml object
-        """
-        with open(file_path, "w") as f:
-            yaml.dump(yaml_doc, f)

integration/helpers/resource.py

@@ -3,6 +3,13 @@
 import random
 import string  # pylint: disable=deprecated-module
 
+from integration.helpers.yaml_utils import load_yaml
+
+try:
+    from pathlib import Path
+except ImportError:
+    from pathlib2 import Path
+
 import boto3
 from botocore.exceptions import ClientError, NoRegionError
 
@@ -114,3 +121,33 @@ def create_bucket(bucket_name, region):
         s3_client = boto3.client("s3", region_name=region)
         location = {"LocationConstraint": region}
         s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration=location)
+
+
+def current_region_does_not_support(services):
+    """
+    Decide if a test should be skipped in the current testing region with the specific resources
+
+    Parameters
+    ----------
+    services : List
+        the services to be tested in the current testing region
+
+    Returns
+    -------
+    Boolean
+        If skip return true otherwise false
+    """
+
+    session = boto3.session.Session()
+    region = session.region_name
+
+    tests_integ_dir = Path(__file__).resolve().parents[1]
+    config_dir = Path(tests_integ_dir, "config")
+    region_exclude_services_file = str(Path(config_dir, "region_service_exclusion.yaml"))
+    region_exclude_services = load_yaml(region_exclude_services_file)
+
+    if region not in region_exclude_services["regions"]:
+        return False
+
+    # check if any one of the services is in the excluded services for current testing region
+    return bool(set(services).intersection(set(region_exclude_services["regions"][region])))

integration/helpers/yaml_utils.py

@@ -0,0 +1,35 @@
+import yaml
+
+
+def load_yaml(file_path):
+    """
+    Loads a yaml file
+
+    Parameters
+    ----------
+    file_path : Path
+        File path
+
+    Returns
+    -------
+    Object
+        Yaml object
+    """
+    with open(file_path) as f:
+        data = f.read()
+    return yaml.load(data, Loader=yaml.FullLoader)
+
+
+def dump_yaml(file_path, yaml_doc):
+    """
+    Writes a yaml object to a file
+
+    Parameters
+    ----------
+    file_path : Path
+        File path
+    yaml_doc : Object
+        Yaml object
+    """
+    with open(file_path, "w") as f:
+        yaml.dump(yaml_doc, f)

integration/single/test_basic_application.py

@@ -1,11 +1,17 @@
+from unittest.case import skipIf
+
 from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
 
 
 class TestBasicApplication(BaseTest):
     """
     Basic AWS::Serverless::Application tests
     """
 
+    @skipIf(
+        current_region_does_not_support(["ServerlessRepo"]), "ServerlessRepo is not supported in this testing region"
+    )
     def test_basic_application_s3_location(self):
         """
         Creates an application with its properties defined as a template
@@ -19,6 +25,9 @@ def test_basic_application_s3_location(self):
         self.assertEqual(len(tables), 1)
         self.assertEqual(tables[0]["LogicalResourceId"], "MyTable")
 
+    @skipIf(
+        current_region_does_not_support(["ServerlessRepo"]), "ServerlessRepo is not supported in this testing region"
+    )
     def test_basic_application_sar_location(self):
         """
         Creates an application with a lamda function
@@ -31,6 +40,9 @@ def test_basic_application_sar_location(self):
         self.assertEqual(len(functions), 1)
         self.assertEqual(functions[0]["LogicalResourceId"], "helloworldpython")
 
+    @skipIf(
+        current_region_does_not_support(["ServerlessRepo"]), "ServerlessRepo is not supported in this testing region"
+    )
     def test_basic_application_sar_location_with_intrinsics(self):
         """
         Creates an application with a lambda function with intrinsics

integration/single/test_basic_function.py

@@ -1,4 +1,8 @@
+from unittest.case import skipIf
+
 import requests
+
+from integration.helpers.resource import current_region_does_not_support
 from parameterized import parameterized
 from integration.helpers.base_test import BaseTest
 
@@ -74,6 +78,7 @@ def test_basic_function_with_dlq(self, file_name, action):
         self.assertEqual(statements[0]["Resource"], dlq_arn)
         self.assertEqual(statements[0]["Effect"], "Allow")
 
+    @skipIf(current_region_does_not_support(["KMS"]), "KMS is not supported in this testing region")
     def test_basic_function_with_kms_key_arn(self):
         """
         Creates a basic lambda function with KMS key arn
@@ -148,6 +153,7 @@ def test_basic_function_event_destinations(self):
             "MaximumRetryAttempts value is not set or incorrect.",
         )
 
+    @skipIf(current_region_does_not_support(["XRay"]), "XRay is not supported in this testing region")
     def test_basic_function_with_tracing(self):
         """
         Creates a basic lambda function with tracing

integration/single/test_basic_http_api.py

@@ -1,11 +1,15 @@
+from unittest.case import skipIf
+
 from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
 
 
 class TestBasicHttpApi(BaseTest):
     """
     Basic AWS::Serverless::HttpApi tests
     """
 
+    @skipIf(current_region_does_not_support(["HttpApi"]), "HttpApi is not supported in this testing region")
     def test_basic_http_api(self):
         """
         Creates a HTTP API

integration/single/test_basic_layer_version.py

@@ -1,11 +1,15 @@
+from unittest.case import skipIf
+
 from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
 
 
 class TestBasicLayerVersion(BaseTest):
     """
     Basic AWS::Lambda::LayerVersion tests
     """
 
+    @skipIf(current_region_does_not_support(["Layers"]), "Layers is not supported in this testing region")
     def test_basic_layer_version(self):
         """
         Creates a basic lambda layer version
@@ -22,6 +26,7 @@ def test_basic_layer_version(self):
 
         self.assertFalse(layer_logical_id_1 == layer_logical_id_2)
 
+    @skipIf(current_region_does_not_support(["Layers"]), "Layers is not supported in this testing region")
     def test_basic_layer_with_parameters(self):
         """
         Creates a basic lambda layer version with parameters

integration/single/test_basic_state_machine.py

@@ -1,4 +1,7 @@
+from unittest.case import skipIf
+
 from integration.helpers.base_test import BaseTest
+from integration.helpers.resource import current_region_does_not_support
 
 
 class TestBasicLayerVersion(BaseTest):
@@ -12,6 +15,7 @@ def test_basic_state_machine_inline_definition(self):
         """
         self.create_and_verify_stack("basic_state_machine_inline_definition")
 
+    @skipIf(current_region_does_not_support(["XRay"]), "XRay is not supported in this testing region")
     def test_basic_state_machine_with_tags(self):
         """
         Creates a State Machine with tags