Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
params_diff=[],
fqn=None))
self.assertEqual(prompt.call_count, 1)
safe_dump_counter += 1
self.assertEqual(mock_safe_dump.call_count, safe_dump_counter)
self.assertEqual(patched_format.call_count, 0)
for v in ['n', 'N']:
with patch(get_input_path, return_value=v) as prompt:
output_full_changeset(full_changeset=[], params_diff=[],
answer=None, fqn=None)
self.assertEqual(prompt.call_count, 1)
self.assertEqual(mock_safe_dump.call_count, safe_dump_counter)
self.assertEqual(patched_format.call_count, 0)
with self.assertRaises(exceptions.CancelExecution):
output_full_changeset(full_changeset=[], params_diff=[],
answer='x', fqn=None)
output_full_changeset(full_changeset=[], params_diff=['mock'],
answer='y', fqn=None)
safe_dump_counter += 1
self.assertEqual(mock_safe_dump.call_count, safe_dump_counter)
self.assertEqual(patched_format.call_count, 1)
def test_missing_params_no_existing_stack(self):
"""Test missing params no existing stack."""
all_params = ["Address", "StackName"]
required = ["Address"]
parameter_values = {}
with self.assertRaises(exceptions.MissingParameterException) as result:
_handle_missing_parameters(parameter_values, all_params, required)
self.assertEqual(result.exception.parameters, required)
def test_render_missing_env(self):
"""Test render missing env."""
env = {"a": "A"}
with self.assertRaises(exceptions.MissingEnvironment) as expected:
render(CONFIG, env)
self.assertEqual(expected.exception.key, "b")
def test_destroy_stack_canceled(self, patched_input):
"""Test destroy stack canceled."""
stack = {'StackName': 'MockStack'}
patched_input.return_value = 'n'
with self.assertRaises(exceptions.CancelExecution):
self.provider.destroy_stack(stack)
def get_stack(self, stack_name, *args, **kwargs):
"""Get stack."""
if stack_name not in self._outputs:
raise exceptions.StackDoesNotExist(stack_name)
return {"name": stack_name, "outputs": self._outputs[stack_name]}
def test_prepare_stack_for_update_non_recreatable(self):
"""Test prepare stack for update non recreatable."""
stack_name = "MockStack"
stack = generate_describe_stacks_stack(
stack_name, stack_status="REVIEW_IN_PROGRESS")
with self.assertRaises(exceptions.StackUpdateBadStatus) as raised:
with self.stubber:
self.provider.prepare_stack_for_update(stack, [])
self.assertIn('Unsupported state', str(raised.exception))
Returns:
Dict[str, Any]: Stack outputs with inferred changes.
"""
try:
stack_details = self.get_stack(stack.fqn)
# handling for orphaned changeset temp stacks
if self.get_stack_status(
stack_details) == self.REVIEW_STATUS:
raise exceptions.StackDoesNotExist(stack.fqn)
_old_template, old_params = self.get_stack_info(
stack_details
)
old_template = parse_cloudformation_template(_old_template)
change_type = 'UPDATE'
except exceptions.StackDoesNotExist:
old_params = {}
old_template = {}
change_type = 'CREATE'
changes, change_set_id = create_change_set(
self.cloudformation, stack.fqn, template, parameters, tags,
change_type, service_role=self.service_role
)
new_parameters_as_dict = self.params_as_dict(
[x
if 'ParameterValue' in x
else {'ParameterKey': x['ParameterKey'],
'ParameterValue': old_params[x['ParameterKey']]}
for x in parameters]
)
params_diff = diff_parameters(old_params, new_parameters_as_dict)
provider = self.build_provider(stack)
if not build.should_update(stack):
stack.set_outputs(provider.get_outputs(stack.fqn))
return NotUpdatedStatus()
tags = build.build_stack_tags(stack)
try:
stack.resolve(self.context, provider)
parameters = self.build_parameters(stack)
outputs = provider.get_stack_changes(
stack, self._template(stack.blueprint), parameters, tags
)
stack.set_outputs(outputs)
except exceptions.StackDidNotChange:
LOGGER.info('No changes: %s', stack.fqn)
stack.set_outputs(provider.get_outputs(stack.fqn))
except exceptions.StackDoesNotExist:
if self.context.persistent_graph:
return SkippedStatus('persistent graph: stack does not '
'exist, will be removed')
return StackDoesNotExistStatus()
except AttributeError as err:
if (self.context.persistent_graph and
'defined class or template path' in str(err)):
return SkippedStatus('persistent graph: will be destroyed')
raise
return COMPLETE
change_set_id = response["Id"]
response = wait_till_change_set_complete(
cfn_client, change_set_id
)
status = response["Status"]
if status == "FAILED":
status_reason = response["StatusReason"]
if ("didn't contain changes" in response["StatusReason"] or
"No updates are to be performed" in response["StatusReason"]):
LOGGER.debug(
"Stack %s did not change, not updating and removing "
"changeset.",
fqn,
)
cfn_client.delete_change_set(ChangeSetName=change_set_id)
raise exceptions.StackDidNotChange()
LOGGER.warning(
"Got strange status, '%s' for changeset '%s'. Not deleting for "
"further investigation - you will need to delete the changeset "
"manually.",
status, change_set_id
)
raise exceptions.UnhandledChangeSetStatus(
fqn, change_set_id, status, status_reason
)
execution_status = response["ExecutionStatus"]
if execution_status != "AVAILABLE":
raise exceptions.UnableToExecuteChangeSet(fqn,
change_set_id,
execution_status)