Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_stack_commands(self, command, success, yes_flag, exit_code):
run_command = getattr(self.mock_stack_actions, command)
run_command.return_value = \
StackStatus.COMPLETE if success else StackStatus.FAILED
kwargs = {"args": [command, "dev/vpc.yaml"]}
if yes_flag:
kwargs["args"].append("-y")
else:
kwargs["input"] = "y\n"
result = self.runner.invoke(cli, **kwargs)
run_command.assert_called_with()
assert result.exit_code == exit_code
("STACK_COMPLETE", StackStatus.COMPLETE),
("STACK_IN_PROGRESS", StackStatus.IN_PROGRESS),
("STACK_FAILED", StackStatus.FAILED)
])
def test_get_simplified_status_with_known_stack_statuses(
self, test_input, expected
):
response = self.actions._get_simplified_status(test_input)
assert response == expected
def test_wait_for_completion_calls_log_new_events(
self, mock_get_simplified_status, mock_get_status,
mock_log_new_events
):
mock_get_simplified_status.return_value = StackStatus.COMPLETE
self.actions._wait_for_completion()
mock_log_new_events.assert_called_once_with()
def test_launch_with_complete_stack_with_no_updates_to_perform(
self, mock_get_status, mock_update
):
mock_get_status.return_value = "CREATE_COMPLETE"
mock_update.return_value = StackStatus.COMPLETE
response = self.actions.launch()
mock_update.assert_called_once_with()
assert response == StackStatus.COMPLETE
def create_stack(ctx, environment, stack):
"""
Creates the stack.
Creates ENVIRONMENT/STACK.
"""
stack = _get_stack(environment, stack)
response = stack.create()
if response != StackStatus.COMPLETE:
exit(1)
The simplified Stack status is represented by the struct
``sceptre.StackStatus()`` and can take one of the following options:
* complete
* in_progress
* failed
:param status: The CloudFormation Stack status to simplify.
:type status: str
:returns: The Stack's simplified status
:rtype: sceptre.stack_status.StackStatus
"""
if status.endswith("ROLLBACK_COMPLETE"):
return StackStatus.FAILED
elif status.endswith("_COMPLETE"):
return StackStatus.COMPLETE
elif status.endswith("_IN_PROGRESS"):
return StackStatus.IN_PROGRESS
elif status.endswith("_FAILED"):
return StackStatus.FAILED
else:
raise UnknownStackStatusError(
"{0} is unknown".format(status)
)
def stack_status_exit_code(statuses):
if not all(
status == StackStatus.COMPLETE
for status in statuses):
return 1
else:
return 0
self.logger.debug(
"%s - Update Stack response: %s", self.stack.name, response
)
# Cancel update after timeout
if status == StackStatus.IN_PROGRESS:
status = self.cancel_stack_update()
return status
except botocore.exceptions.ClientError as exp:
error_message = exp.response["Error"]["Message"]
if error_message == "No updates are to be performed.":
self.logger.info(
"%s - No updates to perform.", self.stack.name
)
return StackStatus.COMPLETE
else:
raise
delete_stack_kwargs = {"StackName": self.stack.external_name}
delete_stack_kwargs.update(self._get_role_arn())
self.connection_manager.call(
service="cloudformation",
command="delete_stack",
kwargs=delete_stack_kwargs
)
try:
status = self._wait_for_completion()
except StackDoesNotExistError:
status = StackStatus.COMPLETE
except botocore.exceptions.ClientError as error:
if error.response["Error"]["Message"].endswith("does not exist"):
status = StackStatus.COMPLETE
else:
raise
self.logger.info("%s - delete %s", self.stack.name, status)
return status
self.logger.info("%s - Does not exist.", self.stack.name)
status = StackStatus.COMPLETE
return status
delete_stack_kwargs = {"StackName": self.stack.external_name}
delete_stack_kwargs.update(self._get_role_arn())
self.connection_manager.call(
service="cloudformation",
command="delete_stack",
kwargs=delete_stack_kwargs
)
try:
status = self._wait_for_completion()
except StackDoesNotExistError:
status = StackStatus.COMPLETE
except botocore.exceptions.ClientError as error:
if error.response["Error"]["Message"].endswith("does not exist"):
status = StackStatus.COMPLETE
else:
raise
self.logger.info("%s - delete %s", self.stack.name, status)
return status