test: Add simple integration tests

Makefile

@@ -6,13 +6,16 @@ init:
 	pip install -e '.[dev]'
 
 test:
-	pytest --cov samtranslator --cov-report term-missing --cov-fail-under 95 tests
+	pytest --cov samtranslator --cov-report term-missing --cov-fail-under 95 tests/*
+
+test-integ:
+	pytest --no-cov tests_integ/*
 
 black:
-	black setup.py samtranslator/* tests/* bin/*
+	black setup.py samtranslator/* tests/* tests_integ/* bin/*
 
 black-check:
-	black --check setup.py samtranslator/* tests/* bin/*
+	black --check setup.py samtranslator/* tests/* tests_integ/* bin/*
 
 # Command to run everytime you make changes to verify everything works
 dev: test
@@ -27,6 +30,7 @@ Usage: $ make [TARGETS]
 TARGETS
 	init        Initialize and install the requirements and dev-requirements for this project.
 	test        Run the Unit tests.
+	test-integ  Run the Integration tests.
 	dev         Run all development tests after a change.
 	pr          Perform all checks before submitting a Pull Request.
 

tests_integ/helpers/base_test.py

@@ -0,0 +1,108 @@
+import os
+from pathlib import Path
+from unittest.case import TestCase
+
+import boto3
+import pytest
+import yaml
+from samcli.lib.deploy.deployer import Deployer
+from tests_integ.helpers.helpers import transform_template, verify_stack_resources, generate_suffix, create_bucket
+
+STACK_NAME_PREFIX = "sam-integ-stack-"
+S3_BUCKET_PREFIX = "sam-integ-bucket-"
+CODE_KEY_TO_FILE_MAP = {"codeuri": "code.zip", "contenturi": "layer1.zip", "definitionuri": "swagger1.json"}
+
+
+class BaseTest(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        BaseTest.tests_integ_dir = Path(__file__).resolve().parents[1]
+        BaseTest.resources_dir = Path(BaseTest.tests_integ_dir, "resources")
+        BaseTest.template_dir = Path(BaseTest.resources_dir, "templates", "single")
+        BaseTest.output_dir = BaseTest.tests_integ_dir
+        BaseTest.expected_dir = Path(BaseTest.resources_dir, "expected", "single")
+        code_dir = Path(BaseTest.resources_dir, "code")
+
+        BaseTest.s3_bucket_name = S3_BUCKET_PREFIX + generate_suffix()
+        session = boto3.session.Session()
+        my_region = session.region_name
+        create_bucket(BaseTest.s3_bucket_name, region=my_region)
+
+        s3_client = boto3.client("s3")
+        BaseTest.code_key_to_url = {}
+
+        for key, file_name in CODE_KEY_TO_FILE_MAP.items():
+            code_path = str(Path(code_dir, file_name))
+            s3_client.upload_file(code_path, BaseTest.s3_bucket_name, file_name)
+            code_url = f"s3://{BaseTest.s3_bucket_name}/{file_name}"
+            BaseTest.code_key_to_url[key] = code_url
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        BaseTest._clean_bucket()
+
+    @classmethod
+    def _clean_bucket(cls):
+        s3_client = boto3.client("s3")
+        response = s3_client.list_objects(Bucket=BaseTest.s3_bucket_name)
+        for content in response["Contents"]:
+            s3_client.delete_object(Key=content["Key"], Bucket=BaseTest.s3_bucket_name)
+        s3_client.delete_bucket(Bucket=BaseTest.s3_bucket_name)
+
+    def setUp(self):
+        self.cloudformation_client = boto3.client("cloudformation")
+        self.deployer = Deployer(self.cloudformation_client, changeset_prefix="sam-integ-")
+
+    def create_and_verify_stack(self, file_name):
+        input_file_path = str(Path(BaseTest.template_dir, file_name + ".yaml"))
+        self.output_file_path = str(Path(BaseTest.output_dir, "cfn_" + file_name + ".yaml"))
+        expected_resource_path = str(Path(BaseTest.expected_dir, file_name + ".json"))
+        self.stack_name = STACK_NAME_PREFIX + file_name.replace("_", "-") + generate_suffix()
+
+        self.sub_input_file_path = self._update_template(input_file_path, file_name)
+        transform_template(self.sub_input_file_path, self.output_file_path)
+        self._deploy_stack()
+        self._verify_stack(expected_resource_path)
+
+    def tearDown(self):
+        self.cloudformation_client.delete_stack(StackName=self.stack_name)
+        if os.path.exists(self.output_file_path):
+            os.remove(self.output_file_path)
+        if os.path.exists(self.sub_input_file_path):
+            os.remove(self.sub_input_file_path)
+
+    def _update_template(self, input_file_path, file_name):
+        updated_template_path = str(Path(BaseTest.output_dir, "sub_" + file_name + ".yaml"))
+        with open(input_file_path, "r") as f:
+            data = f.read()
+        for key, s3_url in BaseTest.code_key_to_url.items():
+            data = data.replace(f"${{{key}}}", s3_url)
+        yaml_doc = yaml.load(data, Loader=yaml.FullLoader)
+
+        with open(updated_template_path, "w") as f:
+            yaml.dump(yaml_doc, f)
+
+        return updated_template_path
+
+    def _deploy_stack(self):
+        with open(self.output_file_path, "r") as cfn_file:
+            result, changeset_type = self.deployer.create_and_wait_for_changeset(
+                stack_name=self.stack_name,
+                cfn_template=cfn_file.read(),
+                parameter_values=[],
+                capabilities=["CAPABILITY_IAM", "CAPABILITY_AUTO_EXPAND"],
+                role_arn=None,
+                notification_arns=[],
+                s3_uploader=None,
+                tags=[],
+            )
+            self.deployer.execute_changeset(result["Id"], self.stack_name)
+            self.deployer.wait_for_execute(self.stack_name, changeset_type)
+
+    def _verify_stack(self, expected_resource_path):
+        stacks_description = self.cloudformation_client.describe_stacks(StackName=self.stack_name)
+        stack_resources = self.cloudformation_client.list_stack_resources(StackName=self.stack_name)
+        # verify if the stack was successfully created
+        self.assertEqual(stacks_description["Stacks"][0]["StackStatus"], "CREATE_COMPLETE")
+        # verify if the stack contains the expected resources
+        self.assertTrue(verify_stack_resources(expected_resource_path, stack_resources))

tests_integ/helpers/helpers.py

@@ -0,0 +1,89 @@
+import json
+import logging
+import re
+import random
+import string  # not deprecated, a bug from pylint https://www.logilab.org/ticket/2481
+from functools import reduce
+
+import boto3
+from botocore.exceptions import ClientError
+
+from samtranslator.model.exceptions import InvalidDocumentException
+from samtranslator.translator.managed_policy_translator import ManagedPolicyLoader
+from samtranslator.translator.transform import transform
+from samtranslator.yaml_helper import yaml_parse
+
+RANDOM_SUFFIX_LENGTH = 12
+
+
+def transform_template(input_file_path, output_file_path):
+    LOG = logging.getLogger(__name__)
+    iam_client = boto3.client("iam")
+
+    with open(input_file_path, "r") as f:
+        sam_template = yaml_parse(f)
+
+    try:
+        cloud_formation_template = transform(sam_template, {}, ManagedPolicyLoader(iam_client))
+        cloud_formation_template_prettified = json.dumps(cloud_formation_template, indent=2)
+
+        with open(output_file_path, "w") as f:
+            f.write(cloud_formation_template_prettified)
+
+        print("Wrote transformed CloudFormation template to: " + output_file_path)
+    except InvalidDocumentException as e:
+        errorMessage = reduce(lambda message, error: message + " " + error.message, e.causes, e.message)
+        LOG.error(errorMessage)
+        errors = map(lambda cause: cause.message, e.causes)
+        LOG.error(errors)
+
+
+def verify_stack_resources(expected_file_path, stack_resources):
+    with open(expected_file_path, "r") as expected_data:
+        expected_resources = _sort_resources(json.load(expected_data))
+    parsed_resources = _sort_resources(stack_resources["StackResourceSummaries"])
+
+    if len(expected_resources) != len(parsed_resources):
+        return False
+
+    for i in range(len(expected_resources)):
+        exp = expected_resources[i]
+        parsed = parsed_resources[i]
+        if not re.fullmatch(exp["LogicalResourceId"] + "([0-9a-f]{10})?", parsed["LogicalResourceId"]):
+            return False
+        if exp["ResourceType"] != parsed["ResourceType"]:
+            return False
+    return True
+
+
+def generate_suffix():
+    # Very basic random letters generator
+    return "".join(random.choice(string.ascii_lowercase) for i in range(RANDOM_SUFFIX_LENGTH))
+
+
+def _sort_resources(resources):
+    return sorted(resources, key=lambda d: d["LogicalResourceId"])
+
+
+def create_bucket(bucket_name, region=None):
+    """Create an S3 bucket in a specified region
+
+    copy code from boto3 doc example
+    MG: removed the try so that the exception bubbles up and interrupts the test
+
+    If a region is not specified, the bucket is created in the S3 default
+    region (us-east-1).
+
+    :param bucket_name: Bucket to create
+    :param region: String region to create bucket in, e.g., 'us-west-2'
+    :return: True if bucket created, else False
+    """
+
+    # Create bucket
+    if region is None:
+        s3_client = boto3.client("s3")
+        s3_client.create_bucket(Bucket=bucket_name)
+    else:
+        s3_client = boto3.client("s3", region_name=region)
+        location = {"LocationConstraint": region}
+        s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration=location)