from __future__ import unicode_literals
import json
from six.moves.urllib.parse import urlparse
from moto.core.responses import BaseResponse
from moto.s3 import s3_backend
from .models import cloudformation_backends
from .exceptions import ValidationError
class CloudFormationResponse(BaseResponse):
@property
def cloudformation_backend(self):
return cloudformation_backends[self.region]
def _get_stack_from_s3_url(self, template_url):
template_url_parts = urlparse(template_url)
if "localhost" in template_url:
bucket_name, key_name = template_url_parts.path.lstrip(
"/").split("/")
else:
bucket_name = template_url_parts.netloc.split(".")[0]
key_name = template_url_parts.path.lstrip("/")
key = s3_backend.get_key(bucket_name, key_name)
return key.value.decode("utf-8")
def create_stack(self):
stack_name = self._get_param('StackName')
stack_body = self._get_param('TemplateBody')
template_url = self._get_param('TemplateURL')
role_arn = self._get_param('RoleARN')
parameters_list = self._get_list_prefix("Parameters.member")
tags = dict((item['key'], item['value'])
for item in self._get_list_prefix("Tags.member"))
# Hack dict-comprehension
parameters = dict([
(parameter['parameter_key'], parameter['parameter_value'])
for parameter
in parameters_list
])
if template_url:
stack_body = self._get_stack_from_s3_url(template_url)
stack_notification_arns = self._get_multi_param(
'NotificationARNs.member')
stack = self.cloudformation_backend.create_stack(
name=stack_name,
template=stack_body,
parameters=parameters,
region_name=self.region,
notification_arns=stack_notification_arns,
tags=tags,
role_arn=role_arn,
)
if self.request_json:
return json.dumps({
'CreateStackResponse': {
'CreateStackResult': {
'StackId': stack.stack_id,
}
}
})
else:
template = self.response_template(CREATE_STACK_RESPONSE_TEMPLATE)
return template.render(stack=stack)
def describe_stacks(self):
stack_name_or_id = None
if self._get_param('StackName'):
stack_name_or_id = self.querystring.get('StackName')[0]
token = self._get_param('NextToken')
stacks = self.cloudformation_backend.describe_stacks(stack_name_or_id)
stack_ids = [stack.stack_id for stack in stacks]
if token:
start = stack_ids.index(token) + 1
else:
start = 0
max_results = 50 # using this to mske testing of paginated stacks more convenient than default 1 MB
stacks_resp = stacks[start:start + max_results]
next_token = None
if len(stacks) > (start + max_results):
next_token = stacks_resp[-1].stack_id
template = self.response_template(DESCRIBE_STACKS_TEMPLATE)
return template.render(stacks=stacks_resp, next_token=next_token)
def describe_stack_resource(self):
stack_name = self._get_param('StackName')
stack = self.cloudformation_backend.get_stack(stack_name)
logical_resource_id = self._get_param('LogicalResourceId')
for stack_resource in stack.stack_resources:
if stack_resource.logical_resource_id == logical_resource_id:
resource = stack_resource
break
else:
raise ValidationError(logical_resource_id)
template = self.response_template(
DESCRIBE_STACK_RESOURCE_RESPONSE_TEMPLATE)
return template.render(stack=stack, resource=resource)
def describe_stack_resources(self):
stack_name = self._get_param('StackName')
stack = self.cloudformation_backend.get_stack(stack_name)
template = self.response_template(DESCRIBE_STACK_RESOURCES_RESPONSE)
return template.render(stack=stack)
def describe_stack_events(self):
stack_name = self._get_param('StackName')
stack = self.cloudformation_backend.get_stack(stack_name)
template = self.response_template(DESCRIBE_STACK_EVENTS_RESPONSE)
return template.render(stack=stack)
def list_stacks(self):
stacks = self.cloudformation_backend.list_stacks()
template = self.response_template(LIST_STACKS_RESPONSE)
return template.render(stacks=stacks)
def list_stack_resources(self):
stack_name_or_id = self._get_param('StackName')
resources = self.cloudformation_backend.list_stack_resources(
stack_name_or_id)
template = self.response_template(LIST_STACKS_RESOURCES_RESPONSE)
return template.render(resources=resources)
def get_template(self):
name_or_stack_id = self.querystring.get('StackName')[0]
stack = self.cloudformation_backend.get_stack(name_or_stack_id)
if self.request_json:
return json.dumps({
"GetTemplateResponse": {
"GetTemplateResult": {
"TemplateBody": stack.template,
"ResponseMetadata": {
"RequestId": "2d06e36c-ac1d-11e0-a958-f9382b6eb86bEXAMPLE"
}
}
}
})
else:
template = self.response_template(GET_TEMPLATE_RESPONSE_TEMPLATE)
return template.render(stack=stack)
def update_stack(self):
stack_name = self._get_param('StackName')
role_arn = self._get_param('RoleARN')
if self._get_param('UsePreviousTemplate') == "true":
stack_body = self.cloudformation_backend.get_stack(
stack_name).template
else:
stack_body = self._get_param('TemplateBody')
parameters = dict([
(parameter['parameter_key'], parameter['parameter_value'])
for parameter
in self._get_list_prefix("Parameters.member")
])
# boto3 is supposed to let you clear the tags by passing an empty value, but the request body doesn't
# end up containing anything we can use to differentiate between passing an empty value versus not
# passing anything. so until that changes, moto won't be able to clear tags, only update them.
tags = dict((item['key'], item['value'])
for item in self._get_list_prefix("Tags.member"))
# so that if we don't pass the parameter, we don't clear all the tags accidentally
if not tags:
tags = None
stack = self.cloudformation_backend.get_stack(stack_name)
if stack.status == 'ROLLBACK_COMPLETE':
raise ValidationError(
stack.stack_id, message="Stack:{0} is in ROLLBACK_COMPLETE state and can not be updated.".format(stack.stack_id))
stack = self.cloudformation_backend.update_stack(
name=stack_name,
template=stack_body,
role_arn=role_arn,
parameters=parameters,
tags=tags,
)
if self.request_json:
stack_body = {
'UpdateStackResponse': {
'UpdateStackResult': {
'StackId': stack.name,
}
}
}
return json.dumps(stack_body)
else:
template = self.response_template(UPDATE_STACK_RESPONSE_TEMPLATE)
return template.render(stack=stack)
def delete_stack(self):
name_or_stack_id = self.querystring.get('StackName')[0]
self.cloudformation_backend.delete_stack(name_or_stack_id)
if self.request_json:
return json.dumps({
'DeleteStackResponse': {
'DeleteStackResult': {},
}
})
else:
template = self.response_template(DELETE_STACK_RESPONSE_TEMPLATE)
return template.render()
def list_exports(self):
token = self._get_param('NextToken')
exports, next_token = self.cloudformation_backend.list_exports(token=token)
template = self.response_template(LIST_EXPORTS_RESPONSE)
return template.render(exports=exports, next_token=next_token)
CREATE_STACK_RESPONSE_TEMPLATE = """
{{ stack.stack_id }}
b9b4b068-3a41-11e5-94eb-example
"""
UPDATE_STACK_RESPONSE_TEMPLATE = """
{{ stack.stack_id }}
b9b5b068-3a41-11e5-94eb-example
"""
DESCRIBE_STACKS_TEMPLATE = """
{% for stack in stacks %}
{{ stack.name }}
{{ stack.stack_id }}
2010-07-27T22:28:28Z
{{ stack.status }}
{% if stack.notification_arns %}
{% for notification_arn in stack.notification_arns %}
{{ notification_arn }}
{% endfor %}
{% else %}
{% endif %}
false
{% for output in stack.stack_outputs %}
{{ output.key }}
{{ output.value }}
{% endfor %}
{% for param_name, param_value in stack.stack_parameters.items() %}
{{ param_name }}
{{ param_value }}
{% endfor %}
{% if stack.role_arn %}
{{ stack.role_arn }}
{% endif %}
{% for tag_key, tag_value in stack.tags.items() %}
{{ tag_key }}
{{ tag_value }}
{% endfor %}
{% endfor %}
{% if next_token %}
{{ next_token }}
{% endif %}
"""
DESCRIBE_STACK_RESOURCE_RESPONSE_TEMPLATE = """
{{ stack.stack_id }}
{{ stack.name }}
{{ resource.logical_resource_id }}
{{ resource.physical_resource_id }}
{{ resource.type }}
2010-07-27T22:27:28Z
{{ stack.status }}
"""
DESCRIBE_STACK_RESOURCES_RESPONSE = """
{% for resource in stack.stack_resources %}
{{ stack.stack_id }}
{{ stack.name }}
{{ resource.logical_resource_id }}
{{ resource.physical_resource_id }}
{{ resource.type }}
2010-07-27T22:27:28Z
{{ stack.status }}
{% endfor %}
"""
DESCRIBE_STACK_EVENTS_RESPONSE = """
{% for event in stack.events[::-1] %}
{{ event.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ') }}
{{ event.resource_status }}
{{ event.stack_id }}
{{ event.event_id }}
{{ event.logical_resource_id }}
{% if event.resource_status_reason %}{{ event.resource_status_reason }}{% endif %}
{{ event.stack_name }}
{{ event.physical_resource_id }}
{% if event.resource_properties %}{{ event.resource_properties }}{% endif %}
{{ event.resource_type }}
{% endfor %}
b9b4b068-3a41-11e5-94eb-example
"""
LIST_STACKS_RESPONSE = """
{% for stack in stacks %}
{{ stack.stack_id }}
{{ stack.status }}
{{ stack.name }}
2011-05-23T15:47:44Z
{{ stack.description }}
{% endfor %}
"""
LIST_STACKS_RESOURCES_RESPONSE = """
{% for resource in resources %}
CREATE_COMPLETE
{{ resource.logical_resource_id }}
2011-06-21T20:15:58Z
{{ resource.physical_resource_id }}
{{ resource.type }}
{% endfor %}
2d06e36c-ac1d-11e0-a958-f9382b6eb86b
"""
GET_TEMPLATE_RESPONSE_TEMPLATE = """
{{ stack.template }}
b9b4b068-3a41-11e5-94eb-example
"""
UPDATE_STACK_RESPONSE_TEMPLATE = """
{{ stack.stack_id }}
b9b4b068-3a41-11e5-94eb-example
"""
DELETE_STACK_RESPONSE_TEMPLATE = """
5ccc7dcd-744c-11e5-be70-example
"""
LIST_EXPORTS_RESPONSE = """
{% for export in exports %}
{{ export.exporting_stack_id }}
{{ export.name }}
{{ export.value }}
{% endfor %}
{% if next_token %}
{{ next_token }}
{% endif %}
5ccc7dcd-744c-11e5-be70-example
"""