import os import pathlib import re from subprocess import Popen, PIPE, TimeoutExpired import tempfile from unittest import skipIf from parameterized import parameterized, param from samcli.lib.utils.hash import dir_checksum from .package_integ_base import PackageIntegBase from tests.testing_utils import RUNNING_ON_CI, RUNNING_TEST_FOR_MASTER_ON_CI, RUN_BY_CANARY # Package tests require credentials and CI/CD will only add credentials to the env if the PR is from the same repo. # This is to restrict package tests to run outside of CI/CD, when the branch is not master and tests are not run by Canary. SKIP_PACKAGE_TESTS = RUNNING_ON_CI and RUNNING_TEST_FOR_MASTER_ON_CI and not RUN_BY_CANARY TIMEOUT = 300 @skipIf(SKIP_PACKAGE_TESTS, "Skip package tests in CI/CD only") class TestPackageZip(PackageIntegBase): def setUp(self): super().setUp() def tearDown(self): super().tearDown() @parameterized.expand(["aws-serverless-function.yaml", "cdk_v1_synthesized_template_zip_functions.json"]) def test_package_template_flag(self, template_file): template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, s3_prefix=self.s3_prefix, template=template_path ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn("{bucket_name}".format(bucket_name=self.s3_bucket.name), process_stdout.decode("utf-8")) @parameterized.expand( [ ("cdk_v1_synthesized_template_Level1_nested_zip_functions.json", 3), ("cdk_v1_synthesized_template_Level2_nested_zip_functions.json", 2), ] ) def test_package_nested_template(self, template_file, uploading_count): template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, s3_prefix=self.s3_prefix, template=template_path, force_upload=True ) process = Popen(command_list, stderr=PIPE) try: _, stderr = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stderr = stderr.strip().decode(("utf-8")) uploads = re.findall(r"Uploading to.+", process_stderr) self.assertEqual(len(uploads), uploading_count) @parameterized.expand( [ "cdk_v1_synthesized_template_zip_functions.json", "aws-serverless-function.yaml", "aws-serverless-api.yaml", "aws-serverless-httpapi.yaml", "aws-appsync-graphqlschema.yaml", "aws-appsync-resolver.yaml", "aws-appsync-functionconfiguration.yaml", "aws-lambda-function.yaml", "aws-apigateway-restapi.yaml", "aws-apigatewayv2-httpapi.yaml", "aws-elasticbeanstalk-applicationversion.yaml", "aws-cloudformation-moduleversion.yaml", "aws-cloudformation-resourceversion.yaml", "aws-cloudformation-stack.yaml", "aws-serverless-application.yaml", "aws-lambda-layerversion.yaml", "aws-serverless-layerversion.yaml", "aws-glue-job.yaml", "aws-serverlessrepo-application.yaml", "aws-serverless-statemachine.yaml", "aws-stepfunctions-statemachine.yaml", "aws-include-transform.yaml", "aws-serverless-graphqlapi.yaml", ] ) def test_package_barebones(self, template_file): template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, s3_prefix=self.s3_prefix, template_file=template_path ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn("{bucket_name}".format(bucket_name=self.s3_bucket.name), process_stdout.decode("utf-8")) def test_package_without_required_args(self): command_list = PackageIntegBase.get_command_list() process = Popen(command_list, stdout=PIPE) try: process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise self.assertNotEqual(process.returncode, 0) @parameterized.expand( [ "cdk_v1_synthesized_template_zip_functions.json", "aws-serverless-function.yaml", "aws-serverless-api.yaml", "aws-serverless-httpapi.yaml", "aws-appsync-graphqlschema.yaml", "aws-appsync-resolver.yaml", "aws-appsync-functionconfiguration.yaml", "aws-lambda-function.yaml", "aws-apigateway-restapi.yaml", "aws-apigatewayv2-httpapi.yaml", "aws-elasticbeanstalk-applicationversion.yaml", "aws-cloudformation-moduleversion.yaml", "aws-cloudformation-resourceversion.yaml", "aws-cloudformation-stack.yaml", "aws-serverless-application.yaml", "aws-lambda-layerversion.yaml", "aws-serverless-layerversion.yaml", "aws-glue-job.yaml", "aws-serverlessrepo-application.yaml", "aws-serverless-statemachine.yaml", "aws-stepfunctions-statemachine.yaml", "aws-serverless-graphqlapi.yaml", ] ) def test_package_with_prefix(self, template_file): template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, template_file=template_path, s3_prefix=self.s3_prefix ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn("{bucket_name}".format(bucket_name=self.s3_bucket.name), process_stdout.decode("utf-8")) self.assertIn("{s3_prefix}".format(s3_prefix=self.s3_prefix), process_stdout.decode("utf-8")) @parameterized.expand( [ "cdk_v1_synthesized_template_zip_functions.json", "aws-serverless-function.yaml", "aws-serverless-api.yaml", "aws-serverless-httpapi.yaml", "aws-appsync-graphqlschema.yaml", "aws-appsync-resolver.yaml", "aws-appsync-functionconfiguration.yaml", "aws-lambda-function.yaml", "aws-apigateway-restapi.yaml", "aws-apigatewayv2-httpapi.yaml", "aws-elasticbeanstalk-applicationversion.yaml", "aws-cloudformation-moduleversion.yaml", "aws-cloudformation-resourceversion.yaml", "aws-cloudformation-stack.yaml", "aws-serverless-application.yaml", "aws-lambda-layerversion.yaml", "aws-serverless-layerversion.yaml", "aws-glue-job.yaml", "aws-serverlessrepo-application.yaml", "aws-serverless-statemachine.yaml", "aws-stepfunctions-statemachine.yaml", "aws-serverless-graphqlapi.yaml", ] ) def test_package_with_output_template_file(self, template_file): template_path = self.test_data_path.joinpath(template_file) with tempfile.NamedTemporaryFile(delete=False) as output_template: command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, template_file=template_path, s3_prefix=self.s3_prefix, output_template_file=output_template.name, ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn( bytes( "Successfully packaged artifacts and wrote output template to file {output_template_file}".format( output_template_file=str(output_template.name) ), encoding="utf-8", ), process_stdout, ) @parameterized.expand( [ "cdk_v1_synthesized_template_zip_functions.json", "aws-serverless-function.yaml", "aws-serverless-api.yaml", "aws-serverless-httpapi.yaml", "aws-appsync-graphqlschema.yaml", "aws-appsync-resolver.yaml", "aws-appsync-functionconfiguration.yaml", "aws-lambda-function.yaml", "aws-apigateway-restapi.yaml", "aws-apigatewayv2-httpapi.yaml", "aws-elasticbeanstalk-applicationversion.yaml", "aws-cloudformation-moduleversion.yaml", "aws-cloudformation-resourceversion.yaml", "aws-cloudformation-stack.yaml", "aws-serverless-application.yaml", "aws-lambda-layerversion.yaml", "aws-serverless-layerversion.yaml", "aws-glue-job.yaml", "aws-serverlessrepo-application.yaml", "aws-serverless-statemachine.yaml", "aws-stepfunctions-statemachine.yaml", "aws-serverless-graphqlapi.yaml", ] ) def test_package_with_json(self, template_file): template_path = self.test_data_path.joinpath(template_file) with tempfile.NamedTemporaryFile(delete=False) as output_template: command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, template_file=template_path, s3_prefix=self.s3_prefix, output_template_file=output_template.name, use_json=True, ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn( bytes( "Successfully packaged artifacts and wrote output template to file {output_template_file}".format( output_template_file=str(output_template.name) ), encoding="utf-8", ), process_stdout, ) @parameterized.expand( [ "cdk_v1_synthesized_template_zip_functions.json", "aws-serverless-function.yaml", "aws-serverless-api.yaml", "aws-serverless-httpapi.yaml", "aws-appsync-graphqlschema.yaml", "aws-appsync-resolver.yaml", "aws-appsync-functionconfiguration.yaml", "aws-lambda-function.yaml", "aws-apigateway-restapi.yaml", "aws-apigatewayv2-httpapi.yaml", "aws-elasticbeanstalk-applicationversion.yaml", "aws-cloudformation-moduleversion.yaml", "aws-cloudformation-resourceversion.yaml", "aws-cloudformation-stack.yaml", "aws-serverless-application.yaml", "aws-lambda-layerversion.yaml", "aws-serverless-layerversion.yaml", "aws-glue-job.yaml", "aws-serverlessrepo-application.yaml", "aws-serverless-statemachine.yaml", "aws-stepfunctions-statemachine.yaml", "aws-serverless-graphqlapi.yaml", ] ) def test_package_with_force_upload(self, template_file): template_path = self.test_data_path.joinpath(template_file) with tempfile.NamedTemporaryFile(delete=False) as output_template: # Upload twice and see the string to have packaged artifacts both times. for _ in range(2): command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, template_file=template_path, s3_prefix=self.s3_prefix, output_template_file=output_template.name, force_upload=True, ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn( bytes( "Successfully packaged artifacts and wrote output template to file {output_template_file}".format( output_template_file=str(output_template.name) ), encoding="utf-8", ), process_stdout, ) @parameterized.expand( [ "cdk_v1_synthesized_template_zip_functions.json", "aws-serverless-function.yaml", "aws-serverless-api.yaml", "aws-serverless-httpapi.yaml", "aws-appsync-graphqlschema.yaml", "aws-appsync-resolver.yaml", "aws-appsync-functionconfiguration.yaml", "aws-lambda-function.yaml", "aws-apigateway-restapi.yaml", "aws-apigatewayv2-httpapi.yaml", "aws-elasticbeanstalk-applicationversion.yaml", "aws-cloudformation-moduleversion.yaml", "aws-cloudformation-resourceversion.yaml", "aws-cloudformation-stack.yaml", "aws-serverless-application.yaml", "aws-lambda-layerversion.yaml", "aws-serverless-layerversion.yaml", "aws-glue-job.yaml", "aws-serverlessrepo-application.yaml", "aws-serverless-statemachine.yaml", "aws-stepfunctions-statemachine.yaml", "aws-serverless-graphqlapi.yaml", ] ) def test_package_with_kms_key(self, template_file): template_path = self.test_data_path.joinpath(template_file) with tempfile.NamedTemporaryFile(delete=False) as output_template: command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, template_file=template_path, s3_prefix=self.s3_prefix, output_template_file=output_template.name, force_upload=True, kms_key_id=self.kms_key, ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn( bytes( "Successfully packaged artifacts and wrote output template to file {output_template_file}".format( output_template_file=str(output_template.name) ), encoding="utf-8", ), process_stdout, ) @parameterized.expand( [ "cdk_v1_synthesized_template_zip_functions.json", "aws-serverless-function.yaml", "aws-serverless-api.yaml", "aws-serverless-httpapi.yaml", "aws-appsync-graphqlschema.yaml", "aws-appsync-resolver.yaml", "aws-appsync-functionconfiguration.yaml", "aws-lambda-function.yaml", "aws-apigateway-restapi.yaml", "aws-apigatewayv2-httpapi.yaml", "aws-elasticbeanstalk-applicationversion.yaml", "aws-cloudformation-moduleversion.yaml", "aws-cloudformation-resourceversion.yaml", "aws-cloudformation-stack.yaml", "aws-serverless-application.yaml", "aws-lambda-layerversion.yaml", "aws-serverless-layerversion.yaml", "aws-glue-job.yaml", "aws-serverlessrepo-application.yaml", "aws-serverless-statemachine.yaml", "aws-stepfunctions-statemachine.yaml", "aws-serverless-graphqlapi.yaml", ] ) def test_package_with_metadata(self, template_file): template_path = self.test_data_path.joinpath(template_file) with tempfile.NamedTemporaryFile(delete=False) as output_template: command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, template_file=template_path, s3_prefix=self.s3_prefix, output_template_file=output_template.name, force_upload=True, metadata={"integ": "yes"}, ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn( bytes( "Successfully packaged artifacts and wrote output template to file {output_template_file}".format( output_template_file=str(output_template.name) ), encoding="utf-8", ), process_stdout, ) @parameterized.expand( [ "cdk_v1_synthesized_template_zip_functions.json", "aws-serverless-function.yaml", "aws-serverless-api.yaml", "aws-appsync-graphqlschema.yaml", "aws-appsync-resolver.yaml", "aws-appsync-functionconfiguration.yaml", "aws-lambda-function.yaml", "aws-apigateway-restapi.yaml", "aws-apigatewayv2-httpapi.yaml", "aws-elasticbeanstalk-applicationversion.yaml", "aws-cloudformation-moduleversion.yaml", "aws-cloudformation-resourceversion.yaml", "aws-cloudformation-stack.yaml", "aws-serverless-application.yaml", "aws-lambda-layerversion.yaml", "aws-serverless-layerversion.yaml", "aws-glue-job.yaml", "aws-serverlessrepo-application.yaml", "aws-serverless-statemachine.yaml", "aws-stepfunctions-statemachine.yaml", "aws-serverless-graphqlapi.yaml", ] ) def test_package_with_resolve_s3(self, template_file): template_path = self.test_data_path.joinpath(template_file) with tempfile.NamedTemporaryFile(delete=False) as output_template: command_list = PackageIntegBase.get_command_list( template_file=template_path, s3_prefix=self.s3_prefix, output_template_file=output_template.name, force_upload=True, resolve_s3=True, ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() self.assertIn( bytes( "Successfully packaged artifacts and wrote output template to file {output_template_file}".format( output_template_file=str(output_template.name) ), encoding="utf-8", ), process_stdout, ) @parameterized.expand([(True,), (False,)]) def test_package_with_no_progressbar(self, no_progressbar): template_path = self.test_data_path.joinpath("aws-serverless-function.yaml") with tempfile.NamedTemporaryFile(delete=False) as output_template: command_list = PackageIntegBase.get_command_list( template_file=template_path, s3_prefix=self.s3_prefix, output_template_file=output_template.name, force_upload=True, no_progressbar=no_progressbar, resolve_s3=True, ) process = Popen(command_list, stdout=PIPE, stderr=PIPE) try: _, stderr = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stderr = stderr.strip() upload_message = bytes("Uploading to", encoding="utf-8") if no_progressbar: self.assertNotIn( upload_message, process_stderr, ) else: self.assertIn( upload_message, process_stderr, ) @parameterized.expand( [ param("aws-serverless-function-codedeploy-warning.yaml", "CodeDeploy"), param("aws-serverless-function-codedeploy-condition-warning.yaml", "CodeDeploy DeploymentGroups"), ] ) def test_package_with_warning_template(self, template_file, warning_keyword): template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, s3_prefix=self.s3_prefix, template=template_path ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip().decode("utf-8") # Not comparing with full warning message because of line ending mismatch on # windows and non-windows self.assertIn(warning_keyword, process_stdout) def test_package_with_deep_nested_template(self): """ this template contains two nested stacks: - root - FunctionA - ChildStackX - FunctionB - ChildStackY - FunctionA - MyLayerVersion """ template_file = os.path.join("deep-nested", "template.yaml") template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, s3_prefix=self.s3_prefix, template=template_path, force_upload=True ) process = Popen(command_list, stdout=PIPE, stderr=PIPE) try: _, stderr = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stderr = stderr.strip().decode("utf-8") # there are in total 3 function dir, 1 layer dir and 2 child templates to upload uploads = re.findall(r"Uploading to.+", process_stderr) self.assertEqual(len(uploads), 6) # make sure uploads' checksum match the dirs and child templates build_dir = pathlib.Path(os.path.dirname(__file__)).parent.joinpath("testdata", "package", "deep-nested") dirs = [ build_dir.joinpath("FunctionA"), build_dir.joinpath("ChildStackX", "FunctionB"), build_dir.joinpath("ChildStackX", "ChildStackY", "FunctionA"), build_dir.joinpath("ChildStackX", "ChildStackY", "MyLayerVersion"), ] # here we only verify function/layer code dirs' hash # because templates go through some pre-process before being uploaded and the hash can not be determined for dir in dirs: checksum = dir_checksum(dir.absolute()) self.assertIn(checksum, process_stderr) # verify both child templates are uploaded uploads = re.findall(r"\.template", process_stderr) self.assertEqual(len(uploads), 2) def test_package_with_stackset(self): """ this template contains a stack set: - root - FunctionA - StackSetA """ template_file = os.path.join("stackset", "template.yaml") template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, s3_prefix=self.s3_prefix, template=template_path, force_upload=True ) prevdir = os.getcwd() os.chdir(os.path.expanduser(os.path.dirname(template_path))) process = Popen(command_list, stdout=PIPE, stderr=PIPE) try: _, stderr = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stderr = stderr.strip().decode("utf-8") os.chdir(prevdir) # there are in total 1 function dir, 1 stackset template to upload uploads = re.findall(r"Uploading to.+", process_stderr) self.assertEqual(len(uploads), 2) # make sure uploads' checksum match the dirs and stackset template build_dir = pathlib.Path(os.path.dirname(__file__)).parent.joinpath("testdata", "package", "stackset") dirs = [build_dir.joinpath("FunctionA")] # here we only verify function/layer code dirs' hash # because templates go through some pre-process before being uploaded and the hash can not be determined for dir in dirs: checksum = dir_checksum(dir.absolute()) self.assertIn(checksum, process_stderr) # verify stack set template is uploaded uploads = re.findall(r"\.template", process_stderr) self.assertEqual(len(uploads), 1) def test_package_with_stackset_in_a_substack(self): """ this template contains a stack set: - root - ChildStackX - FunctionA - StackSetA """ template_file = os.path.join("stackset", "nested-template.yaml") template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, s3_prefix=self.s3_prefix, template=template_path, force_upload=True ) prevdir = os.getcwd() os.chdir(os.path.expanduser(os.path.dirname(template_path))) process = Popen(command_list, stdout=PIPE, stderr=PIPE) try: _, stderr = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stderr = stderr.strip().decode("utf-8") os.chdir(prevdir) # there are in total 1 child stack, 1 function dir, 1 stackset template to upload uploads = re.findall(r"Uploading to.+", process_stderr) self.assertEqual(len(uploads), 3) # make sure uploads' checksum match the dirs and stackset template build_dir = pathlib.Path(os.path.dirname(__file__)).parent.joinpath("testdata", "package", "stackset") dirs = [build_dir.joinpath("FunctionA")] # here we only verify function/layer code dirs' hash # because templates go through some pre-process before being uploaded and the hash can not be determined for dir in dirs: checksum = dir_checksum(dir.absolute()) self.assertIn(checksum, process_stderr) # verify child stack and stack set templates are uploaded uploads = re.findall(r"\.template", process_stderr) self.assertEqual(len(uploads), 2) @parameterized.expand(["aws-serverless-function-cdk.yaml", "cdk_v1_synthesized_template_zip_functions.json"]) def test_package_logs_warning_for_cdk_project(self, template_file): template_path = self.test_data_path.joinpath(template_file) command_list = PackageIntegBase.get_command_list( s3_bucket=self.s3_bucket.name, s3_prefix=self.s3_prefix, template_file=template_path ) process = Popen(command_list, stdout=PIPE) try: stdout, _ = process.communicate(timeout=TIMEOUT) except TimeoutExpired: process.kill() raise process_stdout = stdout.strip() warning_message = bytes( f"Warning: CDK apps are not officially supported with this command.{os.linesep}" "We recommend you use this alternative command: cdk deploy", encoding="utf-8", ) self.assertIn(warning_message, stdout) self.assertIn("{bucket_name}".format(bucket_name=self.s3_bucket.name), process_stdout.decode("utf-8"))