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
26
@attr(tag=['func', 'wordpress', 'api', 'cfn', 'boto'])
27
class CfnApiBotoFunctionalTest(unittest.TestCase):
29
This test launches a wordpress stack then attempts to verify
30
correct operation of all actions supported by the heat CFN API
32
Note we use class-level fixtures to avoid setting up a new stack
33
for every test method, we set up the stack once then do all the
34
tests, this means all tests methods are performed on one class
35
instance, instead of creating a new class for every method, which
36
is the normal nose unittest.TestCase behavior.
38
The nose docs are a bit vague on how to do this, but it seems that
39
(setup|teardown)All works and they have to be classmethods.
41
Contrary to the nose docs, the class can be a unittest.TestCase subclass
43
This version of the test uses the boto client library, hence uses AWS auth
44
and checks the boto-parsed results rather than parsing the XML directly
49
template = 'WordPress_Single_Instance.template'
51
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
53
'DBPassword=' + os.environ['OS_PASSWORD']])
55
cls.logical_resource_name = 'WikiDatabase'
56
cls.logical_resource_type = 'AWS::EC2::Instance'
58
# Just to get the assert*() methods
59
class CfnApiFunctions(unittest.TestCase):
60
@unittest.skip('Not a real test case')
64
inst = CfnApiFunctions()
65
cls.stack = util.StackBoto(inst, template, 'F17', 'x86_64', 'cfntools',
67
cls.WikiDatabase = util.Instance(inst, cls.logical_resource_name)
71
cls.WikiDatabase.wait_for_boot()
72
cls.WikiDatabase.check_cfntools()
73
cls.WikiDatabase.wait_for_provisioning()
75
cls.logical_resource_status = "CREATE_COMPLETE"
77
# Save some compiled regexes and strings for response validation
78
cls.time_re = re.compile(
79
"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$")
80
cls.description_re = re.compile(
81
"^AWS CloudFormation Sample Template")
82
cls.stack_status = "CREATE_COMPLETE"
83
cls.stack_status_reason = "Stack successfully created"
84
cls.stack_timeout = 60
85
cls.stack_disable_rollback = True
87
# Match the expected format for an instance's physical resource ID
88
cls.phys_res_id_re = re.compile(
89
"^[0-9a-z]*-[0-9a-z]*-[0-9a-z]*-[0-9a-z]*-[0-9a-z]*$")
90
except Exception as ex:
91
print "setupAll failed : %s" % ex
100
def test_instance(self):
101
# ensure wordpress was installed by checking for expected
102
# configuration file over ssh
103
# This is the same as the standard wordress template test
104
# but we still do it to prove the stack is OK
105
self.assertTrue(self.WikiDatabase.file_present
106
('/etc/wordpress/wp-config.php'))
107
print "Wordpress installation detected"
109
# Verify the output URL parses as expected, ie check that
110
# the wordpress installation is operational
111
stack_url = self.stack.get_stack_output("WebsiteURL")
112
print "Got stack output WebsiteURL=%s, verifying" % stack_url
113
ver = verify.VerifyStack()
114
self.assertTrue(ver.verify_wordpress(stack_url))
116
def testListStacks(self):
117
response = self.stack.heatclient.list_stacks()
118
prefix = '/ListStacksResponse/ListStacksResult/StackSummaries/member'
120
# Extract the StackSummary for this stack
121
summary = [s for s in response
122
if s.stack_name == self.stack.stackname]
123
self.assertEqual(len(summary), 1)
125
# Note the boto StackSummary object does not contain every item
126
# output by our API (ie defined in the AWS docs), we can only
127
# test what boto encapsulates in the StackSummary class
128
self.stack.check_stackid(summary[0].stack_id)
130
self.assertEqual(type(summary[0].creation_time), datetime.datetime)
132
self.assertTrue(self.description_re.match(
133
summary[0].template_description) is not None)
135
self.assertEqual(summary[0].stack_name, self.stack.stackname)
137
self.assertEqual(summary[0].stack_status, self.stack_status)
139
print "ListStacks : OK"
141
def testDescribeStacks(self):
143
parameters['StackName'] = self.stack.stackname
144
response = self.stack.heatclient.describe_stacks(**parameters)
146
# Extract the Stack object for this stack
147
stacks = [s for s in response
148
if s.stack_name == self.stack.stackname]
149
self.assertEqual(len(stacks), 1)
151
self.assertEqual(type(stacks[0].creation_time), datetime.datetime)
153
self.stack.check_stackid(stacks[0].stack_id)
155
self.assertTrue(self.description_re.match(stacks[0].description)
158
self.assertEqual(stacks[0].stack_status_reason,
159
self.stack_status_reason)
161
self.assertEqual(stacks[0].stack_name, self.stack.stackname)
163
self.assertEqual(stacks[0].stack_status, self.stack_status)
165
self.assertEqual(stacks[0].timeout_in_minutes, self.stack_timeout)
167
self.assertEqual(stacks[0].disable_rollback,
168
self.stack_disable_rollback)
170
# Create a dict to lookup the expected template parameters
171
# NoEcho parameters are masked with 6 asterisks
172
template_parameters = {'DBUsername': '******',
173
'LinuxDistribution': 'F17',
174
'InstanceType': 'm1.xlarge',
175
'DBRootPassword': '******',
176
'KeyName': self.stack.keyname,
177
'DBPassword': '******',
178
'DBName': 'wordpress'}
180
for key, value in template_parameters.iteritems():
181
# The parameters returned via the API include a couple
182
# of fields which we don't care about (region/stackname)
183
# and may possibly end up getting removed, so we just
184
# look for the list of expected parameters above
185
plist = [p for p in s.parameters if p.key == key]
186
self.assertEqual(len(plist), 1)
187
self.assertEqual(key, plist[0].key)
188
self.assertEqual(value, plist[0].value)
190
# Then to a similar lookup to verify the Outputs section
191
expected_url = "http://" + self.WikiDatabase.ip + "/wordpress"
192
self.assertEqual(len(s.outputs), 1)
193
self.assertEqual(s.outputs[0].key, 'WebsiteURL')
194
self.assertEqual(s.outputs[0].value, expected_url)
196
print "DescribeStacks : OK"
198
def testDescribeStackEvents(self):
200
parameters['StackName'] = self.stack.stackname
201
response = self.stack.heatclient.list_stack_events(**parameters)
202
events = [e for e in response
203
if e.logical_resource_id == self.logical_resource_name
204
and e.resource_status == self.logical_resource_status]
206
self.assertEqual(len(events), 1)
208
self.stack.check_stackid(events[0].stack_id)
210
self.assertTrue(re.match("[0-9]*$", events[0].event_id) is not None)
212
self.assertEqual(events[0].resource_status,
213
self.logical_resource_status)
215
self.assertEqual(events[0].resource_type, self.logical_resource_type)
217
self.assertEqual(type(events[0].timestamp), datetime.datetime)
219
self.assertEqual(events[0].resource_status_reason, "state changed")
221
self.assertEqual(events[0].stack_name, self.stack.stackname)
223
self.assertEqual(events[0].logical_resource_id,
224
self.logical_resource_name)
226
self.assertTrue(self.phys_res_id_re.match(
227
events[0].physical_resource_id) is not None)
229
# Check ResourceProperties, skip pending resolution of #245
230
properties = json.loads(events[0].resource_properties)
231
self.assertEqual(properties["InstanceType"], "m1.xlarge")
233
print "DescribeStackEvents : OK"
235
def testGetTemplate(self):
237
parameters['StackName'] = self.stack.stackname
238
response = self.stack.heatclient.get_template(**parameters)
239
self.assertTrue(response is not None)
241
result = response['GetTemplateResponse']['GetTemplateResult']
242
self.assertTrue(result is not None)
243
template = result['TemplateBody']
244
self.assertTrue(template is not None)
246
# Then sanity check content - I guess we could diff
247
# with the template file but for now just check the
248
# description looks sane..
249
description = template['Description']
250
self.assertTrue(self.description_re.match(description) is not None)
252
print "GetTemplate : OK"
254
def testDescribeStackResource(self):
255
parameters = {'StackName': self.stack.stackname,
256
'LogicalResourceId': self.logical_resource_name}
257
response = self.stack.heatclient.describe_stack_resource(**parameters)
259
# Note boto_client response for this is a dict, if upstream
260
# pull request ever gets merged, this will change, see note/
261
# link in boto_client.py
262
desc_resp = response['DescribeStackResourceResponse']
263
self.assertTrue(desc_resp is not None)
264
desc_result = desc_resp['DescribeStackResourceResult']
265
self.assertTrue(desc_result is not None)
266
res = desc_result['StackResourceDetail']
267
self.assertTrue(res is not None)
269
self.stack.check_stackid(res['StackId'])
271
self.assertEqual(res['ResourceStatus'], self.logical_resource_status)
273
self.assertEqual(res['ResourceType'], self.logical_resource_type)
275
# Note due to issue mentioned above timestamp is a string in this case
276
# not a datetime.datetime object
277
self.assertTrue(self.time_re.match(res['LastUpdatedTimestamp'])
280
self.assertEqual(res['ResourceStatusReason'], 'state changed')
282
self.assertEqual(res['StackName'], self.stack.stackname)
284
self.assertEqual(res['LogicalResourceId'], self.logical_resource_name)
286
self.assertTrue(self.phys_res_id_re.match(res['PhysicalResourceId'])
289
self.assertTrue("AWS::CloudFormation::Init" in res['Metadata'])
291
print "DescribeStackResource : OK"
293
def testDescribeStackResources(self):
294
parameters = {'NameOrPid': self.stack.stackname,
295
'LogicalResourceId': self.logical_resource_name}
296
response = self.stack.heatclient.describe_stack_resources(**parameters)
297
self.assertEqual(len(response), 1)
300
self.assertTrue(res is not None)
302
self.stack.check_stackid(res.stack_id)
304
self.assertEqual(res.resource_status, self.logical_resource_status)
306
self.assertEqual(res.resource_type, self.logical_resource_type)
308
self.assertEqual(type(res.timestamp), datetime.datetime)
310
self.assertEqual(res.resource_status_reason, 'state changed')
312
self.assertEqual(res.stack_name, self.stack.stackname)
314
self.assertEqual(res.logical_resource_id, self.logical_resource_name)
316
self.assertTrue(self.phys_res_id_re.match(res.physical_resource_id)
319
print "DescribeStackResources : OK"
321
def testListStackResources(self):
323
parameters['StackName'] = self.stack.stackname
324
response = self.stack.heatclient.list_stack_resources(**parameters)
325
self.assertEqual(len(response), 1)
328
self.assertTrue(res is not None)
330
self.assertEqual(res.resource_status, self.logical_resource_status)
332
self.assertEqual(res.resource_status_reason, 'state changed')
334
self.assertEqual(type(res.last_updated_timestamp), datetime.datetime)
336
self.assertEqual(res.resource_type, self.logical_resource_type)
338
self.assertEqual(res.logical_resource_id, self.logical_resource_name)
340
self.assertTrue(self.phys_res_id_re.match(res.physical_resource_id)
343
print "ListStackResources : OK"
345
def testValidateTemplate(self):
346
# Use stack.format_parameters to get the TemplateBody
347
params = self.stack.format_parameters()
348
val_params = {'TemplateBody': params['TemplateBody']}
349
response = self.stack.heatclient.validate_template(**val_params)
350
# Check the response contains all the expected paramter keys
351
templ_params = ['DBUsername', 'LinuxDistribution', 'InstanceType',
352
'DBRootPassword', 'KeyName', 'DBPassword', 'DBName']
354
resp_params = [p.parameter_key for p in response.template_parameters]
355
for param in templ_params:
356
self.assertTrue(param in resp_params)
357
print "ValidateTemplate : OK"
added
removed
heat/tests/functional/test_CFN_API_Actions_Boto.py
