diff --git a/tests/integration/pipeline/test_bootstrap_command.py b/tests/integration/pipeline/test_bootstrap_command.py index 43b5729e16..0cf7741c5c 100644 --- a/tests/integration/pipeline/test_bootstrap_command.py +++ b/tests/integration/pipeline/test_bootstrap_command.py @@ -2,6 +2,8 @@ from parameterized import parameterized +from samcli.commands.pipeline.bootstrap.cli import PIPELINE_CONFIG_FILENAME, PIPELINE_CONFIG_DIR +from samcli.lib.config.samconfig import SamConfig from tests.integration.pipeline.base import BootstrapIntegBase from tests.testing_utils import ( run_command_with_input, @@ -21,6 +23,13 @@ # In order to run bootstrap integration test locally make sure your test account is configured as `default` account. CREDENTIAL_PROFILE = "2" if not RUN_BY_CANARY else "1" +CFN_OUTPUT_TO_CONFIG_KEY = { + "ArtifactsBucket": "artifacts_bucket", + "CloudFormationExecutionRole": "cloudformation_execution_role", + "PipelineExecutionRole": "pipeline_execution_role", + "PipelineUser": "pipeline_user", +} + @skipIf(SKIP_BOOTSTRAP_TESTS, "Skip bootstrap tests in CI/CD only") class TestBootstrap(BootstrapIntegBase): @@ -76,8 +85,12 @@ def test_interactive_with_no_resources_provided(self, create_image_repository): }, set(self._extract_created_resource_logical_ids(stack_name)), ) + CFN_OUTPUT_TO_CONFIG_KEY["ImageRepository"] = "image_repository" + self.validate_pipeline_config(stack_name, stage_name, list(CFN_OUTPUT_TO_CONFIG_KEY.keys())) + del CFN_OUTPUT_TO_CONFIG_KEY["ImageRepository"] else: self.assertSetEqual(common_resources, set(self._extract_created_resource_logical_ids(stack_name))) + self.validate_pipeline_config(stack_name, stage_name) @parameterized.expand([("create_image_repository",), (False,)]) def test_non_interactive_with_no_resources_provided(self, create_image_repository): @@ -142,6 +155,34 @@ def test_no_interactive_with_all_required_resources_provided(self): stdout = bootstrap_process_execute.stdout.decode() self.assertIn("skipping creation", stdout) + def validate_pipeline_config(self, stack_name, stage_name, cfn_keys_to_check=None): + # Get output values from cloudformation + if cfn_keys_to_check is None: + cfn_keys_to_check = list(CFN_OUTPUT_TO_CONFIG_KEY.keys()) + response = self.cf_client.describe_stacks(StackName=stack_name) + stacks = response["Stacks"] + self.assertTrue(len(stacks) > 0) # in case stack name is invalid + stack_outputs = stacks[0]["Outputs"] + output_values = {} + for value in stack_outputs: + output_values[value["OutputKey"]] = value["OutputValue"] + + # Get values saved in config file + config = SamConfig(PIPELINE_CONFIG_DIR, PIPELINE_CONFIG_FILENAME) + config_values = config.get_all(["pipeline", "bootstrap"], "parameters", stage_name) + config_values = {**config_values, **config.get_all(["pipeline", "bootstrap"], "parameters")} + + for key in CFN_OUTPUT_TO_CONFIG_KEY: + if key not in cfn_keys_to_check: + continue + value = CFN_OUTPUT_TO_CONFIG_KEY[key] + cfn_value = output_values[key] + config_value = config_values[value] + if key == "ImageRepository": + self.assertEqual(cfn_value.split("/")[-1], config_value.split("/")[-1]) + else: + self.assertTrue(cfn_value.endswith(config_value) or cfn_value == config_value) + @parameterized.expand([("confirm_changeset",), (False,)]) def test_no_interactive_with_some_required_resources_provided(self, confirm_changeset: bool): stage_name, stack_name = self._get_stage_and_stack_name() @@ -222,6 +263,7 @@ def test_interactive_with_some_required_resources_provided(self): self.assertIn("Successfully created!", stdout) # make sure the not provided resource is the only resource created. self.assertIn("CloudFormationExecutionRole", self._extract_created_resource_logical_ids(stack_name)) + self.validate_pipeline_config(stack_name, stage_name) def test_interactive_pipeline_user_only_created_once(self): """ @@ -264,6 +306,7 @@ def test_interactive_pipeline_user_only_created_once(self): self.assertTrue("PipelineUser" in resources) self.assertTrue("PipelineUserAccessKey" in resources) self.assertTrue("PipelineUserSecretKey" in resources) + self.validate_pipeline_config(self.stack_names[i], stage_name) else: self.assertIn("skipping creation", stdout)