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_update_cancels_after_timeout(self, mock_wait_for_completion):
self.actions.stack._template = Mock(spec=Template)
self.actions.stack._template.get_boto_call_parameter.return_value = {
"Template": sentinel.template
}
mock_wait_for_completion.return_value = StackStatus.IN_PROGRESS
self.actions.update()
calls = [
call(
service="cloudformation",
command="update_stack",
kwargs={
"StackName": sentinel.external_name,
"Template": sentinel.template,
"Parameters": [{
"ParameterKey": "key1",
"ParameterValue": "val1"
}],
"Capabilities": ['CAPABILITY_IAM',
'CAPABILITY_NAMED_IAM',
'CAPABILITY_AUTO_EXPAND'],
("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 _wait_for_completion(self, timeout=0):
"""
Waits for a Stack operation to finish. Prints CloudFormation events
while it waits.
:param timeout: Timeout before returning, in minutes.
:returns: The final Stack status.
:rtype: sceptre.stack_status.StackStatus
"""
timeout = 60 * timeout
def timed_out(elapsed):
return elapsed >= timeout if timeout else False
status = StackStatus.IN_PROGRESS
self.most_recent_event_datetime = (
datetime.now(tzutc()) - timedelta(seconds=3)
)
elapsed = 0
while status == StackStatus.IN_PROGRESS and not timed_out(elapsed):
status = self._get_simplified_status(self._get_status())
self._log_new_events()
time.sleep(4)
elapsed += 4
return status
}
update_stack_kwargs.update(
self.stack.template.get_boto_call_parameter())
update_stack_kwargs.update(self._get_role_arn())
response = self.connection_manager.call(
service="cloudformation",
command="update_stack",
kwargs=update_stack_kwargs
)
status = self._wait_for_completion(self.stack.stack_timeout)
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
"%s - Stack is in the %s state", self.stack.name, existing_status
)
if existing_status == "PENDING":
status = self.create()
elif existing_status in ["CREATE_FAILED", "ROLLBACK_COMPLETE"]:
self.delete()
status = self.create()
elif existing_status.endswith("COMPLETE"):
status = self.update()
elif existing_status.endswith("IN_PROGRESS"):
self.logger.info(
"%s - Stack action is already in progress state and cannot "
"be updated", self.stack.name
)
status = StackStatus.IN_PROGRESS
elif existing_status.endswith("FAILED"):
status = StackStatus.FAILED
raise CannotUpdateFailedStackError(
"'{0}' is in a the state '{1}' and cannot be updated".format(
self.stack.name, existing_status
)
)
else:
raise UnknownStackStatusError(
"{0} is unknown".format(existing_status)
)
return status
:returns: The final Stack status.
:rtype: sceptre.stack_status.StackStatus
"""
timeout = 60 * timeout
def timed_out(elapsed):
return elapsed >= timeout if timeout else False
status = StackStatus.IN_PROGRESS
self.most_recent_event_datetime = (
datetime.now(tzutc()) - timedelta(seconds=3)
)
elapsed = 0
while status == StackStatus.IN_PROGRESS and not timed_out(elapsed):
status = self._get_simplified_status(self._get_status())
self._log_new_events()
time.sleep(4)
elapsed += 4
return status
* 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)
)