1#!/usr/bin/env python
2import unittest
3from datetime import datetime
4from mock import Mock
5
6from tests.unit import AWSMockServiceTestCase
7from boto.cloudformation.connection import CloudFormationConnection
8from boto.exception import BotoServerError
9from boto.compat import json
10
11SAMPLE_TEMPLATE = r"""
12{
13  "AWSTemplateFormatVersion" : "2010-09-09",
14  "Description" : "Sample template",
15  "Parameters" : {
16    "KeyName" : {
17      "Description" : "key pair",
18      "Type" : "String"
19    }
20  },
21  "Resources" : {
22    "Ec2Instance" : {
23      "Type" : "AWS::EC2::Instance",
24      "Properties" : {
25        "KeyName" : { "Ref" : "KeyName" },
26        "ImageId" : "ami-7f418316",
27        "UserData" : { "Fn::Base64" : "80" }
28      }
29    }
30  },
31  "Outputs" : {
32    "InstanceId" : {
33      "Description" : "InstanceId of the newly created EC2 instance",
34      "Value" : { "Ref" : "Ec2Instance" }
35    }
36}
37"""
38
39class CloudFormationConnectionBase(AWSMockServiceTestCase):
40    connection_class = CloudFormationConnection
41
42    def setUp(self):
43        super(CloudFormationConnectionBase, self).setUp()
44        self.stack_id = u'arn:aws:cloudformation:us-east-1:18:stack/Name/id'
45
46
47class TestCloudFormationCreateStack(CloudFormationConnectionBase):
48    def default_body(self):
49        return json.dumps(
50            {u'CreateStackResponse':
51                 {u'CreateStackResult': {u'StackId': self.stack_id},
52                  u'ResponseMetadata': {u'RequestId': u'1'}}}).encode('utf-8')
53
54    def test_create_stack_has_correct_request_params(self):
55        self.set_http_response(status_code=200)
56        api_response = self.service_connection.create_stack(
57            'stack_name', template_url='http://url',
58            template_body=SAMPLE_TEMPLATE,
59            parameters=[('KeyName', 'myKeyName')],
60            tags={'TagKey': 'TagValue'},
61            notification_arns=['arn:notify1', 'arn:notify2'],
62            disable_rollback=True,
63            timeout_in_minutes=20, capabilities=['CAPABILITY_IAM']
64        )
65        self.assertEqual(api_response, self.stack_id)
66        # These are the parameters that are actually sent to the CloudFormation
67        # service.
68        self.assert_request_parameters({
69            'Action': 'CreateStack',
70            'Capabilities.member.1': 'CAPABILITY_IAM',
71            'ContentType': 'JSON',
72            'DisableRollback': 'true',
73            'NotificationARNs.member.1': 'arn:notify1',
74            'NotificationARNs.member.2': 'arn:notify2',
75            'Parameters.member.1.ParameterKey': 'KeyName',
76            'Parameters.member.1.ParameterValue': 'myKeyName',
77            'Tags.member.1.Key': 'TagKey',
78            'Tags.member.1.Value': 'TagValue',
79            'StackName': 'stack_name',
80            'Version': '2010-05-15',
81            'TimeoutInMinutes': 20,
82            'TemplateBody': SAMPLE_TEMPLATE,
83            'TemplateURL': 'http://url',
84        })
85
86    # The test_create_stack_has_correct_request_params verified all of the
87    # params needed when making a create_stack service call.  The rest of the
88    # tests for create_stack only verify specific parts of the params sent
89    # to CloudFormation.
90
91    def test_create_stack_with_minimum_args(self):
92        # This will fail in practice, but the API docs only require stack_name.
93        self.set_http_response(status_code=200)
94        api_response = self.service_connection.create_stack('stack_name')
95        self.assertEqual(api_response, self.stack_id)
96        self.assert_request_parameters({
97            'Action': 'CreateStack',
98            'ContentType': 'JSON',
99            'DisableRollback': 'false',
100            'StackName': 'stack_name',
101            'Version': '2010-05-15',
102        })
103
104    def test_create_stack_fails(self):
105        self.set_http_response(status_code=400, reason='Bad Request',
106            body=b'{"Error": {"Code": 1, "Message": "Invalid arg."}}')
107        with self.assertRaisesRegexp(self.service_connection.ResponseError,
108            'Invalid arg.'):
109            api_response = self.service_connection.create_stack(
110                'stack_name', template_body=SAMPLE_TEMPLATE,
111                parameters=[('KeyName', 'myKeyName')])
112
113    def test_create_stack_fail_error(self):
114        self.set_http_response(status_code=400, reason='Bad Request',
115            body=b'{"RequestId": "abc", "Error": {"Code": 1, "Message": "Invalid arg."}}')
116        try:
117            api_response = self.service_connection.create_stack(
118                'stack_name', template_body=SAMPLE_TEMPLATE,
119                parameters=[('KeyName', 'myKeyName')])
120        except BotoServerError as e:
121            self.assertEqual('abc', e.request_id)
122            self.assertEqual(1, e.error_code)
123            self.assertEqual('Invalid arg.', e.message)
124
125class TestCloudFormationUpdateStack(CloudFormationConnectionBase):
126    def default_body(self):
127        return json.dumps(
128            {u'UpdateStackResponse':
129                 {u'UpdateStackResult': {u'StackId': self.stack_id},
130                  u'ResponseMetadata': {u'RequestId': u'1'}}}).encode('utf-8')
131
132    def test_update_stack_all_args(self):
133        self.set_http_response(status_code=200)
134        api_response = self.service_connection.update_stack(
135            'stack_name', template_url='http://url',
136            template_body=SAMPLE_TEMPLATE,
137            parameters=[('KeyName', 'myKeyName'), ('KeyName2', "", True),
138                        ('KeyName3', "", False), ('KeyName4', None, True),
139                        ('KeyName5', "Ignore Me", True)],
140            tags={'TagKey': 'TagValue'},
141            notification_arns=['arn:notify1', 'arn:notify2'],
142            disable_rollback=True,
143            timeout_in_minutes=20,
144            use_previous_template=True
145        )
146        self.assert_request_parameters({
147            'Action': 'UpdateStack',
148            'ContentType': 'JSON',
149            'DisableRollback': 'true',
150            'NotificationARNs.member.1': 'arn:notify1',
151            'NotificationARNs.member.2': 'arn:notify2',
152            'Parameters.member.1.ParameterKey': 'KeyName',
153            'Parameters.member.1.ParameterValue': 'myKeyName',
154            'Parameters.member.2.ParameterKey': 'KeyName2',
155            'Parameters.member.2.UsePreviousValue': 'true',
156            'Parameters.member.3.ParameterKey': 'KeyName3',
157            'Parameters.member.3.ParameterValue': '',
158            'Parameters.member.4.UsePreviousValue': 'true',
159            'Parameters.member.4.ParameterKey': 'KeyName4',
160            'Parameters.member.5.UsePreviousValue': 'true',
161            'Parameters.member.5.ParameterKey': 'KeyName5',
162            'Tags.member.1.Key': 'TagKey',
163            'Tags.member.1.Value': 'TagValue',
164            'StackName': 'stack_name',
165            'Version': '2010-05-15',
166            'TimeoutInMinutes': 20,
167            'TemplateBody': SAMPLE_TEMPLATE,
168            'TemplateURL': 'http://url',
169            'UsePreviousTemplate': 'true',
170        })
171
172    def test_update_stack_with_minimum_args(self):
173        self.set_http_response(status_code=200)
174        api_response = self.service_connection.update_stack('stack_name')
175        self.assertEqual(api_response, self.stack_id)
176        self.assert_request_parameters({
177            'Action': 'UpdateStack',
178            'ContentType': 'JSON',
179            'DisableRollback': 'false',
180            'StackName': 'stack_name',
181            'Version': '2010-05-15',
182        })
183
184    def test_update_stack_fails(self):
185        self.set_http_response(status_code=400, reason='Bad Request',
186                               body=b'Invalid arg.')
187        with self.assertRaises(self.service_connection.ResponseError):
188            api_response = self.service_connection.update_stack(
189                'stack_name', template_body=SAMPLE_TEMPLATE,
190                parameters=[('KeyName', 'myKeyName')])
191
192
193class TestCloudFormationDeleteStack(CloudFormationConnectionBase):
194    def default_body(self):
195        return json.dumps(
196            {u'DeleteStackResponse':
197                 {u'ResponseMetadata': {u'RequestId': u'1'}}}).encode('utf-8')
198
199    def test_delete_stack(self):
200        self.set_http_response(status_code=200)
201        api_response = self.service_connection.delete_stack('stack_name')
202        self.assertEqual(api_response, json.loads(self.default_body().decode('utf-8')))
203        self.assert_request_parameters({
204            'Action': 'DeleteStack',
205            'ContentType': 'JSON',
206            'StackName': 'stack_name',
207            'Version': '2010-05-15',
208        })
209
210    def test_delete_stack_fails(self):
211        self.set_http_response(status_code=400)
212        with self.assertRaises(self.service_connection.ResponseError):
213            api_response = self.service_connection.delete_stack('stack_name')
214
215
216class TestCloudFormationDescribeStackResource(CloudFormationConnectionBase):
217    def default_body(self):
218        return json.dumps('fake server response').encode('utf-8')
219
220    def test_describe_stack_resource(self):
221        self.set_http_response(status_code=200)
222        api_response = self.service_connection.describe_stack_resource(
223            'stack_name', 'resource_id')
224        self.assertEqual(api_response, 'fake server response')
225        self.assert_request_parameters({
226            'Action': 'DescribeStackResource',
227            'ContentType': 'JSON',
228            'LogicalResourceId': 'resource_id',
229            'StackName': 'stack_name',
230            'Version': '2010-05-15',
231        })
232
233    def test_describe_stack_resource_fails(self):
234        self.set_http_response(status_code=400)
235        with self.assertRaises(self.service_connection.ResponseError):
236            api_response = self.service_connection.describe_stack_resource(
237                'stack_name', 'resource_id')
238
239
240class TestCloudFormationGetTemplate(CloudFormationConnectionBase):
241    def default_body(self):
242        return json.dumps('fake server response').encode('utf-8')
243
244    def test_get_template(self):
245        self.set_http_response(status_code=200)
246        api_response = self.service_connection.get_template('stack_name')
247        self.assertEqual(api_response, 'fake server response')
248        self.assert_request_parameters({
249            'Action': 'GetTemplate',
250            'ContentType': 'JSON',
251            'StackName': 'stack_name',
252            'Version': '2010-05-15',
253        })
254
255
256    def test_get_template_fails(self):
257        self.set_http_response(status_code=400)
258        with self.assertRaises(self.service_connection.ResponseError):
259            api_response = self.service_connection.get_template('stack_name')
260
261
262class TestCloudFormationGetStackevents(CloudFormationConnectionBase):
263    def default_body(self):
264        return b"""
265            <DescribeStackEventsResult>
266              <StackEvents>
267                <member>
268                  <EventId>Event-1-Id</EventId>
269                  <StackId>arn:aws:cfn:us-east-1:1:stack</StackId>
270                  <StackName>MyStack</StackName>
271                  <LogicalResourceId>MyStack</LogicalResourceId>
272                  <PhysicalResourceId>MyStack_One</PhysicalResourceId>
273                  <ResourceType>AWS::CloudFormation::Stack</ResourceType>
274                  <Timestamp>2010-07-27T22:26:28Z</Timestamp>
275                  <ResourceStatus>CREATE_IN_PROGRESS</ResourceStatus>
276                  <ResourceStatusReason>User initiated</ResourceStatusReason>
277                </member>
278                <member>
279                  <EventId>Event-2-Id</EventId>
280                  <StackId>arn:aws:cfn:us-east-1:1:stack</StackId>
281                  <StackName>MyStack</StackName>
282                  <LogicalResourceId>MySG1</LogicalResourceId>
283                  <PhysicalResourceId>MyStack_SG1</PhysicalResourceId>
284                  <ResourceType>AWS::SecurityGroup</ResourceType>
285                  <Timestamp>2010-07-27T22:28:28Z</Timestamp>
286                  <ResourceStatus>CREATE_COMPLETE</ResourceStatus>
287                </member>
288              </StackEvents>
289            </DescribeStackEventsResult>
290        """
291
292    def test_describe_stack_events(self):
293        self.set_http_response(status_code=200)
294        first, second = self.service_connection.describe_stack_events('stack_name', next_token='next_token')
295        self.assertEqual(first.event_id, 'Event-1-Id')
296        self.assertEqual(first.logical_resource_id, 'MyStack')
297        self.assertEqual(first.physical_resource_id, 'MyStack_One')
298        self.assertEqual(first.resource_properties, None)
299        self.assertEqual(first.resource_status, 'CREATE_IN_PROGRESS')
300        self.assertEqual(first.resource_status_reason, 'User initiated')
301        self.assertEqual(first.resource_type, 'AWS::CloudFormation::Stack')
302        self.assertEqual(first.stack_id, 'arn:aws:cfn:us-east-1:1:stack')
303        self.assertEqual(first.stack_name, 'MyStack')
304        self.assertIsNotNone(first.timestamp)
305
306        self.assertEqual(second.event_id, 'Event-2-Id')
307        self.assertEqual(second.logical_resource_id, 'MySG1')
308        self.assertEqual(second.physical_resource_id, 'MyStack_SG1')
309        self.assertEqual(second.resource_properties, None)
310        self.assertEqual(second.resource_status, 'CREATE_COMPLETE')
311        self.assertEqual(second.resource_status_reason, None)
312        self.assertEqual(second.resource_type, 'AWS::SecurityGroup')
313        self.assertEqual(second.stack_id, 'arn:aws:cfn:us-east-1:1:stack')
314        self.assertEqual(second.stack_name, 'MyStack')
315        self.assertIsNotNone(second.timestamp)
316
317        self.assert_request_parameters({
318            'Action': 'DescribeStackEvents',
319            'NextToken': 'next_token',
320            'StackName': 'stack_name',
321            'Version': '2010-05-15',
322        })
323
324
325class TestCloudFormationDescribeStackResources(CloudFormationConnectionBase):
326    def default_body(self):
327        return b"""
328            <DescribeStackResourcesResult>
329              <StackResources>
330                <member>
331                  <StackId>arn:aws:cfn:us-east-1:1:stack</StackId>
332                  <StackName>MyStack</StackName>
333                  <LogicalResourceId>MyDBInstance</LogicalResourceId>
334                  <PhysicalResourceId>MyStack_DB1</PhysicalResourceId>
335                  <ResourceType>AWS::DBInstance</ResourceType>
336                  <Timestamp>2010-07-27T22:27:28Z</Timestamp>
337                  <ResourceStatus>CREATE_COMPLETE</ResourceStatus>
338                </member>
339                <member>
340                  <StackId>arn:aws:cfn:us-east-1:1:stack</StackId>
341                  <StackName>MyStack</StackName>
342                  <LogicalResourceId>MyAutoScalingGroup</LogicalResourceId>
343                  <PhysicalResourceId>MyStack_ASG1</PhysicalResourceId>
344                  <ResourceType>AWS::AutoScalingGroup</ResourceType>
345                  <Timestamp>2010-07-27T22:28:28Z</Timestamp>
346                  <ResourceStatus>CREATE_IN_PROGRESS</ResourceStatus>
347                </member>
348              </StackResources>
349            </DescribeStackResourcesResult>
350        """
351
352    def test_describe_stack_resources(self):
353        self.set_http_response(status_code=200)
354        first, second = self.service_connection.describe_stack_resources(
355            'stack_name', 'logical_resource_id', 'physical_resource_id')
356        self.assertEqual(first.description, None)
357        self.assertEqual(first.logical_resource_id, 'MyDBInstance')
358        self.assertEqual(first.physical_resource_id, 'MyStack_DB1')
359        self.assertEqual(first.resource_status, 'CREATE_COMPLETE')
360        self.assertEqual(first.resource_status_reason, None)
361        self.assertEqual(first.resource_type, 'AWS::DBInstance')
362        self.assertEqual(first.stack_id, 'arn:aws:cfn:us-east-1:1:stack')
363        self.assertEqual(first.stack_name, 'MyStack')
364        self.assertIsNotNone(first.timestamp)
365
366        self.assertEqual(second.description, None)
367        self.assertEqual(second.logical_resource_id, 'MyAutoScalingGroup')
368        self.assertEqual(second.physical_resource_id, 'MyStack_ASG1')
369        self.assertEqual(second.resource_status, 'CREATE_IN_PROGRESS')
370        self.assertEqual(second.resource_status_reason, None)
371        self.assertEqual(second.resource_type, 'AWS::AutoScalingGroup')
372        self.assertEqual(second.stack_id, 'arn:aws:cfn:us-east-1:1:stack')
373        self.assertEqual(second.stack_name, 'MyStack')
374        self.assertIsNotNone(second.timestamp)
375
376        self.assert_request_parameters({
377            'Action': 'DescribeStackResources',
378            'LogicalResourceId': 'logical_resource_id',
379            'PhysicalResourceId': 'physical_resource_id',
380            'StackName': 'stack_name',
381            'Version': '2010-05-15',
382        })
383
384
385class TestCloudFormationDescribeStacks(CloudFormationConnectionBase):
386    def default_body(self):
387        return b"""
388          <DescribeStacksResponse>
389            <DescribeStacksResult>
390              <Stacks>
391                <member>
392                  <StackId>arn:aws:cfn:us-east-1:1:stack</StackId>
393                  <StackStatus>CREATE_COMPLETE</StackStatus>
394                  <StackStatusReason>REASON</StackStatusReason>
395                  <StackName>MyStack</StackName>
396                  <Description>My Description</Description>
397                  <CreationTime>2012-05-16T22:55:31Z</CreationTime>
398                  <Capabilities>
399                    <member>CAPABILITY_IAM</member>
400                  </Capabilities>
401                  <NotificationARNs>
402                    <member>arn:aws:sns:region-name:account-name:topic-name</member>
403                  </NotificationARNs>
404                  <DisableRollback>false</DisableRollback>
405                  <Parameters>
406                    <member>
407                      <ParameterValue>MyValue</ParameterValue>
408                      <ParameterKey>MyKey</ParameterKey>
409                    </member>
410                  </Parameters>
411                  <Outputs>
412                    <member>
413                      <OutputValue>http://url/</OutputValue>
414                      <Description>Server URL</Description>
415                      <OutputKey>ServerURL</OutputKey>
416                    </member>
417                  </Outputs>
418                  <Tags>
419                    <member>
420                      <Key>MyTagKey</Key>
421                      <Value>MyTagValue</Value>
422                    </member>
423                  </Tags>
424                </member>
425              </Stacks>
426            </DescribeStacksResult>
427            <ResponseMetadata>
428              <RequestId>12345</RequestId>
429            </ResponseMetadata>
430        </DescribeStacksResponse>
431        """
432
433    def test_describe_stacks(self):
434        self.set_http_response(status_code=200)
435
436        stacks = self.service_connection.describe_stacks('MyStack')
437        self.assertEqual(len(stacks), 1)
438
439        stack = stacks[0]
440        self.assertEqual(stack.creation_time,
441                         datetime(2012, 5, 16, 22, 55, 31))
442        self.assertEqual(stack.description, 'My Description')
443        self.assertEqual(stack.disable_rollback, False)
444        self.assertEqual(stack.stack_id, 'arn:aws:cfn:us-east-1:1:stack')
445        self.assertEqual(stack.stack_status, 'CREATE_COMPLETE')
446        self.assertEqual(stack.stack_name, 'MyStack')
447        self.assertEqual(stack.stack_name_reason, 'REASON')
448        self.assertEqual(stack.stack_status_reason, 'REASON')
449        self.assertEqual(stack.timeout_in_minutes, None)
450
451        self.assertEqual(len(stack.outputs), 1)
452        self.assertEqual(stack.outputs[0].description, 'Server URL')
453        self.assertEqual(stack.outputs[0].key, 'ServerURL')
454        self.assertEqual(stack.outputs[0].value, 'http://url/')
455
456        self.assertEqual(len(stack.parameters), 1)
457        self.assertEqual(stack.parameters[0].key, 'MyKey')
458        self.assertEqual(stack.parameters[0].value, 'MyValue')
459
460        self.assertEqual(len(stack.capabilities), 1)
461        self.assertEqual(stack.capabilities[0].value, 'CAPABILITY_IAM')
462
463        self.assertEqual(len(stack.notification_arns), 1)
464        self.assertEqual(stack.notification_arns[0].value, 'arn:aws:sns:region-name:account-name:topic-name')
465
466        self.assertEqual(len(stack.tags), 1)
467        self.assertEqual(stack.tags['MyTagKey'], 'MyTagValue')
468
469        self.assert_request_parameters({
470            'Action': 'DescribeStacks',
471            'StackName': 'MyStack',
472            'Version': '2010-05-15',
473        })
474
475
476class TestCloudFormationListStackResources(CloudFormationConnectionBase):
477    def default_body(self):
478        return b"""
479            <ListStackResourcesResponse>
480              <ListStackResourcesResult>
481                <StackResourceSummaries>
482                  <member>
483                    <ResourceStatus>CREATE_COMPLETE</ResourceStatus>
484                    <LogicalResourceId>SampleDB</LogicalResourceId>
485                    <LastUpdatedTime>2011-06-21T20:25:57Z</LastUpdatedTime>
486                    <PhysicalResourceId>My-db-ycx</PhysicalResourceId>
487                    <ResourceType>AWS::RDS::DBInstance</ResourceType>
488                  </member>
489                  <member>
490                    <ResourceStatus>CREATE_COMPLETE</ResourceStatus>
491                    <LogicalResourceId>CPUAlarmHigh</LogicalResourceId>
492                    <LastUpdatedTime>2011-06-21T20:29:23Z</LastUpdatedTime>
493                    <PhysicalResourceId>MyStack-CPUH-PF</PhysicalResourceId>
494                    <ResourceType>AWS::CloudWatch::Alarm</ResourceType>
495                  </member>
496                </StackResourceSummaries>
497              </ListStackResourcesResult>
498              <ResponseMetadata>
499                <RequestId>2d06e36c-ac1d-11e0-a958-f9382b6eb86b</RequestId>
500              </ResponseMetadata>
501            </ListStackResourcesResponse>
502        """
503
504    def test_list_stack_resources(self):
505        self.set_http_response(status_code=200)
506        resources = self.service_connection.list_stack_resources('MyStack',
507                                                              next_token='next_token')
508        self.assertEqual(len(resources), 2)
509        self.assertEqual(resources[0].last_updated_time,
510                         datetime(2011, 6, 21, 20, 25, 57))
511        self.assertEqual(resources[0].logical_resource_id, 'SampleDB')
512        self.assertEqual(resources[0].physical_resource_id, 'My-db-ycx')
513        self.assertEqual(resources[0].resource_status, 'CREATE_COMPLETE')
514        self.assertEqual(resources[0].resource_status_reason, None)
515        self.assertEqual(resources[0].resource_type, 'AWS::RDS::DBInstance')
516
517        self.assertEqual(resources[1].last_updated_time,
518                         datetime(2011, 6, 21, 20, 29, 23))
519        self.assertEqual(resources[1].logical_resource_id, 'CPUAlarmHigh')
520        self.assertEqual(resources[1].physical_resource_id, 'MyStack-CPUH-PF')
521        self.assertEqual(resources[1].resource_status, 'CREATE_COMPLETE')
522        self.assertEqual(resources[1].resource_status_reason, None)
523        self.assertEqual(resources[1].resource_type, 'AWS::CloudWatch::Alarm')
524
525        self.assert_request_parameters({
526            'Action': 'ListStackResources',
527            'NextToken': 'next_token',
528            'StackName': 'MyStack',
529            'Version': '2010-05-15',
530        })
531
532
533class TestCloudFormationListStacks(CloudFormationConnectionBase):
534    def default_body(self):
535        return b"""
536            <ListStacksResponse>
537             <ListStacksResult>
538              <StackSummaries>
539                <member>
540                    <StackId>arn:aws:cfn:us-east-1:1:stack/Test1/aa</StackId>
541                    <StackStatus>CREATE_IN_PROGRESS</StackStatus>
542                    <StackName>vpc1</StackName>
543                    <CreationTime>2011-05-23T15:47:44Z</CreationTime>
544                    <TemplateDescription>My Description.</TemplateDescription>
545                </member>
546              </StackSummaries>
547             </ListStacksResult>
548            </ListStacksResponse>
549        """
550
551    def test_list_stacks(self):
552        self.set_http_response(status_code=200)
553        stacks = self.service_connection.list_stacks(['CREATE_IN_PROGRESS'],
554                                                  next_token='next_token')
555        self.assertEqual(len(stacks), 1)
556        self.assertEqual(stacks[0].stack_id,
557                         'arn:aws:cfn:us-east-1:1:stack/Test1/aa')
558        self.assertEqual(stacks[0].stack_status, 'CREATE_IN_PROGRESS')
559        self.assertEqual(stacks[0].stack_name, 'vpc1')
560        self.assertEqual(stacks[0].creation_time,
561                         datetime(2011, 5, 23, 15, 47, 44))
562        self.assertEqual(stacks[0].deletion_time, None)
563        self.assertEqual(stacks[0].template_description, 'My Description.')
564
565        self.assert_request_parameters({
566            'Action': 'ListStacks',
567            'NextToken': 'next_token',
568            'StackStatusFilter.member.1': 'CREATE_IN_PROGRESS',
569            'Version': '2010-05-15',
570        })
571
572
573class TestCloudFormationValidateTemplate(CloudFormationConnectionBase):
574    def default_body(self):
575        return b"""
576            <ValidateTemplateResponse xmlns="http://cloudformation.amazonaws.com/doc/2010-05-15/">
577              <ValidateTemplateResult>
578                <Description>My Description.</Description>
579                <Parameters>
580                  <member>
581                    <NoEcho>false</NoEcho>
582                    <ParameterKey>InstanceType</ParameterKey>
583                    <Description>Type of instance to launch</Description>
584                    <DefaultValue>m1.small</DefaultValue>
585                  </member>
586                  <member>
587                    <NoEcho>false</NoEcho>
588                    <ParameterKey>KeyName</ParameterKey>
589                    <Description>EC2 KeyPair</Description>
590                  </member>
591                </Parameters>
592                <CapabilitiesReason>Reason</CapabilitiesReason>
593                <Capabilities>
594                  <member>CAPABILITY_IAM</member>
595                </Capabilities>
596              </ValidateTemplateResult>
597              <ResponseMetadata>
598                <RequestId>0be7b6e8-e4a0-11e0-a5bd-9f8d5a7dbc91</RequestId>
599              </ResponseMetadata>
600            </ValidateTemplateResponse>
601        """
602
603    def test_validate_template(self):
604        self.set_http_response(status_code=200)
605        template = self.service_connection.validate_template(template_body=SAMPLE_TEMPLATE,
606                                                          template_url='http://url')
607        self.assertEqual(template.description, 'My Description.')
608        self.assertEqual(len(template.template_parameters), 2)
609        param1, param2 = template.template_parameters
610        self.assertEqual(param1.default_value, 'm1.small')
611        self.assertEqual(param1.description, 'Type of instance to launch')
612        self.assertEqual(param1.no_echo, True)
613        self.assertEqual(param1.parameter_key, 'InstanceType')
614
615        self.assertEqual(param2.default_value, None)
616        self.assertEqual(param2.description, 'EC2 KeyPair')
617        self.assertEqual(param2.no_echo, True)
618        self.assertEqual(param2.parameter_key, 'KeyName')
619
620        self.assertEqual(template.capabilities_reason, 'Reason')
621
622        self.assertEqual(len(template.capabilities), 1)
623        self.assertEqual(template.capabilities[0].value, 'CAPABILITY_IAM')
624
625        self.assert_request_parameters({
626            'Action': 'ValidateTemplate',
627            'TemplateBody': SAMPLE_TEMPLATE,
628            'TemplateURL': 'http://url',
629            'Version': '2010-05-15',
630        })
631
632
633class TestCloudFormationCancelUpdateStack(CloudFormationConnectionBase):
634    def default_body(self):
635        return b"""<CancelUpdateStackResult/>"""
636
637    def test_cancel_update_stack(self):
638        self.set_http_response(status_code=200)
639        api_response = self.service_connection.cancel_update_stack('stack_name')
640        self.assertEqual(api_response, True)
641        self.assert_request_parameters({
642            'Action': 'CancelUpdateStack',
643            'StackName': 'stack_name',
644            'Version': '2010-05-15',
645        })
646
647
648class TestCloudFormationEstimateTemplateCost(CloudFormationConnectionBase):
649    def default_body(self):
650        return b"""
651            {
652                "EstimateTemplateCostResponse": {
653                    "EstimateTemplateCostResult": {
654                        "Url": "http://calculator.s3.amazonaws.com/calc5.html?key=cf-2e351785-e821-450c-9d58-625e1e1ebfb6"
655                    }
656                }
657            }
658        """
659
660    def test_estimate_template_cost(self):
661        self.set_http_response(status_code=200)
662        api_response = self.service_connection.estimate_template_cost(
663            template_body='{}')
664        self.assertEqual(api_response,
665            'http://calculator.s3.amazonaws.com/calc5.html?key=cf-2e351785-e821-450c-9d58-625e1e1ebfb6')
666        self.assert_request_parameters({
667            'Action': 'EstimateTemplateCost',
668            'ContentType': 'JSON',
669            'TemplateBody': '{}',
670            'Version': '2010-05-15',
671        })
672
673
674class TestCloudFormationGetStackPolicy(CloudFormationConnectionBase):
675    def default_body(self):
676        return b"""
677            {
678                "GetStackPolicyResponse": {
679                    "GetStackPolicyResult": {
680                        "StackPolicyBody": "{...}"
681                    }
682                }
683            }
684        """
685
686    def test_get_stack_policy(self):
687        self.set_http_response(status_code=200)
688        api_response = self.service_connection.get_stack_policy('stack-id')
689        self.assertEqual(api_response, '{...}')
690        self.assert_request_parameters({
691            'Action': 'GetStackPolicy',
692            'ContentType': 'JSON',
693            'StackName': 'stack-id',
694            'Version': '2010-05-15',
695        })
696
697
698class TestCloudFormationSetStackPolicy(CloudFormationConnectionBase):
699    def default_body(self):
700        return b"""
701            {
702                "SetStackPolicyResponse": {
703                    "SetStackPolicyResult": {
704                        "Some": "content"
705                    }
706                }
707            }
708        """
709
710    def test_set_stack_policy(self):
711        self.set_http_response(status_code=200)
712        api_response = self.service_connection.set_stack_policy('stack-id',
713            stack_policy_body='{}')
714        self.assertDictEqual(api_response, {'SetStackPolicyResult': {'Some': 'content'}})
715        self.assert_request_parameters({
716            'Action': 'SetStackPolicy',
717            'ContentType': 'JSON',
718            'StackName': 'stack-id',
719            'StackPolicyBody': '{}',
720            'Version': '2010-05-15',
721        })
722
723
724if __name__ == '__main__':
725    unittest.main()
726