feat: connectors (#241)

Author: hoffa

MANIFEST.in

@@ -4,6 +4,7 @@ include requirements/dev.txt
 recursive-include samtranslator/validator/sam_schema *.json
 include samtranslator/policy_templates_data/policy_templates.json
 include samtranslator/policy_templates_data/schema.json
+include samtranslator/model/connector_profiles/profiles.json
 include README.md
 include THIRD_PARTY_LICENSES
 

integration/combination/test_connectors.py

@@ -0,0 +1,146 @@
+from time import sleep
+from parameterized import parameterized
+from integration.conftest import clean_bucket
+from integration.helpers.base_test import S3_BUCKET_PREFIX, BaseTest
+from integration.helpers.resource import generate_suffix
+
+
+class TestConnectors(BaseTest):
+    @parameterized.expand(
+        [
+            ("combination/connector_function_to_function",),
+            ("combination/connector_restapi_to_function",),
+            ("combination/connector_httpapi_to_function",),
+            ("combination/connector_function_to_bucket_read",),
+            ("combination/connector_function_to_bucket_write",),
+            ("combination/connector_function_to_table_read",),
+            ("combination/connector_function_to_table_write",),
+            ("combination/connector_function_to_sfn_read",),
+            ("combination/connector_function_to_sfn_write",),
+            ("combination/connector_function_to_queue_write",),
+            ("combination/connector_function_to_queue_read",),
+            ("combination/connector_function_to_topic_write",),
+            ("combination/connector_function_to_eventbus_write",),
+            ("combination/connector_topic_to_queue_write",),
+            ("combination/connector_event_rule_to_sqs_write",),
+            ("combination/connector_event_rule_to_sns_write",),
+            ("combination/connector_event_rule_to_sfn_write",),
+            ("combination/connector_event_rule_to_eb_default_write",),
+            ("combination/connector_event_rule_to_eb_custom_write",),
+            ("combination/connector_event_rule_to_lambda_write",),
+            ("combination/connector_sqs_to_function",),
+            ("combination/connector_sns_to_function_write",),
+            ("combination/connector_table_to_function_read",),
+        ]
+    )
+    def test_connector_by_invoking_a_function(self, template_file_path):
+        self.skip_using_service_detector(template_file_path)
+        self.create_and_verify_stack(template_file_path)
+
+        lambda_function_name = self.get_physical_id_by_logical_id("TriggerFunction")
+        lambda_client = self.client_provider.lambda_client
+        s3_client = self.client_provider.s3_client
+
+        request_params = {
+            "FunctionName": lambda_function_name,
+            "InvocationType": "RequestResponse",
+            "Payload": "{}",
+        }
+        response = lambda_client.invoke(**request_params)
+        self.assertEqual(response.get("StatusCode"), 200)
+        self.assertEqual(response.get("FunctionError"), None)
+
+        # Some tests will create items in S3 Bucket, which result in stack DELETE_FAILED state
+        # manually empty the bucket to allow stacks to be deleted successfully.
+        bucket_name = self.get_physical_id_by_type("AWS::S3::Bucket")
+        if bucket_name:
+            clean_bucket(bucket_name, s3_client)
+
+    @parameterized.expand(
+        [
+            ("combination/connector_sfn_to_table_read",),
+            ("combination/connector_sfn_to_table_write",),
+            ("combination/connector_sfn_to_sqs_write",),
+            ("combination/connector_sfn_to_sns_write",),
+            ("combination/connector_sfn_to_function_write",),
+            ("combination/connector_sfn_to_bucket_write",),
+            ("combination/connector_sfn_to_bucket_read",),
+            ("combination/connector_sfn_to_sfn_async",),
+            ("combination/connector_sfn_to_eb_default_write",),
+            ("combination/connector_sfn_to_eb_custom_write",),
+        ]
+    )
+    def test_connector_by_sync_execute_an_state_machine(self, template_file_path):
+        self.skip_using_service_detector(template_file_path)
+        self.create_and_verify_stack(template_file_path)
+
+        state_machine_arn = self.get_physical_id_by_logical_id("TriggerStateMachine")
+        sfn_client = self.client_provider.sfn_client
+
+        response = sfn_client.start_sync_execution(
+            stateMachineArn=state_machine_arn,
+        )
+        # Without permission, it will be "FAILED"
+        self.assertEqual(response.get("status"), "SUCCEEDED")
+
+    @parameterized.expand(
+        [
+            ("combination/connector_sfn_to_sfn_sync",),
+        ]
+    )
+    def test_connector_by_async_execute_an_state_machine(self, template_file_path):
+        self.skip_using_service_detector(template_file_path)
+        self.create_and_verify_stack(template_file_path)
+
+        state_machine_arn = self.get_physical_id_by_logical_id("TriggerStateMachine")
+        sfn_client = self.client_provider.sfn_client
+
+        response = sfn_client.start_execution(
+            stateMachineArn=state_machine_arn,
+        )
+        execution_arn = response["executionArn"]
+
+        status = None
+        wait_tries = 5
+        while wait_tries > 0:
+            response = sfn_client.describe_execution(executionArn=execution_arn)
+            status = response["status"]
+            if status == "RUNNING":
+                wait_tries -= 1
+                sleep(5)
+                continue
+            else:
+                break
+
+        # Without permission, it will be "FAILED"
+        self.assertEqual(status, "SUCCEEDED")
+
+    @parameterized.expand(
+        [
+            ("combination/connector_bucket_to_function_write",),
+        ]
+    )
+    def test_connector_by_execute_a_s3_bucket(self, template_file_path):
+        self.skip_using_service_detector(template_file_path)
+        bucket_name = S3_BUCKET_PREFIX + "connector" + generate_suffix()
+        self.create_and_verify_stack(
+            template_file_path, [{"ParameterKey": "BucketName", "ParameterValue": bucket_name}]
+        )
+
+        lambda_function_name = self.get_physical_id_by_logical_id("TriggerFunction")
+        lambda_client = self.client_provider.lambda_client
+        s3_client = self.client_provider.s3_client
+
+        request_params = {
+            "FunctionName": lambda_function_name,
+            "InvocationType": "RequestResponse",
+            "Payload": "{}",
+        }
+        response = lambda_client.invoke(**request_params)
+        self.assertEqual(response.get("StatusCode"), 200)
+        self.assertEqual(response.get("FunctionError"), None)
+
+        # Some tests will create items in S3 Bucket, which result in stack DELETE_FAILED state
+        # manually empty the bucket to allow stacks to be deleted successfully.
+        bucket_name = self.get_physical_id_by_type("AWS::S3::Bucket")
+        clean_bucket(bucket_name, s3_client)

integration/conftest.py

@@ -35,7 +35,7 @@ def _get_all_buckets():
     return s3.buckets.all()
 
 
-def _clean_bucket(s3_bucket_name, s3_client):
+def clean_bucket(s3_bucket_name, s3_client):
     """
     Empties and deletes the bucket used for the tests
     """
@@ -60,7 +60,7 @@ def clean_all_integ_buckets():
     s3_client = ClientProvider().s3_client
     for bucket in buckets:
         if bucket.name.startswith(S3_BUCKET_PREFIX):
-            _clean_bucket(bucket.name, s3_client)
+            clean_bucket(bucket.name, s3_client)
 
 
 @pytest.fixture()

integration/helpers/base_test.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 import requests
@@ -12,7 +13,12 @@
 from integration.helpers.deployer.utils.retry import retry_with_exponential_backoff_and_jitter
 from integration.helpers.exception import StatusCodeError
 from integration.helpers.request_utils import RequestUtils
-from integration.helpers.resource import generate_suffix, create_bucket, verify_stack_resources
+from integration.helpers.resource import (
+    current_region_does_not_support,
+    detect_services,
+    generate_suffix,
+    verify_stack_resources,
+)
 from integration.helpers.s3_uploader import S3Uploader
 from integration.helpers.yaml_utils import dump_yaml, load_yaml
 from integration.helpers.resource import read_test_config_file
@@ -151,6 +157,24 @@ def create_and_verify_stack(self, file_path, parameters=None, s3_uploader=None):
         self.expected_resource_path = str(Path(self.expected_dir, folder, file_name + ".json"))
         self.verify_stack()
 
+    def skip_using_service_detector(self, file_path):
+        """
+        Skips the test if it cannot pass the test of
+        current_region_does_not_support() with detected services.
+        """
+        folder, file_name = file_path.split("/")
+
+        input_file_path = str(Path(self.template_dir, folder, file_name + ".yaml"))
+        expected_resource_path = str(Path(self.expected_dir, folder, file_name + ".json"))
+
+        template_dict = yaml_parse(open(input_file_path).read())
+        cfn_resource_types = {item["ResourceType"] for item in json.load(open(expected_resource_path))}
+
+        detected_services = detect_services(template_dict, cfn_resource_types)
+
+        if current_region_does_not_support(detected_services):
+            self.skipTest(f"Some/All of {detected_services} are not supported in this testing region")
+
     def update_stack(self, parameters=None, file_path=None):
         """
         Updates the Cloud Formation stack

integration/helpers/resource.py

@@ -2,7 +2,9 @@
 import re
 import random
 import string  # pylint: disable=deprecated-module
+from typing import Any, Callable, Dict, List, Set
 
+from integration.config.service_names import DYNAMO_DB, HTTP_API, S3_EVENTS, SQS, STATE_MACHINE_INLINE_DEFINITION
 from integration.helpers.yaml_utils import load_yaml
 
 try:
@@ -183,6 +185,58 @@ def current_region_does_not_support(services):
     return bool(set(services).intersection(set(region_exclude_services["regions"][region])))
 
 
+def _resource_using_inline_statemachine_definition(resource: Dict[str, Any]) -> bool:
+    resource_type = resource.get("Type")
+    properties = resource.get("Properties", {})
+    if resource_type == "AWS::StepFunctions::StateMachine" and properties.get("DefinitionString"):
+        return True
+    if resource_type == "AWS::Serverless::StateMachine" and properties.get("Definition"):
+        return True
+    return False
+
+
+def _resource_using_s3_events(resource: Dict[str, Any]) -> bool:
+    resource_type = resource.get("Type")
+    properties = resource.get("Properties", {})
+    return resource_type == "AWS::S3::Bucket" and properties.get("NotificationConfiguration")
+
+
+SERVICE_DETECTORS: Dict[str, Callable[[Dict[str, Any], Set[str]], bool]] = {
+    HTTP_API: lambda template_dict, cfn_resource_types: "AWS::ApiGatewayV2::Api" in cfn_resource_types,
+    SQS: lambda template_dict, cfn_resource_types: "AWS::SQS::Queue" in cfn_resource_types,
+    DYNAMO_DB: lambda template_dict, cfn_resource_types: "AWS::DynamoDB::Table" in cfn_resource_types,
+    STATE_MACHINE_INLINE_DEFINITION: lambda template_dict, cfn_resource_types: any(
+        _resource_using_inline_statemachine_definition(resource)
+        for resource in template_dict.get("Resources", {}).values()
+    ),
+    S3_EVENTS: lambda template_dict, cfn_resource_types: any(
+        _resource_using_s3_events(resource) for resource in template_dict.get("Resources", {}).values()
+    ),
+}
+
+
+def detect_services(template_dict: Dict[str, Any], cfn_resource_types: Set[str]):
+    """
+    Detect which services are used in the template.
+
+    TODO: Only used for connector integ testing for now.
+    this is not cannot detect all the services. Adding more if needed.
+
+    Parameters
+    ----------
+    template_dict: Dict[str, Any]
+        the template dict
+    cfn_resource_types : Set[str]
+        the transformed cfn resource types to be created
+
+    Returns
+    -------
+    List[str]
+        List of services in integration/config/service_names.py
+    """
+    return [service for service, detector in SERVICE_DETECTORS.items() if detector(template_dict, cfn_resource_types)]
+
+
 def current_region_not_included(services):
     """
     Opposite of current_region_does_not_support.

integration/resources/expected/combination/connector_bucket_to_function_write.json

@@ -0,0 +1,34 @@
+[
+  {
+    "LogicalResourceId": "TriggerFunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "TriggerFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "InvokedFunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "InvokedFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "TriggerBucket",
+    "ResourceType": "AWS::S3::Bucket"
+  },
+  {
+    "LogicalResourceId": "VerificationQueue",
+    "ResourceType": "AWS::SQS::Queue"
+  },
+  {
+    "LogicalResourceId": "MyConnectorWriteLambdaPermission",
+    "ResourceType": "AWS::Lambda::Permission"
+  },
+  {
+    "LogicalResourceId": "ConnectorNotTestedPolicy",
+    "ResourceType": "AWS::IAM::ManagedPolicy"
+  }
+]

integration/resources/expected/combination/connector_event_rule_to_eb_custom_write.json

@@ -0,0 +1,38 @@
+[
+  {
+    "LogicalResourceId": "ConnectorNotBeingTestedQueuePolicy",
+    "ResourceType": "AWS::SQS::QueuePolicy"
+  },
+  {
+    "LogicalResourceId": "CustomEventBus",
+    "ResourceType": "AWS::Events::EventBus"
+  },
+  {
+    "LogicalResourceId": "EBRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "EventRule",
+    "ResourceType": "AWS::Events::Rule"
+  },
+  {
+    "LogicalResourceId": "MyConnectorPolicy",
+    "ResourceType": "AWS::IAM::ManagedPolicy"
+  },
+  {
+    "LogicalResourceId": "TriggerFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "TriggerFunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "VerificationEventRule",
+    "ResourceType": "AWS::Events::Rule"
+  },
+  {
+    "LogicalResourceId": "VerificationQueue",
+    "ResourceType": "AWS::SQS::Queue"
+  }
+]