--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+#
+
+import os
+import util
+import verify
+import re
+import nose
+from nose.plugins.attrib import attr
+import unittest
+import json
+import datetime
+
+
+@attr(speed='slow')
+@attr(tag=['func', 'wordpress', 'api', 'cfn', 'boto'])
+class CfnApiBotoFunctionalTest(unittest.TestCase):
+ '''
+ This test launches a wordpress stack then attempts to verify
+ correct operation of all actions supported by the heat CFN API
+
+ Note we use class-level fixtures to avoid setting up a new stack
+ for every test method, we set up the stack once then do all the
+ tests, this means all tests methods are performed on one class
+ instance, instead of creating a new class for every method, which
+ is the normal nose unittest.TestCase behavior.
+
+ The nose docs are a bit vague on how to do this, but it seems that
+ (setup|teardown)All works and they have to be classmethods.
+
+ Contrary to the nose docs, the class can be a unittest.TestCase subclass
+
+ This version of the test uses the boto client library, hence uses AWS auth
+ and checks the boto-parsed results rather than parsing the XML directly
+ '''
+ @classmethod
+ def setupAll(cls):
+ print "SETUPALL"
+ template = 'WordPress_Single_Instance.template'
+
+ stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
+
+ cls.logical_resource_name = 'WikiDatabase'
+ cls.logical_resource_type = 'AWS::EC2::Instance'
+
+ # Just to get the assert*() methods
+ class CfnApiFunctions(unittest.TestCase):
+ @unittest.skip('Not a real test case')
+ def runTest(self):
+ pass
+
+ inst = CfnApiFunctions()
+ cls.stack = util.StackBoto(inst, template, 'F17', 'x86_64', 'cfntools',
+ stack_paramstr)
+ cls.WikiDatabase = util.Instance(inst, cls.logical_resource_name)
+
+ try:
+ cls.stack.create()
+ cls.WikiDatabase.wait_for_boot()
+ cls.WikiDatabase.check_cfntools()
+ cls.WikiDatabase.wait_for_provisioning()
+
+ cls.logical_resource_status = "CREATE_COMPLETE"
+
+ # Save some compiled regexes and strings for response validation
+ cls.stack_id_re = re.compile("^arn:openstack:heat::admin:stacks/"
+ + cls.stack.stackname)
+ cls.time_re = re.compile(
+ "[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$")
+ cls.description_re = re.compile(
+ "^AWS CloudFormation Sample Template")
+ cls.stack_status = "CREATE_COMPLETE"
+ cls.stack_status_reason = "Stack successfully created"
+ cls.stack_timeout = 60
+ cls.stack_disable_rollback = True
+
+ # Match the expected format for an instance's physical resource ID
+ cls.phys_res_id_re = re.compile(
+ "^[0-9a-z]*-[0-9a-z]*-[0-9a-z]*-[0-9a-z]*-[0-9a-z]*$")
+ except:
+ cls.stack.cleanup()
+ raise
+
+ @classmethod
+ def teardownAll(cls):
+ print "TEARDOWNALL"
+ cls.stack.cleanup()
+
+ def test_instance(self):
+ # ensure wordpress was installed by checking for expected
+ # configuration file over ssh
+ # This is the same as the standard wordress template test
+ # but we still do it to prove the stack is OK
+ self.assertTrue(self.WikiDatabase.file_present
+ ('/etc/wordpress/wp-config.php'))
+ print "Wordpress installation detected"
+
+ # Verify the output URL parses as expected, ie check that
+ # the wordpress installation is operational
+ stack_url = self.stack.get_stack_output("WebsiteURL")
+ print "Got stack output WebsiteURL=%s, verifying" % stack_url
+ ver = verify.VerifyStack()
+ self.assertTrue(ver.verify_wordpress(stack_url))
+
+ def testListStacks(self):
+ client = self.stack.get_heat_client()
+ response = client.list_stacks()
+ prefix = '/ListStacksResponse/ListStacksResult/StackSummaries/member'
+
+ # Extract the StackSummary for this stack
+ summary = [s for s in response
+ if s.stack_name == self.stack.stackname]
+ self.assertEqual(len(summary), 1)
+
+ # Note the boto StackSummary object does not contain every item
+ # output by our API (ie defined in the AWS docs), we can only
+ # test what boto encapsulates in the StackSummary class
+ self.assertTrue(self.stack_id_re.match(summary[0].stack_id) != None)
+
+ self.assertEqual(type(summary[0].creation_time), datetime.datetime)
+
+ self.assertTrue(self.description_re.match(
+ summary[0].template_description) != None)
+
+ self.assertEqual(summary[0].stack_name, self.stack.stackname)
+
+ self.assertEqual(summary[0].stack_status, self.stack_status)
+
+ print "ListStacks : OK"
+
+ def testDescribeStacks(self):
+ client = self.stack.get_heat_client()
+ parameters = {}
+ parameters['StackName'] = self.stack.stackname
+ response = client.describe_stacks(**parameters)
+
+ # Extract the Stack object for this stack
+ stacks = [s for s in response
+ if s.stack_name == self.stack.stackname]
+ self.assertEqual(len(stacks), 1)
+
+ self.assertEqual(type(stacks[0].creation_time), datetime.datetime)
+
+ self.assertTrue(self.stack_id_re.match(stacks[0].stack_id) != None)
+
+ self.assertTrue(self.description_re.match(stacks[0].description)
+ != None)
+
+ self.assertEqual(stacks[0].stack_status_reason,
+ self.stack_status_reason)
+
+ self.assertEqual(stacks[0].stack_name, self.stack.stackname)
+
+ self.assertEqual(stacks[0].stack_status, self.stack_status)
+
+ self.assertEqual(stacks[0].timeout_in_minutes, self.stack_timeout)
+
+ self.assertEqual(stacks[0].disable_rollback,
+ self.stack_disable_rollback)
+
+ # Create a dict to lookup the expected template parameters
+ template_parameters = {'DBUsername': 'dbuser',
+ 'LinuxDistribution': 'F17',
+ 'InstanceType': 'm1.xlarge',
+ 'DBRootPassword': 'admin',
+ 'KeyName': self.stack.keyname,
+ 'DBPassword':
+ os.environ['OS_PASSWORD'],
+ 'DBName': 'wordpress'}
+
+ for key, value in template_parameters.iteritems():
+ # The parameters returned via the API include a couple
+ # of fields which we don't care about (region/stackname)
+ # and may possibly end up getting removed, so we just
+ # look for the list of expected parameters above
+ plist = [p for p in s.parameters if p.key == key]
+ self.assertEqual(len(plist), 1)
+ self.assertEqual(key, plist[0].key)
+ self.assertEqual(value, plist[0].value)
+
+ # Then to a similar lookup to verify the Outputs section
+ expected_url = "http://" + self.WikiDatabase.ip + "/wordpress"
+ self.assertEqual(len(s.outputs), 1)
+ self.assertEqual(s.outputs[0].key, 'WebsiteURL')
+ self.assertEqual(s.outputs[0].value, expected_url)
+
+ print "DescribeStacks : OK"
+
+ def testDescribeStackEvents(self):
+
+ client = self.stack.get_heat_client()
+ parameters = {}
+ parameters['StackName'] = self.stack.stackname
+ response = client.list_stack_events(**parameters)
+ events = [e for e in response
+ if e.logical_resource_id == self.logical_resource_name
+ and e.resource_status == self.logical_resource_status]
+
+ self.assertEqual(len(events), 1)
+
+ self.assertTrue(self.stack_id_re.match(events[0].stack_id) != None)
+
+ self.assertTrue(re.match("[0-9]*$", events[0].event_id) != None)
+
+ self.assertEqual(events[0].resource_status,
+ self.logical_resource_status)
+
+ self.assertEqual(events[0].resource_type, self.logical_resource_type)
+
+ self.assertEqual(type(events[0].timestamp), datetime.datetime)
+
+ self.assertEqual(events[0].resource_status_reason, "state changed")
+
+ self.assertEqual(events[0].stack_name, self.stack.stackname)
+
+ self.assertEqual(events[0].logical_resource_id,
+ self.logical_resource_name)
+
+ self.assertTrue(self.phys_res_id_re.match(
+ events[0].physical_resource_id) != None)
+
+ # Check ResourceProperties, skip pending resolution of #245
+ properties = json.loads(events[0].resource_properties)
+ self.assertEqual(properties["InstanceType"], "m1.xlarge")
+
+ print "DescribeStackEvents : OK"
+
+ def testGetTemplate(self):
+ client = self.stack.get_heat_client()
+ parameters = {}
+ parameters['StackName'] = self.stack.stackname
+ response = client.get_template(**parameters)
+ self.assertTrue(response != None)
+
+ result = response['GetTemplateResponse']['GetTemplateResult']
+ self.assertTrue(result != None)
+ template = result['TemplateBody']
+ self.assertTrue(template != None)
+
+ # Then sanity check content - I guess we could diff
+ # with the template file but for now just check the
+ # description looks sane..
+ description = template['Description']
+ self.assertTrue(self.description_re.match(description) != None)
+
+ print "GetTemplate : OK"
+
+ def testDescribeStackResource(self):
+ client = self.stack.get_heat_client()
+ parameters = {'StackName': self.stack.stackname,
+ 'LogicalResourceId': self.logical_resource_name}
+ response = client.describe_stack_resource(**parameters)
+
+ # Note boto_client response for this is a dict, if upstream
+ # pull request ever gets merged, this will change, see note/
+ # link in boto_client.py
+ desc_resp = response['DescribeStackResourceResponse']
+ self.assertTrue(desc_resp != None)
+ desc_result = desc_resp['DescribeStackResourceResult']
+ self.assertTrue(desc_result != None)
+ res = desc_result['StackResourceDetail']
+ self.assertTrue(res != None)
+
+ self.assertTrue(self.stack_id_re.match(res['StackId']) != None)
+
+ self.assertEqual(res['ResourceStatus'], self.logical_resource_status)
+
+ self.assertEqual(res['ResourceType'], self.logical_resource_type)
+
+ # Note due to issue mentioned above timestamp is a string in this case
+ # not a datetime.datetime object
+ self.assertTrue(self.time_re.match(res['LastUpdatedTimestamp'])
+ != None)
+
+ self.assertEqual(res['ResourceStatusReason'], None)
+
+ self.assertEqual(res['StackName'], self.stack.stackname)
+
+ self.assertEqual(res['LogicalResourceId'], self.logical_resource_name)
+
+ self.assertTrue(self.phys_res_id_re.match(res['PhysicalResourceId'])
+ != None)
+
+ self.assertTrue("AWS::CloudFormation::Init" in res['Metadata'])
+
+ print "DescribeStackResource : OK"
+
+ def testDescribeStackResources(self):
+ client = self.stack.get_heat_client()
+ parameters = {'NameOrPid': self.stack.stackname,
+ 'LogicalResourceId': self.logical_resource_name}
+ response = client.describe_stack_resources(**parameters)
+ self.assertEqual(len(response), 1)
+
+ res = response[0]
+ self.assertTrue(res != None)
+
+ self.assertTrue(self.stack_id_re.match(res.stack_id) != None)
+
+ self.assertEqual(res.resource_status, self.logical_resource_status)
+
+ self.assertEqual(res.resource_type, self.logical_resource_type)
+
+ self.assertEqual(type(res.timestamp), datetime.datetime)
+
+ self.assertEqual(res.resource_status_reason, 'None')
+
+ self.assertEqual(res.stack_name, self.stack.stackname)
+
+ self.assertEqual(res.logical_resource_id, self.logical_resource_name)
+
+ self.assertTrue(self.phys_res_id_re.match(res.physical_resource_id)
+ != None)
+
+ print "DescribeStackResources : OK"
+
+ def testListStackResources(self):
+ client = self.stack.get_heat_client()
+ parameters = {}
+ parameters['StackName'] = self.stack.stackname
+ response = client.list_stack_resources(**parameters)
+ self.assertEqual(len(response), 1)
+
+ res = response[0]
+ self.assertTrue(res != None)
+
+ self.assertEqual(res.resource_status, self.logical_resource_status)
+
+ self.assertEqual(res.resource_status_reason, 'None')
+
+ self.assertEqual(type(res.last_updated_timestamp), datetime.datetime)
+
+ self.assertEqual(res.resource_type, self.logical_resource_type)
+
+ self.assertEqual(res.logical_resource_id, self.logical_resource_name)
+
+ self.assertTrue(self.phys_res_id_re.match(res.physical_resource_id)
+ != None)
+
+ print "ListStackResources : OK"
Code Review / openstack-build / heat-build.git / commitdiff