Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions tests/integration/pipeline/test_bootstrap_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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):
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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):
"""
Expand Down Expand Up @@ -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)

Expand Down