1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Licensed under the Apache License, Version 2.0 (the "License"); you may
4
# not use this file except in compliance with the License. You may obtain
5
# a copy of the License at
7
# http://www.apache.org/licenses/LICENSE-2.0
9
# Unless required by applicable law or agreed to in writing, software
10
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
# License for the specific language governing permissions and limitations
19
from nose.plugins.attrib import attr
23
from heat.common import template_format
27
@attr(tag=['func', 'wordpress', 'api', 'cfn', 'F17'])
28
class CfnApiFunctionalTest(unittest.TestCase):
30
This test launches a wordpress stack then attempts to verify
31
correct operation of all actions supported by the heat CFN API
33
Note we use class-level fixtures to avoid setting up a new stack
34
for every test method, we set up the stack once then do all the
35
tests, this means all tests methods are performed on one class
36
instance, instead of creating a new class for every method, which
37
is the normal nose unittest.TestCase behavior.
39
The nose docs are a bit vague on how to do this, but it seems that
40
(setup|teardown)All works and they have to be classmethods.
42
Contrary to the nose docs, the class can be a unittest.TestCase subclass
47
template = 'WordPress_Single_Instance.template'
49
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
51
'DBPassword=' + os.environ['OS_PASSWORD']])
53
cls.logical_resource_name = 'WikiDatabase'
54
cls.logical_resource_type = 'AWS::EC2::Instance'
56
# Just to get the assert*() methods
57
class CfnApiFunctions(unittest.TestCase):
58
@unittest.skip('Not a real test case')
62
inst = CfnApiFunctions()
63
cls.stack = util.Stack(inst, template, 'F17', 'x86_64', 'cfntools',
65
cls.WikiDatabase = util.Instance(inst, cls.logical_resource_name)
69
cls.WikiDatabase.wait_for_boot()
70
cls.WikiDatabase.check_cfntools()
71
cls.WikiDatabase.wait_for_provisioning()
73
cls.logical_resource_status = "CREATE_COMPLETE"
75
# Save some compiled regexes and strings for response validation
76
cls.time_re = re.compile(
77
"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$")
78
cls.description_re = re.compile(
79
"^AWS CloudFormation Sample Template")
80
cls.stack_status = "CREATE_COMPLETE"
81
cls.stack_status_reason = "Stack successfully created"
82
cls.stack_timeout = str(60)
83
cls.stack_disable_rollback = "True"
85
# Match the expected format for an instance's physical resource ID
86
cls.phys_res_id_re = re.compile(
87
"^[0-9a-z]*-[0-9a-z]*-[0-9a-z]*-[0-9a-z]*-[0-9a-z]*$")
88
except Exception as ex:
89
print "setupAll failed : %s" % ex
98
def test_instance(self):
99
# ensure wordpress was installed by checking for expected
100
# configuration file over ssh
101
# This is the same as the standard wordress template test
102
# but we still do it to prove the stack is OK
103
self.assertTrue(self.WikiDatabase.file_present
104
('/etc/wordpress/wp-config.php'))
105
print "Wordpress installation detected"
107
# Verify the output URL parses as expected, ie check that
108
# the wordpress installation is operational
109
stack_url = self.stack.get_stack_output("WebsiteURL")
110
print "Got stack output WebsiteURL=%s, verifying" % stack_url
111
ver = verify.VerifyStack()
112
self.assertTrue(ver.verify_wordpress(stack_url))
114
def testListStacks(self):
115
response = self.stack.heatclient.list_stacks()
116
prefix = '/ListStacksResponse/ListStacksResult/StackSummaries/member'
118
stack_id = self.stack.response_xml_item(response, prefix, "StackId")
119
self.stack.check_stackid(stack_id)
121
update_time = self.stack.response_xml_item(response, prefix,
123
self.assertTrue(self.time_re.match(update_time) is not None)
125
create_time = self.stack.response_xml_item(response, prefix,
127
self.assertTrue(self.time_re.match(create_time) is not None)
129
description = self.stack.response_xml_item(response, prefix,
130
"TemplateDescription")
131
self.assertTrue(self.description_re.match(description) is not None)
133
status_reason = self.stack.response_xml_item(response, prefix,
135
self.assertEqual(status_reason, self.stack_status_reason)
137
stack_name = self.stack.response_xml_item(response, prefix,
139
self.assertEqual(stack_name, self.stack.stackname)
141
stack_status = self.stack.response_xml_item(response, prefix,
143
self.assertEqual(stack_status, self.stack_status)
145
print "ListStacks : OK"
147
def testDescribeStacks(self):
149
parameters['StackName'] = self.stack.stackname
150
response = self.stack.heatclient.describe_stacks(**parameters)
151
prefix = '/DescribeStacksResponse/DescribeStacksResult/Stacks/member'
153
stack_id = self.stack.response_xml_item(response, prefix, "StackId")
154
self.stack.check_stackid(stack_id)
156
update_time = self.stack.response_xml_item(response, prefix,
158
self.assertTrue(self.time_re.match(update_time) is not None)
160
create_time = self.stack.response_xml_item(response, prefix,
162
self.assertTrue(self.time_re.match(create_time) is not None)
164
description = self.stack.response_xml_item(response, prefix,
166
self.assertTrue(self.description_re.match(description) is not None)
168
status_reason = self.stack.response_xml_item(response, prefix,
170
self.assertEqual(status_reason, self.stack_status_reason)
172
stack_name = self.stack.response_xml_item(response, prefix,
174
self.assertEqual(stack_name, self.stack.stackname)
176
stack_status = self.stack.response_xml_item(response, prefix,
178
self.assertEqual(stack_status, self.stack_status)
180
stack_timeout = self.stack.response_xml_item(response, prefix,
182
self.assertEqual(stack_timeout, self.stack_timeout)
184
disable_rollback = self.stack.response_xml_item(response, prefix,
186
self.assertEqual(disable_rollback, self.stack_disable_rollback)
188
# Create a dict to lookup the expected template parameters
189
# NoEcho parameters are masked with 6 asterisks
190
template_parameters = {'DBUsername': '******',
191
'LinuxDistribution': 'F17',
192
'InstanceType': 'm1.xlarge',
193
'DBRootPassword': '******',
194
'KeyName': self.stack.keyname,
195
'DBPassword': '******',
196
'DBName': 'wordpress'}
198
# We do a fully qualified xpath lookup to extract the paramter
199
# value for each key, then check the extracted value
200
param_prefix = prefix + "/Parameters/member"
201
for key, value in template_parameters.iteritems():
202
lookup = '[ParameterKey="' + key + '" and ParameterValue="' +\
204
lookup_value = self.stack.response_xml_item(response,
205
param_prefix + lookup,
207
self.assertEqual(lookup_value, value)
209
# Then to a similar lookup to verify the Outputs section
210
expected_url = "http://" + self.WikiDatabase.ip + "/wordpress"
212
outputs_prefix = prefix + "/Outputs/member"
213
lookup = '[OutputKey="WebsiteURL" and OutputValue="' + expected_url +\
214
'" and Description="URL for Wordpress wiki"]'
215
lookup_value = self.stack.response_xml_item(response,
216
outputs_prefix + lookup,
218
self.assertEqual(lookup_value, expected_url)
220
print "DescribeStacks : OK"
222
def testDescribeStackEvents(self):
224
parameters['StackName'] = self.stack.stackname
225
response = self.stack.heatclient.list_stack_events(**parameters)
226
prefix = '/DescribeStackEventsResponse/DescribeStackEventsResult/' +\
227
'StackEvents/member[LogicalResourceId="' +\
228
self.logical_resource_name + '" and ResourceStatus="' +\
229
self.logical_resource_status + '"]'
231
stack_id = self.stack.response_xml_item(response, prefix, "StackId")
232
self.stack.check_stackid(stack_id)
234
event_id = self.stack.response_xml_item(response, prefix, "EventId")
235
self.assertTrue(re.match("[0-9]*$", event_id) is not None)
237
resource_status = self.stack.response_xml_item(response, prefix,
239
self.assertEqual(resource_status, self.logical_resource_status)
241
resource_type = self.stack.response_xml_item(response, prefix,
243
self.assertEqual(resource_type, self.logical_resource_type)
245
update_time = self.stack.response_xml_item(response, prefix,
247
self.assertTrue(self.time_re.match(update_time) is not None)
249
status_data = self.stack.response_xml_item(response, prefix,
250
"ResourceStatusReason")
251
self.assertEqual(status_data, "state changed")
253
stack_name = self.stack.response_xml_item(response, prefix,
255
self.assertEqual(stack_name, self.stack.stackname)
257
log_res_id = self.stack.response_xml_item(response, prefix,
259
self.assertEqual(log_res_id, self.logical_resource_name)
261
phys_res_id = self.stack.response_xml_item(response, prefix,
262
"PhysicalResourceId")
263
self.assertTrue(self.phys_res_id_re.match(phys_res_id) is not None)
265
# ResourceProperties format is defined as a string "blob" by AWS
266
# we return JSON encoded properies, so decode and check one key
267
prop_json = self.stack.response_xml_item(response, prefix,
268
"ResourceProperties")
269
self.assertTrue(prop_json is not None)
271
prop = json.loads(prop_json)
272
self.assertEqual(prop["InstanceType"], "m1.xlarge")
274
print "DescribeStackEvents : OK"
276
def testGetTemplate(self):
278
parameters['StackName'] = self.stack.stackname
279
response = self.stack.heatclient.get_template(**parameters)
280
prefix = '/GetTemplateResponse/GetTemplateResult'
282
# Extract the JSON TemplateBody and prove it parses
283
template = self.stack.response_xml_item(response, prefix,
285
json_load = template_format.parse(template)
286
self.assertTrue(json_load is not None)
288
# Then sanity check content - I guess we could diff
289
# with the template file but for now just check the
290
# description looks sane..
291
description = json_load['Description']
292
self.assertTrue(self.description_re.match(description) is not None)
294
print "GetTemplate : OK"
296
def testDescribeStackResource(self):
297
parameters = {'StackName': self.stack.stackname,
298
'LogicalResourceId': self.logical_resource_name}
299
response = self.stack.heatclient.describe_stack_resource(**parameters)
300
prefix = '/DescribeStackResourceResponse/DescribeStackResourceResult'\
301
+ '/StackResourceDetail'
303
stack_id = self.stack.response_xml_item(response, prefix, "StackId")
304
self.stack.check_stackid(stack_id)
306
resource_status = self.stack.response_xml_item(response, prefix,
308
self.assertEqual(resource_status, self.logical_resource_status)
310
resource_type = self.stack.response_xml_item(response, prefix,
312
self.assertEqual(resource_type, self.logical_resource_type)
314
update_time = self.stack.response_xml_item(response, prefix,
315
"LastUpdatedTimestamp")
316
self.assertTrue(self.time_re.match(update_time) is not None)
318
status_reason = self.stack.response_xml_item(response, prefix,
319
"ResourceStatusReason")
320
self.assertEqual(status_reason, "state changed")
322
stack_name = self.stack.response_xml_item(response, prefix,
324
self.assertEqual(stack_name, self.stack.stackname)
326
log_res_id = self.stack.response_xml_item(response, prefix,
328
self.assertEqual(log_res_id, self.logical_resource_name)
330
phys_res_id = self.stack.response_xml_item(response, prefix,
331
"PhysicalResourceId")
332
self.assertTrue(self.phys_res_id_re.match(phys_res_id) is not None)
334
metadata = self.stack.response_xml_item(response, prefix, "Metadata")
335
json_load = json.loads(metadata)
336
self.assertTrue(json_load is not None)
337
self.assertTrue("AWS::CloudFormation::Init" in json_load)
339
print "DescribeStackResource : OK"
341
def testDescribeStackResources(self):
342
parameters = {'NameOrPid': self.stack.stackname,
343
'LogicalResourceId': self.logical_resource_name}
344
response = self.stack.heatclient.describe_stack_resources(**parameters)
345
prefix = '/DescribeStackResourcesResponse/' +\
346
'DescribeStackResourcesResult/StackResources/member'
348
stack_id = self.stack.response_xml_item(response, prefix, "StackId")
349
self.stack.check_stackid(stack_id)
351
resource_status = self.stack.response_xml_item(response, prefix,
353
self.assertEqual(resource_status, self.logical_resource_status)
355
resource_type = self.stack.response_xml_item(response, prefix,
357
self.assertEqual(resource_type, self.logical_resource_type)
359
update_time = self.stack.response_xml_item(response, prefix,
361
self.assertTrue(self.time_re.match(update_time) is not None)
363
status_reason = self.stack.response_xml_item(response, prefix,
364
"ResourceStatusReason")
365
self.assertEqual(status_reason, "state changed")
367
stack_name = self.stack.response_xml_item(response, prefix,
369
self.assertEqual(stack_name, self.stack.stackname)
371
log_res_id = self.stack.response_xml_item(response, prefix,
373
self.assertEqual(log_res_id, self.logical_resource_name)
375
phys_res_id = self.stack.response_xml_item(response, prefix,
376
"PhysicalResourceId")
377
self.assertTrue(self.phys_res_id_re.match(phys_res_id) is not None)
379
print "DescribeStackResources : OK"
381
def testListStackResources(self):
383
parameters['StackName'] = self.stack.stackname
384
response = self.stack.heatclient.list_stack_resources(**parameters)
385
prefix = '/ListStackResourcesResponse/ListStackResourcesResult' +\
386
'/StackResourceSummaries/member'
388
resource_status = self.stack.response_xml_item(response, prefix,
390
self.assertEqual(resource_status, self.logical_resource_status)
392
status_reason = self.stack.response_xml_item(response, prefix,
393
"ResourceStatusReason")
394
self.assertEqual(status_reason, "state changed")
396
update_time = self.stack.response_xml_item(response, prefix,
397
"LastUpdatedTimestamp")
398
self.assertTrue(self.time_re.match(update_time) is not None)
400
resource_type = self.stack.response_xml_item(response, prefix,
402
self.assertEqual(resource_type, self.logical_resource_type)
404
log_res_id = self.stack.response_xml_item(response, prefix,
406
self.assertEqual(log_res_id, self.logical_resource_name)
408
phys_res_id = self.stack.response_xml_item(response, prefix,
409
"PhysicalResourceId")
410
self.assertTrue(self.phys_res_id_re.match(phys_res_id) is not None)
412
print "ListStackResources : OK"
414
def testValidateTemplate(self):
415
# Use stack.format_parameters to get the TemplateBody
416
params = self.stack.format_parameters()
417
val_params = {'TemplateBody': params['TemplateBody']}
418
response = self.stack.heatclient.validate_template(**val_params)
419
prefix = '/ValidateTemplateResponse/ValidateTemplateResult' +\
421
# Check the response contains all the expected paramter keys
422
templ_params = ['DBUsername', 'LinuxDistribution', 'InstanceType',
423
'DBRootPassword', 'KeyName', 'DBPassword', 'DBName']
425
for param in templ_params:
426
lookup = '[ParameterKey="' + param + '"]'
427
val = self.stack.response_xml_item(response, prefix + lookup,
429
self.assertEqual(param, val)
430
print "ValidateTemplate : OK"
added
removed
heat/tests/functional/test_CFN_API_Actions.py
