diff --git a/heat/tests/functional/test_CFN_API_Actions.py b/heat/tests/functional/test_CFN_API_Actions.py new file mode 100644 index 0000000000..eeb5d1c75d --- /dev/null +++ b/heat/tests/functional/test_CFN_API_Actions.py @@ -0,0 +1,402 @@ +# 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 + + +@attr(speed='slow') +@attr(tag=['func', 'wordpress', 'api', 'cfn']) +class CfnApiFunctionalTest(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 + ''' + @classmethod + def setupAll(self): + print "SETUPALL" + template = 'WordPress_Single_Instance.template' + + stack_paramstr = ';'.join(['InstanceType=m1.xlarge', + 'DBUsername=dbuser', + 'DBPassword=' + os.environ['OS_PASSWORD']]) + + self.logical_resource_name = 'WikiDatabase' + self.logical_resource_type = 'AWS::EC2::Instance' + self.stack = util.Stack(template, 'F17', 'x86_64', 'cfntools', + stack_paramstr) + self.WikiDatabase = util.Instance(self.logical_resource_name) + self.stack.create() + self.WikiDatabase.wait_for_boot() + self.WikiDatabase.check_cfntools() + self.WikiDatabase.wait_for_provisioning() + + self.logical_resource_status = "CREATE_COMPLETE" + + # Save some compiled regexes and strings for response validation + self.stack_id_re = re.compile("^arn:openstack:heat::admin:stacks/" + + self.stack.stackname) + self.time_re = re.compile( + "[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$") + self.description_re = re.compile("^AWS CloudFormation Sample Template") + self.stack_status = "CREATE_COMPLETE" + self.stack_status_reason = "Stack successfully created" + self.stack_timeout = str(60) + self.stack_disable_rollback = "True" + + # Match the expected format for physical resource ID for an instance + self.phys_res_id_re = re.compile( + "^[0-9a-z]*-[0-9a-z]*-[0-9a-z]*-[0-9a-z]*-[0-9a-z]*$") + + @classmethod + def teardownAll(self): + print "TEARDOWNALL" + self.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' + + stack_id = self.stack.response_xml_item(response, prefix, "StackId") + self.assertTrue(self.stack_id_re.match(stack_id) != None) + + update_time = self.stack.response_xml_item(response, prefix, + "LastUpdatedTime") + self.assertTrue(self.time_re.match(update_time) != None) + + create_time = self.stack.response_xml_item(response, prefix, + "CreationTime") + self.assertTrue(self.time_re.match(create_time) != None) + + description = self.stack.response_xml_item(response, prefix, + "TemplateDescription") + self.assertTrue(self.description_re.match(description) != None) + + status_reason = self.stack.response_xml_item(response, prefix, + "StackStatusReason") + self.assertEqual(status_reason, self.stack_status_reason) + + stack_name = self.stack.response_xml_item(response, prefix, + "StackName") + self.assertEqual(stack_name, self.stack.stackname) + + stack_status = self.stack.response_xml_item(response, prefix, + "StackStatus") + self.assertEqual(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) + prefix = '/DescribeStacksResponse/DescribeStacksResult/Stacks/member' + + stack_id = self.stack.response_xml_item(response, prefix, "StackId") + self.assertTrue(self.stack_id_re.match(stack_id) != None) + + update_time = self.stack.response_xml_item(response, prefix, + "LastUpdatedTime") + self.assertTrue(self.time_re.match(update_time) != None) + + create_time = self.stack.response_xml_item(response, prefix, + "CreationTime") + self.assertTrue(self.time_re.match(create_time) != None) + + description = self.stack.response_xml_item(response, prefix, + "Description") + self.assertTrue(self.description_re.match(description) != None) + + status_reason = self.stack.response_xml_item(response, prefix, + "StackStatusReason") + self.assertEqual(status_reason, self.stack_status_reason) + + stack_name = self.stack.response_xml_item(response, prefix, + "StackName") + self.assertEqual(stack_name, self.stack.stackname) + + stack_status = self.stack.response_xml_item(response, prefix, + "StackStatus") + self.assertEqual(stack_status, self.stack_status) + + stack_timeout = self.stack.response_xml_item(response, prefix, + "TimeoutInMinutes") + self.assertEqual(stack_timeout, self.stack_timeout) + + disable_rollback = self.stack.response_xml_item(response, prefix, + "DisableRollback") + self.assertEqual(disable_rollback, self.stack_disable_rollback) + + # Create a dict to lookup the expected template parameters + template_parameters = {'DBUsername': 'admin', + 'LinuxDistribution': 'F16', + 'InstanceType': 'm1.large', + 'DBRootPassword': 'admin', + 'KeyName': 'None', + 'DBPassword': 'admin', + 'DBName': 'wordpress'} + + # We do a fully qualified xpath lookup to extract the paramter + # value for each key, then check the extracted value + param_prefix = prefix + "/Parameters/member" + for key, value in template_parameters.iteritems(): + lookup = '[ParameterKey="' + key + '" and ParameterValue="' +\ + value + '"]' + lookup_value = self.stack.response_xml_item(response, + param_prefix + lookup, + "ParameterValue") + self.assertEqual(lookup_value, value) + + # Then to a similar lookup to verify the Outputs section + expected_url = "http://" + self.WikiDatabase.ip + "/wordpress" + + outputs_prefix = prefix + "/Outputs/member" + lookup = '[OutputKey="WebsiteURL" and OutputValue="' + expected_url +\ + '" and Description="URL for Wordpress wiki"]' + lookup_value = self.stack.response_xml_item(response, + outputs_prefix + lookup, + "OutputValue") + self.assertEqual(lookup_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) + prefix = '/DescribeStackEventsResponse/DescribeStackEventsResult/' +\ + 'StackEvents/member[LogicalResourceId="' +\ + self.logical_resource_name + '" and ResourceStatus="' +\ + self.logical_resource_status + '"]' + + stack_id = self.stack.response_xml_item(response, prefix, "StackId") + self.assertTrue(self.stack_id_re.match(stack_id) != None) + + event_id = self.stack.response_xml_item(response, prefix, "EventId") + self.assertTrue(re.match("[0-9]*$", event_id) != None) + + resource_status = self.stack.response_xml_item(response, prefix, + "ResourceStatus") + self.assertEqual(resource_status, self.logical_resource_status) + + resource_type = self.stack.response_xml_item(response, prefix, + "ResourceType") + self.assertEqual(resource_type, self.logical_resource_type) + + update_time = self.stack.response_xml_item(response, prefix, + "Timestamp") + self.assertTrue(self.time_re.match(update_time) != None) + + status_data = self.stack.response_xml_item(response, prefix, + "ResourceStatusData") + self.assertEqual(status_data, "state changed") + + stack_name = self.stack.response_xml_item(response, prefix, + "StackName") + self.assertEqual(stack_name, self.stack.stackname) + + log_res_id = self.stack.response_xml_item(response, prefix, + "LogicalResourceId") + self.assertEqual(log_res_id, self.logical_resource_name) + + phys_res_id = self.stack.response_xml_item(response, prefix, + "PhysicalResourceId") + self.assertTrue(self.phys_res_id_re.match(phys_res_id) != None) + + # Check ResourceProperties, the format of this is not defined + # by the AWS API spec, so we just check one expected key is there + rp_prefix = prefix + '/ResourceProperties' + prop_type = self.stack.response_xml_item(response, rp_prefix, + "InstanceType") + self.assertEqual(prop_type, "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) + prefix = '/GetTemplateResponse/GetTemplateResult' + + # Extract the JSON TemplateBody and prove it parses + template = self.stack.response_xml_item(response, prefix, + "TemplateBody") + json_load = json.loads(template) + self.assertTrue(json_load != None) + + # Then sanity check content - I guess we could diff + # with the template file but for now just check the + # description looks sane.. + description = json_load['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) + prefix = '/DescribeStackResourceResponse/DescribeStackResourceResult'\ + + '/StackResourceDetail' + + stack_id = self.stack.response_xml_item(response, prefix, "StackId") + self.assertTrue(self.stack_id_re.match(stack_id) != None) + + resource_status = self.stack.response_xml_item(response, prefix, + "ResourceStatus") + self.assertEqual(resource_status, self.logical_resource_status) + + resource_type = self.stack.response_xml_item(response, prefix, + "ResourceType") + self.assertEqual(resource_type, self.logical_resource_type) + + update_time = self.stack.response_xml_item(response, prefix, + "LastUpdatedTimestamp") + self.assertTrue(self.time_re.match(update_time) != None) + + status_reason = self.stack.response_xml_item(response, prefix, + "ResourceStatusReason") + self.assertEqual(status_reason, "None") + + stack_name = self.stack.response_xml_item(response, prefix, + "StackName") + self.assertEqual(stack_name, self.stack.stackname) + + log_res_id = self.stack.response_xml_item(response, prefix, + "LogicalResourceId") + self.assertEqual(log_res_id, self.logical_resource_name) + + phys_res_id = self.stack.response_xml_item(response, prefix, + "PhysicalResourceId") + self.assertTrue(self.phys_res_id_re.match(phys_res_id) != None) + + metadata = self.stack.response_xml_item(response, prefix, "Metadata") + json_load = json.loads(metadata) + self.assertTrue(json_load != None) + self.assertTrue("AWS::CloudFormation::Init" in json_load) + + 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) + prefix = '/DescribeStackResourcesResponse/' +\ + 'DescribeStackResourcesResult/StackResources/member' + + stack_id = self.stack.response_xml_item(response, prefix, "StackId") + self.assertTrue(self.stack_id_re.match(stack_id) != None) + + resource_status = self.stack.response_xml_item(response, prefix, + "ResourceStatus") + self.assertEqual(resource_status, self.logical_resource_status) + + resource_type = self.stack.response_xml_item(response, prefix, + "ResourceType") + self.assertEqual(resource_type, self.logical_resource_type) + + update_time = self.stack.response_xml_item(response, prefix, + "Timestamp") + self.assertTrue(self.time_re.match(update_time) != None) + + status_reason = self.stack.response_xml_item(response, prefix, + "ResourceStatusReason") + self.assertEqual(status_reason, "None") + + stack_name = self.stack.response_xml_item(response, prefix, + "StackName") + self.assertEqual(stack_name, self.stack.stackname) + + log_res_id = self.stack.response_xml_item(response, prefix, + "LogicalResourceId") + self.assertEqual(log_res_id, self.logical_resource_name) + + phys_res_id = self.stack.response_xml_item(response, prefix, + "PhysicalResourceId") + self.assertTrue(self.phys_res_id_re.match(phys_res_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) + prefix = '/ListStackResourcesResponse/ListStackResourcesResult' +\ + '/StackResourceSummaries/member' + + resource_status = self.stack.response_xml_item(response, prefix, + "ResourceStatus") + self.assertEqual(resource_status, self.logical_resource_status) + + status_reason = self.stack.response_xml_item(response, prefix, + "ResourceStatusReason") + self.assertEqual(status_reason, "None") + + update_time = self.stack.response_xml_item(response, prefix, + "LastUpdatedTimestamp") + self.assertTrue(self.time_re.match(update_time) != None) + + resource_type = self.stack.response_xml_item(response, prefix, + "ResourceType") + self.assertEqual(resource_type, self.logical_resource_type) + + log_res_id = self.stack.response_xml_item(response, prefix, + "LogicalResourceId") + self.assertEqual(log_res_id, self.logical_resource_name) + + phys_res_id = self.stack.response_xml_item(response, prefix, + "PhysicalResourceId") + self.assertTrue(self.phys_res_id_re.match(phys_res_id) != None) + + print "ListStackResources : OK"