From f40a4bb33346fb4022dfb6b94eab12befd61bd91 Mon Sep 17 00:00:00 2001 From: Clint Byrum Date: Thu, 25 Apr 2013 16:17:24 -0700 Subject: [PATCH] Remove functional tests in preferece for tempest These tests are not unit tests, and take a long time. They are being moved into tempest, which is where integration/functional tests live in OpenStack. Change-Id: Ida7792a00a8565f812c55da448d8c50a8b8a90b9 --- .../test_AutoScalingMultiAZSample.py | 80 --- heat/tests/functional/test_CFN_API_Actions.py | 430 ------------ .../functional/test_CFN_API_Actions_Boto.py | 357 ---------- .../functional/test_CFN_API_UpdateStack.py | 149 ---- .../test_HAProxy_Single_Instance.py | 87 --- .../test_OpenShift_Prebuilt_JEOS.py | 65 -- .../functional/test_WordPress_2_Instances.py | 63 -- .../test_WordPress_2_Instances_With_EBS.py | 75 -- ...test_WordPress_2_Instances_With_EBS_EIP.py | 92 --- .../test_WordPress_Composed_Instances.py | 66 -- .../test_WordPress_Single_Instance.py | 56 -- .../test_WordPress_Single_Instance_Boto.py | 59 -- ...test_WordPress_Single_Instance_With_EBS.py | 68 -- ..._WordPress_Single_Instance_With_EBS_EIP.py | 91 --- ...test_WordPress_Single_Instance_With_EIP.py | 80 --- .../test_WordPress_Single_Instance_With_HA.py | 74 -- ...test_WordPress_Single_Instance_With_IHA.py | 107 --- .../functional/test_WordPress_With_LB.py | 68 -- .../functional/test_WordPress_With_RDS.py | 69 -- heat/tests/functional/util.py | 642 ------------------ heat/tests/functional/verify.py | 86 --- 21 files changed, 2864 deletions(-) delete mode 100644 heat/tests/functional/test_AutoScalingMultiAZSample.py delete mode 100644 heat/tests/functional/test_CFN_API_Actions.py delete mode 100644 heat/tests/functional/test_CFN_API_Actions_Boto.py delete mode 100644 heat/tests/functional/test_CFN_API_UpdateStack.py delete mode 100644 heat/tests/functional/test_HAProxy_Single_Instance.py delete mode 100644 heat/tests/functional/test_OpenShift_Prebuilt_JEOS.py delete mode 100644 heat/tests/functional/test_WordPress_2_Instances.py delete mode 100644 heat/tests/functional/test_WordPress_2_Instances_With_EBS.py delete mode 100644 heat/tests/functional/test_WordPress_2_Instances_With_EBS_EIP.py delete mode 100644 heat/tests/functional/test_WordPress_Composed_Instances.py delete mode 100644 heat/tests/functional/test_WordPress_Single_Instance.py delete mode 100644 heat/tests/functional/test_WordPress_Single_Instance_Boto.py delete mode 100644 heat/tests/functional/test_WordPress_Single_Instance_With_EBS.py delete mode 100644 heat/tests/functional/test_WordPress_Single_Instance_With_EBS_EIP.py delete mode 100644 heat/tests/functional/test_WordPress_Single_Instance_With_EIP.py delete mode 100644 heat/tests/functional/test_WordPress_Single_Instance_With_HA.py delete mode 100644 heat/tests/functional/test_WordPress_Single_Instance_With_IHA.py delete mode 100644 heat/tests/functional/test_WordPress_With_LB.py delete mode 100644 heat/tests/functional/test_WordPress_With_RDS.py delete mode 100644 heat/tests/functional/util.py delete mode 100644 heat/tests/functional/verify.py diff --git a/heat/tests/functional/test_AutoScalingMultiAZSample.py b/heat/tests/functional/test_AutoScalingMultiAZSample.py deleted file mode 100644 index ff7de037..00000000 --- a/heat/tests/functional/test_AutoScalingMultiAZSample.py +++ /dev/null @@ -1,80 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr -import unittest -import os -from time import sleep - - -@attr(speed='slow') -@attr(tag=['func', 'autoscaling', 'AutoScalingMultiAZSample.template']) -class AutoScalingMultiAZSampleFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'AutoScalingMultiAZSample.template' - - stack_paramstr = ';'.join(['InstanceType=m1.small', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WebServerGroup0 = util.Instance(self, 'WebServerGroup-0') - - def tearDown(self): - pass - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - self.WebServerGroup0.wait_for_boot() - self.WebServerGroup0.check_cfntools() - self.WebServerGroup0.wait_for_provisioning() - - # TODO: verify the code below tests the template properly - -# TODO(sdake) use a util exists function for nonexistent instances (needs dev) - # Trigger the load balancer by taking up memory - self.WebServerGroup0.exec_command('memhog -r100000 1500m') - - # Give the load balancer 2 minutes to react - sleep(2 * 60) - - self.WebServerGroup1 = util.Instance(self, 'WebServerGroup-1') - # Verify the second instance gets launched - self.assertTrue(self.WebServerGroup1.exists()) - self.WebServerGroup1.wait_for_boot() - self.WebServerGroup1.check_cfntools() - self.WebServerGroup1.wait_for_provisioning() - - # ensure wordpress was installed by checking for expected - # configuration file over ssh - self.assertTrue(self.WebServerGroup0.file_present - ('/etc/wordpress/wp-config.php')) - print "Wordpress installation detected on WSG0" - - # ensure wordpress was installed by checking for expected - # configuration file over ssh - self.assertTrue(self.WebServerGroup1.file_present - ('/etc/wordpress/wp-config.php')) - print "Wordpress installation detected on WSG1" - - # Verify the output URL parses as expected, ie check that - # the wordpress installation is operational - stack_url = self.stack.get_stack_output("URL") - print "Got stack output WebsiteURL=%s, verifying" % stack_url - ver = verify.VerifyStack() - self.assertTrue(ver.verify_wordpress(stack_url)) diff --git a/heat/tests/functional/test_CFN_API_Actions.py b/heat/tests/functional/test_CFN_API_Actions.py deleted file mode 100644 index 7a661614..00000000 --- a/heat/tests/functional/test_CFN_API_Actions.py +++ /dev/null @@ -1,430 +0,0 @@ -# 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 -from nose.plugins.attrib import attr -import unittest -import json - -from heat.common import template_format - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'api', 'cfn', 'F17']) -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(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.Stack(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.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 = str(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 Exception as ex: - print "setupAll failed : %s" % ex - 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): - response = self.stack.heatclient.list_stacks() - prefix = '/ListStacksResponse/ListStacksResult/StackSummaries/member' - - stack_id = self.stack.response_xml_item(response, prefix, "StackId") - self.stack.check_stackid(stack_id) - - update_time = self.stack.response_xml_item(response, prefix, - "LastUpdatedTime") - self.assertTrue(self.time_re.match(update_time) is not None) - - create_time = self.stack.response_xml_item(response, prefix, - "CreationTime") - self.assertTrue(self.time_re.match(create_time) is not None) - - description = self.stack.response_xml_item(response, prefix, - "TemplateDescription") - self.assertTrue(self.description_re.match(description) is not 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): - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.describe_stacks(**parameters) - prefix = '/DescribeStacksResponse/DescribeStacksResult/Stacks/member' - - stack_id = self.stack.response_xml_item(response, prefix, "StackId") - self.stack.check_stackid(stack_id) - - update_time = self.stack.response_xml_item(response, prefix, - "LastUpdatedTime") - self.assertTrue(self.time_re.match(update_time) is not None) - - create_time = self.stack.response_xml_item(response, prefix, - "CreationTime") - self.assertTrue(self.time_re.match(create_time) is not None) - - description = self.stack.response_xml_item(response, prefix, - "Description") - self.assertTrue(self.description_re.match(description) is not 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 - # NoEcho parameters are masked with 6 asterisks - template_parameters = {'DBUsername': '******', - 'LinuxDistribution': 'F17', - 'InstanceType': 'm1.xlarge', - 'DBRootPassword': '******', - 'KeyName': self.stack.keyname, - 'DBPassword': '******', - '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): - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.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.stack.check_stackid(stack_id) - - event_id = self.stack.response_xml_item(response, prefix, "EventId") - self.assertTrue(re.match("[0-9]*$", event_id) is not 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) is not None) - - status_data = self.stack.response_xml_item(response, prefix, - "ResourceStatusReason") - 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) is not None) - - # ResourceProperties format is defined as a string "blob" by AWS - # we return JSON encoded properies, so decode and check one key - prop_json = self.stack.response_xml_item(response, prefix, - "ResourceProperties") - self.assertTrue(prop_json is not None) - - prop = json.loads(prop_json) - self.assertEqual(prop["InstanceType"], "m1.xlarge") - - print "DescribeStackEvents : OK" - - def testGetTemplate(self): - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.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 = template_format.parse(template) - self.assertTrue(json_load is not 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) is not None) - - print "GetTemplate : OK" - - def testDescribeStackResource(self): - parameters = {'StackName': self.stack.stackname, - 'LogicalResourceId': self.logical_resource_name} - response = self.stack.heatclient.describe_stack_resource(**parameters) - prefix = '/DescribeStackResourceResponse/DescribeStackResourceResult'\ - + '/StackResourceDetail' - - stack_id = self.stack.response_xml_item(response, prefix, "StackId") - self.stack.check_stackid(stack_id) - - 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) is not None) - - status_reason = self.stack.response_xml_item(response, prefix, - "ResourceStatusReason") - self.assertEqual(status_reason, "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) is not None) - - metadata = self.stack.response_xml_item(response, prefix, "Metadata") - json_load = json.loads(metadata) - self.assertTrue(json_load is not None) - self.assertTrue("AWS::CloudFormation::Init" in json_load) - - print "DescribeStackResource : OK" - - def testDescribeStackResources(self): - parameters = {'NameOrPid': self.stack.stackname, - 'LogicalResourceId': self.logical_resource_name} - response = self.stack.heatclient.describe_stack_resources(**parameters) - prefix = '/DescribeStackResourcesResponse/' +\ - 'DescribeStackResourcesResult/StackResources/member' - - stack_id = self.stack.response_xml_item(response, prefix, "StackId") - self.stack.check_stackid(stack_id) - - 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) is not None) - - status_reason = self.stack.response_xml_item(response, prefix, - "ResourceStatusReason") - self.assertEqual(status_reason, "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) is not None) - - print "DescribeStackResources : OK" - - def testListStackResources(self): - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.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, "state changed") - - update_time = self.stack.response_xml_item(response, prefix, - "LastUpdatedTimestamp") - self.assertTrue(self.time_re.match(update_time) is not 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) is not None) - - print "ListStackResources : OK" - - def testValidateTemplate(self): - # Use stack.format_parameters to get the TemplateBody - params = self.stack.format_parameters() - val_params = {'TemplateBody': params['TemplateBody']} - response = self.stack.heatclient.validate_template(**val_params) - prefix = '/ValidateTemplateResponse/ValidateTemplateResult' +\ - '/Parameters/member' - # Check the response contains all the expected paramter keys - templ_params = ['DBUsername', 'LinuxDistribution', 'InstanceType', - 'DBRootPassword', 'KeyName', 'DBPassword', 'DBName'] - - for param in templ_params: - lookup = '[ParameterKey="' + param + '"]' - val = self.stack.response_xml_item(response, prefix + lookup, - 'ParameterKey') - self.assertEqual(param, val) - print "ValidateTemplate : OK" diff --git a/heat/tests/functional/test_CFN_API_Actions_Boto.py b/heat/tests/functional/test_CFN_API_Actions_Boto.py deleted file mode 100644 index 2dff0237..00000000 --- a/heat/tests/functional/test_CFN_API_Actions_Boto.py +++ /dev/null @@ -1,357 +0,0 @@ -# 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 -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.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 Exception as ex: - print "setupAll failed : %s" % ex - 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): - response = self.stack.heatclient.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.stack.check_stackid(summary[0].stack_id) - - self.assertEqual(type(summary[0].creation_time), datetime.datetime) - - self.assertTrue(self.description_re.match( - summary[0].template_description) is not 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): - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.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.stack.check_stackid(stacks[0].stack_id) - - self.assertTrue(self.description_re.match(stacks[0].description) - is not 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 - # NoEcho parameters are masked with 6 asterisks - template_parameters = {'DBUsername': '******', - 'LinuxDistribution': 'F17', - 'InstanceType': 'm1.xlarge', - 'DBRootPassword': '******', - 'KeyName': self.stack.keyname, - 'DBPassword': '******', - '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): - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.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.stack.check_stackid(events[0].stack_id) - - self.assertTrue(re.match("[0-9]*$", events[0].event_id) is not 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) is not 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): - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.get_template(**parameters) - self.assertTrue(response is not None) - - result = response['GetTemplateResponse']['GetTemplateResult'] - self.assertTrue(result is not None) - template = result['TemplateBody'] - self.assertTrue(template is not 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) is not None) - - print "GetTemplate : OK" - - def testDescribeStackResource(self): - parameters = {'StackName': self.stack.stackname, - 'LogicalResourceId': self.logical_resource_name} - response = self.stack.heatclient.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 is not None) - desc_result = desc_resp['DescribeStackResourceResult'] - self.assertTrue(desc_result is not None) - res = desc_result['StackResourceDetail'] - self.assertTrue(res is not None) - - self.stack.check_stackid(res['StackId']) - - 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']) - is not None) - - self.assertEqual(res['ResourceStatusReason'], 'state changed') - - 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']) - is not None) - - self.assertTrue("AWS::CloudFormation::Init" in res['Metadata']) - - print "DescribeStackResource : OK" - - def testDescribeStackResources(self): - parameters = {'NameOrPid': self.stack.stackname, - 'LogicalResourceId': self.logical_resource_name} - response = self.stack.heatclient.describe_stack_resources(**parameters) - self.assertEqual(len(response), 1) - - res = response[0] - self.assertTrue(res is not None) - - self.stack.check_stackid(res.stack_id) - - 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, 'state changed') - - 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) - is not None) - - print "DescribeStackResources : OK" - - def testListStackResources(self): - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.list_stack_resources(**parameters) - self.assertEqual(len(response), 1) - - res = response[0] - self.assertTrue(res is not None) - - self.assertEqual(res.resource_status, self.logical_resource_status) - - self.assertEqual(res.resource_status_reason, 'state changed') - - 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) - is not None) - - print "ListStackResources : OK" - - def testValidateTemplate(self): - # Use stack.format_parameters to get the TemplateBody - params = self.stack.format_parameters() - val_params = {'TemplateBody': params['TemplateBody']} - response = self.stack.heatclient.validate_template(**val_params) - # Check the response contains all the expected paramter keys - templ_params = ['DBUsername', 'LinuxDistribution', 'InstanceType', - 'DBRootPassword', 'KeyName', 'DBPassword', 'DBName'] - - resp_params = [p.parameter_key for p in response.template_parameters] - for param in templ_params: - self.assertTrue(param in resp_params) - print "ValidateTemplate : OK" diff --git a/heat/tests/functional/test_CFN_API_UpdateStack.py b/heat/tests/functional/test_CFN_API_UpdateStack.py deleted file mode 100644 index 4dc86eac..00000000 --- a/heat/tests/functional/test_CFN_API_UpdateStack.py +++ /dev/null @@ -1,149 +0,0 @@ -# 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 -from nose.plugins.attrib import attr -import unittest - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'api', 'cfn', 'F17']) -class CfnApiUpdateStackFunctionalTest(unittest.TestCase): - ''' - This test launches a wordpress stack then attempts to verify - correct operation of all the heat CFN API UpdateStack action - - This is a separate test from the main CfnApiFunctionalTest - because part of the test replaces the instance, making it - quite hard to avoid breaking things for the other tests - ''' - @classmethod - def setupAll(cls): - print "SETUPALL" - template = 'WordPress_Single_Instance.template' - - cls.instance_type = 'm1.xlarge' - cls.db_user = 'dbuser' - cls.stack_paramstr = ';'.join(['InstanceType=%s' % cls.instance_type, - 'DBUsername=%s' % cls.db_user, - '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 - - cls.inst = CfnApiFunctions() - cls.stack = util.Stack(cls.inst, template, 'F17', 'x86_64', 'cfntools', - cls.stack_paramstr) - cls.WikiDatabase = util.Instance(cls.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" - cls.stack_status = "CREATE_COMPLETE" - except Exception as ex: - print "setupAll failed : %s" % ex - cls.stack.cleanup() - raise - - @classmethod - def teardownAll(cls): - print "TEARDOWNALL" - cls.stack.cleanup() - - def testUpdateStack(self): - # If we just update without changing the parameters - # this should result in the stack going to UPDATE_COMPLETE - # without replacing the instance - - physids = self.stack.instance_phys_ids() - self.assertEqual(len(physids), 1) - print "UPDATE start, instance IDs = %s" % physids - - self.stack.update() - self.logical_resource_status = "UPDATE_COMPLETE" - self.stack_status = "UPDATE_COMPLETE" - self.stack_status_reason = "Stack successfully updated" - - updpids = self.stack.instance_phys_ids() - self.assertEqual(len(updpids), 1) - print "UPDATE complete, instance IDs = %s" % updpids - - self.assertEqual(updpids[0], physids[0]) - print "UpdateStack : OK" - - def testUpdateStackReplace(self): - # Then if we change a template parameter, instance should get replaced - physids = self.stack.instance_phys_ids() - self.assertEqual(len(physids), 1) - print "UPDATE start, instance IDs = %s" % physids - - self.db_user = 'anewuser' - self.stack_paramstr = ';'.join(['InstanceType=%s' % self.instance_type, - 'DBUsername=%s' % self.db_user, - 'DBPassword=' + - os.environ['OS_PASSWORD']]) - self.stack.stack_paramstr = self.stack_paramstr - self.stack.update() - tries = 0 - while (tries <= 500): - pollids = self.stack.instance_phys_ids() - print "Waiting for Instance to be replaced %s/%s %s" %\ - (tries, 500, pollids) - if (len(pollids) == 2): - self.assertTrue(pollids[1] != physids[0]) - print "Instance replaced, new ID = %s" % pollids[1] - break - time.sleep(10) - tries += 1 - - # Check we didn't timeout - self.assertTrue(tries < 500) - - # Now use DescribeStacks to check the parameter is updated - parameters = {} - parameters['StackName'] = self.stack.stackname - response = self.stack.heatclient.describe_stacks(**parameters) - prefix = '/DescribeStacksResponse/DescribeStacksResult/Stacks/member' - # value for each key, then check the extracted value - param_prefix = prefix + '/Parameters/member[ParameterKey="DBUsername"]' - lookup_value = self.stack.response_xml_item(response, param_prefix, - "ParameterValue") - print "Updated, got DBUsername=%s" % lookup_value - self.assertEqual(lookup_value, self.db_user) - - # Now we have to Create a new Instance object and wait for - # provisioning to be complete, or the next test will fail - self.WikiDatabase = util.Instance(self.inst, - self.logical_resource_name) - self.WikiDatabase.wait_for_boot() - self.WikiDatabase.check_cfntools() - self.WikiDatabase.wait_for_provisioning() - print "Update completed, instance rebuild complete" - - self.assertTrue(self.WikiDatabase.file_present - ('/etc/wordpress/wp-config.php')) - print "Wordpress installation detected after update" - - print "UpdateStack (Replace) : OK" diff --git a/heat/tests/functional/test_HAProxy_Single_Instance.py b/heat/tests/functional/test_HAProxy_Single_Instance.py deleted file mode 100644 index d1f49ef6..00000000 --- a/heat/tests/functional/test_HAProxy_Single_Instance.py +++ /dev/null @@ -1,87 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'haproxy', 'F17', - 'HAProxy_Single_Instance.template']) -class HAProxyFunctionalTest(unittest.TestCase): - def setUp(self): - # The HAProxy template somewhat un-usefully load-balances a single - # server, so we launch a wordpress stack and stick haproxy in front - wp_template = 'WordPress_Single_Instance.template' - - wp_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, wp_template, 'F17', 'x86_64', 'cfntools', - wp_paramstr) - self.WikiDatabase = util.Instance(self, 'WikiDatabase') - - self.hap_stack = None - - def tearDown(self): - self.stack.cleanup() - if self.hap_stack: - self.hap_stack.cleanup() - - def test_instance(self): - self.stack.create() - self.WikiDatabase.wait_for_boot() - self.WikiDatabase.check_cfntools() - self.WikiDatabase.wait_for_provisioning() - - # ensure wordpress was installed by checking for expected - # configuration file over ssh - 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)) - - # So wordpress instance is up, we now launch the HAProxy instance - # and prove wordpress is accessable via the proxy instance IP - hap_stackname = 'hap_teststack' - hap_template = 'HAProxy_Single_Instance.template' - hap_paramstr = ';'.join(['InstanceType=m1.xlarge', - "Server1=%s:80" % self.WikiDatabase.ip]) - - self.hap_stack = util.Stack(self, hap_template, - 'F17', 'x86_64', 'cfntools', - hap_paramstr, stackname=hap_stackname) - self.LoadBalancerInstance = util.Instance(self, 'LoadBalancerInstance', - hap_stackname) - self.hap_stack.create() - self.LoadBalancerInstance.wait_for_boot() - self.LoadBalancerInstance.check_cfntools() - self.LoadBalancerInstance.wait_for_provisioning() - - # Verify the output URL parses as expected, ie check that - # the wordpress installation is operational via the proxy - proxy_ip = self.hap_stack.get_stack_output("PublicIp") - print "Got haproxy stack output PublicIp=%s, verifying" % proxy_ip - proxy_url = "http://%s/wordpress" % proxy_ip - self.assertTrue(ver.verify_wordpress(proxy_url)) diff --git a/heat/tests/functional/test_OpenShift_Prebuilt_JEOS.py b/heat/tests/functional/test_OpenShift_Prebuilt_JEOS.py deleted file mode 100644 index 598ac88c..00000000 --- a/heat/tests/functional/test_OpenShift_Prebuilt_JEOS.py +++ /dev/null @@ -1,65 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr -import unittest - - -@attr(speed='slow') -@attr(tag=['func', 'openshift', 'F16-openshift', - 'OpenShift_Prebuilt_JEOS.template']) -class OpenShiftFunctionalTest(unittest.TestCase): - - def tearDown(self): - self.stack.cleanup() - - def setUp(self): - template = 'OpenShift_Prebuilt_JEOS.template' - stack_paramstr = ';'.join(['InstanceType=m1.xlarge']) - - self.stack = util.Stack(self, template, 'F16', 'x86_64', - 'cfntools-openshift', stack_paramstr) - - self.Node = util.Instance(self, 'OpenShiftNodeServer') - self.Broker = util.Instance(self, 'OpenShiftBrokerServer') - - def test_instance(self): - self.stack.create() - self.Broker.wait_for_boot() - self.Node.wait_for_boot() - self.Node.check_cfntools() - self.Broker.check_cfntools() - self.Node.wait_for_provisioning() - self.Broker.wait_for_provisioning() - - # ensure wordpress was installed by checking for expected - # configuration file over ssh - self.assertTrue(self.Broker.file_present - ('/etc/sysconfig/stickshift-broker')) - print 'OpenShift installation detected' - - # must change ip lookup so apache rewrite works properly - openshift_host = 'hello-admin.example.com' - util.add_host(self.Broker.ip, openshift_host) - - # Verify the output URL parses as expected, ie check that - # the openshift installation is operational with the deployed hello app - stack_url = 'https://' + openshift_host - print 'Verifying URL=%s' % stack_url - ver = verify.VerifyStack() - self.assertTrue(ver.verify_openshift(stack_url)) - - util.remove_host(self.Broker.ip, openshift_host) diff --git a/heat/tests/functional/test_WordPress_2_Instances.py b/heat/tests/functional/test_WordPress_2_Instances.py deleted file mode 100644 index 802d36a0..00000000 --- a/heat/tests/functional/test_WordPress_2_Instances.py +++ /dev/null @@ -1,63 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr - -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', '2instance', 'ebs', 'F17', - 'WordPress_2_Instances.template']) -class WordPress2Instances(unittest.TestCase): - def setUp(self): - template = 'WordPress_2_Instances.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.DatabaseServer = util.Instance(self, 'DatabaseServer') - self.WebServer = util.Instance(self, 'WebServer') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - - self.DatabaseServer.wait_for_boot() - self.DatabaseServer.check_cfntools() - self.DatabaseServer.wait_for_provisioning() - - self.WebServer.wait_for_boot() - self.WebServer.check_cfntools() - self.WebServer.wait_for_provisioning() - - # ensure wordpress was installed - self.assertTrue(self.WebServer.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)) diff --git a/heat/tests/functional/test_WordPress_2_Instances_With_EBS.py b/heat/tests/functional/test_WordPress_2_Instances_With_EBS.py deleted file mode 100644 index 8f2c7215..00000000 --- a/heat/tests/functional/test_WordPress_2_Instances_With_EBS.py +++ /dev/null @@ -1,75 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr - -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', '2instance', 'ebs', 'F17', - 'WordPress_2_Instances_With_EBS.template']) -class WordPress2InstancesWithEBS(unittest.TestCase): - def setUp(self): - template = 'WordPress_2_Instances_With_EBS.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - - self.WikiDatabase = util.Instance(self, 'WikiDatabase') - self.WebServer = util.Instance(self, 'WebServer') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - - self.WikiDatabase.wait_for_boot() - self.WikiDatabase.check_cfntools() - self.WikiDatabase.wait_for_provisioning() - - self.WebServer.wait_for_boot() - self.WebServer.check_cfntools() - self.WebServer.wait_for_provisioning() - - # ensure wordpress was installed - self.assertTrue(self.WebServer.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)) - - # Check EBS volume is present and mounted - stdin, stdout, sterr = self.WikiDatabase.exec_command( - 'grep vdc /proc/mounts') - result = stdout.readlines().pop().rstrip() - self.assertTrue(len(result)) - print "Checking EBS volume is attached : %s" % result - devname = result.split()[0] - self.assertEqual(devname, '/dev/vdc1') - mountpoint = result.split()[1] - self.assertEqual(mountpoint, '/var/lib/mysql') diff --git a/heat/tests/functional/test_WordPress_2_Instances_With_EBS_EIP.py b/heat/tests/functional/test_WordPress_2_Instances_With_EBS_EIP.py deleted file mode 100644 index fd111a3d..00000000 --- a/heat/tests/functional/test_WordPress_2_Instances_With_EBS_EIP.py +++ /dev/null @@ -1,92 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'eip', 'ebs', 'F17']) -class WordPress2EBSEIPFunctionalTest(unittest.TestCase): - def setUp(self): - self.template = 'WordPress_2_Instances_With_EBS_EIP.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, self.template, - 'F17', 'x86_64', 'cfntools', - stack_paramstr) - - self.webserver = util.Instance(self, 'WebServer') - - self.database = util.Instance(self, 'WikiDatabase') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - - self.webserver.wait_for_boot() - self.webserver.check_cfntools() - self.webserver.wait_for_provisioning() - self.webserver.check_user_data(self.template) - - self.database.wait_for_boot() - self.database.check_cfntools() - self.database.wait_for_provisioning() - self.database.check_user_data(self.template) - - # Check wordpress installation - wp_config_file = '/etc/wordpress/wp-config.php' - self.assertTrue(self.webserver.file_present(wp_config_file), - 'wp-config.php is not present') - - # Check mysql is installed and running - stdin, stdout, sterr = self.database.exec_command( - 'systemctl status mysqld.service') - result = stdout.read() - self.assertTrue('running' in result, 'mysql service is not running') - - # Check EBS volume is present and mounted - stdin, stdout, sterr = self.database.exec_command( - 'grep vdc /proc/mounts') - lines = stdout.readlines() - self.assertTrue(len(lines) > 0) - result = lines.pop().strip() - self.assertTrue(len(result) > 0) - print "Checking EBS volume is attached: %s" % result - self.assertTrue('/dev/vdc1' in result) - self.assertTrue('/var/lib/mysql' in result) - - # Check the floating IPs - self.assertTrue( - self.webserver.floating_ip_present(), - 'WebServer instance does not have a floating IP assigned') - self.assertTrue( - self.database.floating_ip_present(), - 'WikiDatabase instance does not have a floating IP assigned') - - # Check wordpress is running and acessible at the correct URL - 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), - 'Wordpress is not accessible at: %s' % stack_url) diff --git a/heat/tests/functional/test_WordPress_Composed_Instances.py b/heat/tests/functional/test_WordPress_Composed_Instances.py deleted file mode 100644 index 455246f0..00000000 --- a/heat/tests/functional/test_WordPress_Composed_Instances.py +++ /dev/null @@ -1,66 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'composed', 'F17', - 'WordPress_Composed_Instances.template']) -class WordPressComposedInstancesFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_Composed_Instances.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - - self.WebServer = util.Instance(self, 'WebServer') - - self.MySqlDatabaseServer = util.Instance( - self, - 'DatabaseTemplate.MySqlDatabaseServer') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - - self.WebServer.wait_for_boot() - self.MySqlDatabaseServer.wait_for_boot() - - self.WebServer.check_cfntools() - self.WebServer.wait_for_provisioning() - - self.MySqlDatabaseServer.check_cfntools() - self.MySqlDatabaseServer.wait_for_provisioning() - - self.assertTrue(self.WebServer.file_present - ('/etc/wordpress/wp-config.php')) - print 'Wordpress installation detected.' - - # Verify the output URL parses as expected, ie check that - # thewordpress installation is operational - stack_url = self.stack.get_stack_output("WebsiteURL") - print "Verifying stack output from WebsiteUrl=(%s)." % stack_url - ver = verify.VerifyStack() - self.assertTrue(ver.verify_wordpress(stack_url)) diff --git a/heat/tests/functional/test_WordPress_Single_Instance.py b/heat/tests/functional/test_WordPress_Single_Instance.py deleted file mode 100644 index 2dd37b7b..00000000 --- a/heat/tests/functional/test_WordPress_Single_Instance.py +++ /dev/null @@ -1,56 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'F17', 'WordPress_Single_Instance.template']) -class WordPressFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_Single_Instance.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WikiDatabase = util.Instance(self, 'WikiDatabase') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - self.WikiDatabase.wait_for_boot() - self.WikiDatabase.check_cfntools() - self.WikiDatabase.wait_for_provisioning() - - # ensure wordpress was installed by checking for expected - # configuration file over ssh - 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)) diff --git a/heat/tests/functional/test_WordPress_Single_Instance_Boto.py b/heat/tests/functional/test_WordPress_Single_Instance_Boto.py deleted file mode 100644 index 52a770f6..00000000 --- a/heat/tests/functional/test_WordPress_Single_Instance_Boto.py +++ /dev/null @@ -1,59 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'F17', - 'WordPress_Single_Instance.template', 'boto']) -class WordPressBotoFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_Single_Instance.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.StackBoto(self, template, - 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WikiDatabase = util.Instance(self, 'WikiDatabase') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - - self.WikiDatabase.wait_for_boot() - self.WikiDatabase.check_cfntools() - self.WikiDatabase.wait_for_provisioning() - - # ensure wordpress was installed by checking for expected - # configuration file over ssh - 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)) diff --git a/heat/tests/functional/test_WordPress_Single_Instance_With_EBS.py b/heat/tests/functional/test_WordPress_Single_Instance_With_EBS.py deleted file mode 100644 index 325d8508..00000000 --- a/heat/tests/functional/test_WordPress_Single_Instance_With_EBS.py +++ /dev/null @@ -1,68 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr - -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'ebs', 'F17', - 'WordPress_Single_Instance_With_EBS.template']) -class WordPressSingleEBSFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_Single_Instance_With_EBS.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WikiDatabase = util.Instance(self, 'WikiDatabase') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - self.WikiDatabase.wait_for_boot() - self.WikiDatabase.check_cfntools() - self.WikiDatabase.wait_for_provisioning() - - # ensure wordpress was installed - 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)) - - # Check EBS volume is present and mounted - stdin, stdout, sterr = self.WikiDatabase.exec_command( - 'grep vdc /proc/mounts') - result = stdout.readlines().pop().rstrip() - self.assertTrue(len(result)) - print "Checking EBS volume is attached : %s" % result - devname = result.split()[0] - self.assertEqual(devname, '/dev/vdc1') - mountpoint = result.split()[1] - self.assertEqual(mountpoint, '/var/lib/mysql') diff --git a/heat/tests/functional/test_WordPress_Single_Instance_With_EBS_EIP.py b/heat/tests/functional/test_WordPress_Single_Instance_With_EBS_EIP.py deleted file mode 100644 index d4f733b8..00000000 --- a/heat/tests/functional/test_WordPress_Single_Instance_With_EBS_EIP.py +++ /dev/null @@ -1,91 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr - -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'WordPress_Single_Instance_With_EBS_EIP.template', 'F17', - 'wordpress', 'ebs', 'eip']) -class WordPressEBSEIPFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_Single_Instance_With_EBS_EIP.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WikiServer = util.Instance(self, 'WikiServer') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - self.WikiServer.wait_for_boot() - self.WikiServer.check_cfntools() - self.WikiServer.wait_for_provisioning() - - # ensure wordpress was installed - self.assertTrue(self.WikiServer.file_present - ('/etc/wordpress/wp-config.php')) - print "Wordpress installation detected" - - # 2. check floating ip assignment - if len(self.stack.novaclient.floating_ips.list()) == 0: - print 'zero floating IPs detected' - self.assertTrue(False) - else: - found = 0 - mylist = self.stack.novaclient.floating_ips.list() - for item in mylist: - if item.instance_id == self.stack.instance_phys_ids()[0]: - print 'floating IP found', item.ip - found = 1 - break - self.assertEqual(found, 1) - - # Verify the output URL parses as expected, ie check that - # the wordpress installation is operational - # Note that the WebsiteURL uses the non-EIP address - 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)) - - # Then the InstanceIPAddress is the EIP address - # which should also render the wordpress page - stack_eip = self.stack.get_stack_output("InstanceIPAddress") - eip_url = "http://%s/wordpress" % stack_eip - print "Got stack output InstanceIPAddress=%s, verifying url %s" %\ - (stack_eip, eip_url) - self.assertTrue(ver.verify_wordpress(eip_url)) - - # Check EBS volume is present and mounted - stdin, stdout, sterr = self.WikiServer.exec_command( - 'grep vdc /proc/mounts') - result = stdout.readlines().pop().rstrip() - self.assertTrue(len(result)) - print "Checking EBS volume is attached : %s" % result - devname = result.split()[0] - self.assertEqual(devname, '/dev/vdc1') - mountpoint = result.split()[1] - self.assertEqual(mountpoint, '/var/lib/mysql') diff --git a/heat/tests/functional/test_WordPress_Single_Instance_With_EIP.py b/heat/tests/functional/test_WordPress_Single_Instance_With_EIP.py deleted file mode 100644 index fb18d87d..00000000 --- a/heat/tests/functional/test_WordPress_Single_Instance_With_EIP.py +++ /dev/null @@ -1,80 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr - -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'eip', 'F17', - 'WordPress_Single_Instance_With_EIP.template']) -class WordPressEIPFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_Single_Instance_With_EIP.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WebServer = util.Instance(self, 'WebServer') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - self.WebServer.wait_for_boot() - self.WebServer.check_cfntools() - self.WebServer.wait_for_provisioning() - - # ensure wordpress was installed - self.assertTrue(self.WebServer.file_present - ('/etc/wordpress/wp-config.php')) - print "Wordpress installation detected" - - # 2. check floating ip assignment - if len(self.stack.novaclient.floating_ips.list()) == 0: - print 'zero floating IPs detected' - self.assertTrue(False) - else: - found = 0 - mylist = self.stack.novaclient.floating_ips.list() - for item in mylist: - if item.instance_id == self.stack.instance_phys_ids()[0]: - print 'floating IP found', item.ip - found = 1 - break - self.assertEqual(found, 1) - - # Verify the output URL parses as expected, ie check that - # the wordpress installation is operational - # Note that the WebsiteURL uses the non-EIP address - 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)) - - # Then the InstanceIPAddress is the EIP address - # which should also render the wordpress page - stack_eip = self.stack.get_stack_output("InstanceIPAddress") - eip_url = "http://%s/wordpress" % stack_eip - print "Got stack output InstanceIPAddress=%s, verifying url %s" %\ - (stack_eip, eip_url) - self.assertTrue(ver.verify_wordpress(eip_url)) diff --git a/heat/tests/functional/test_WordPress_Single_Instance_With_HA.py b/heat/tests/functional/test_WordPress_Single_Instance_With_HA.py deleted file mode 100644 index e6a861b1..00000000 --- a/heat/tests/functional/test_WordPress_Single_Instance_With_HA.py +++ /dev/null @@ -1,74 +0,0 @@ -# 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 util -from nose.plugins.attrib import attr -import unittest -import os -import time - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'HA', 'F17', - 'WordPress_Single_Instance_With_HA.template']) -class HaFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_Single_Instance_With_HA.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WikiDatabase = util.Instance(self, 'WikiDatabase') - - def tearDown(self): - self.stack.cleanup() - - def service_is_running(self, name): - stdin, stdout, sterr = \ - self.WikiDatabase.exec_command( - 'systemctl status %s' % name + '.service') - - lines = stdout.readlines() - for line in lines: - if 'Active: active (running)' in line: - return True - return False - - def test_instance(self): - self.stack.create() - self.WikiDatabase.wait_for_boot() - self.WikiDatabase.check_cfntools() - self.WikiDatabase.wait_for_provisioning() - - # ensure wordpress was installed - self.assertTrue(self.WikiDatabase.file_present - ('/etc/wordpress/wp-config.php')) - print "Wordpress installation detected" - - # check the httpd service is running - self.assertTrue(self.service_is_running('httpd')) - - # kill httpd - self.WikiDatabase.exec_sudo_command('systemctl stop httpd.service') - - # check that httpd service recovers - # should take less than 60 seconds, but no worse than 70 seconds - tries = 0 - while not self.service_is_running('httpd'): - tries += 1 - self.assertTrue(tries < 8) - time.sleep(10) diff --git a/heat/tests/functional/test_WordPress_Single_Instance_With_IHA.py b/heat/tests/functional/test_WordPress_Single_Instance_With_IHA.py deleted file mode 100644 index 8a100cfd..00000000 --- a/heat/tests/functional/test_WordPress_Single_Instance_With_IHA.py +++ /dev/null @@ -1,107 +0,0 @@ -# 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 -from nose.plugins.attrib import attr -import unittest -import time - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'HA', 'F17', - 'WordPress_Single_Instance_With_IHA.template']) -class WordPressIHAFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_Single_Instance_With_IHA.template' - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WikiDatabase = util.Instance(self, 'WikiDatabase') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - self.WikiDatabase.wait_for_boot() - self.WikiDatabase.check_cfntools() - self.WikiDatabase.wait_for_provisioning() - - # ensure wordpress was installed by checking for expected - # configuration file over ssh - wp_file = '/usr/share/wordpress/wp-config.php' - self.assertTrue(self.WikiDatabase.file_present(wp_file)) - 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)) - - # Save the instance physical resource ID - phys_res_ids = self.stack.instance_phys_ids() - self.assertEqual(len(phys_res_ids), 1) - print "Shutting down instance ID = %s" % phys_res_ids[0] - - # Now shut down the instance via SSH, and wait for HA to reprovision - # note it may not come back on the same IP - stdin, stdout, stderr =\ - self.WikiDatabase.exec_sudo_command('/sbin/halt') - - # Now poll the stack events, as the current WikiDatabase instance - # should be replaced with a new one - # we can then prove the physical resource ID is different, and that - # wordpress is accessible on the new WikiDatabase instance - tries = 0 - while (tries <= 500): - pollids = self.stack.instance_phys_ids() - print "Waiting for Instance to be replaced %s/%s %s" %\ - (tries, 500, pollids) - if (len(pollids) == 2): - self.assertTrue(pollids[1] != phys_res_ids[0]) - print "Instance replaced, new ID = %s" % pollids[1] - break - time.sleep(10) - tries += 1 - - # Check we didn't timeout - self.assertTrue(tries < 500) - - # Create a new Instance object and wait for boot - # It seems to be necessary to close the old WikiDatabase - # instance SSH transport, or paramiko throws an exception - # when the SSH daemon starts on the new instance if it comes - # up on the same IP address, suspect paramiko bug, TBC - self.WikiDatabase.close_ssh_client() - self.WikiDatabaseNew = util.Instance(self, 'WikiDatabase') - self.WikiDatabaseNew.wait_for_boot() - self.WikiDatabaseNew.check_cfntools() - self.WikiDatabaseNew.wait_for_provisioning() - - # Re-check wordpress installation as for the first instance - self.assertTrue(self.WikiDatabaseNew.file_present(wp_file)) - print "Wordpress installation detected on new instance" - - # 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 - self.assertTrue(ver.verify_wordpress(stack_url)) diff --git a/heat/tests/functional/test_WordPress_With_LB.py b/heat/tests/functional/test_WordPress_With_LB.py deleted file mode 100644 index f7a29ede..00000000 --- a/heat/tests/functional/test_WordPress_With_LB.py +++ /dev/null @@ -1,68 +0,0 @@ -# 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 util -import verify -from nose.plugins.attrib import attr -import unittest -import os - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'LB', 'F17', 'WordPress_With_LB.template']) -class WordPressWithLBFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_With_LB.template' - - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - - self.WikiServerOne = util.Instance(self, 'WikiServerOne') - self.LBInstance = util.Instance(self, 'LoadBalancer.LB_instance') - self.MySqlDatabaseServer = util.Instance( - self, - 'DatabaseServer.MySqlDatabaseServer') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - - self.WikiServerOne.wait_for_boot() - self.LBInstance.wait_for_boot() - self.MySqlDatabaseServer.wait_for_boot() - - self.WikiServerOne.check_cfntools() - self.LBInstance.check_cfntools() - self.MySqlDatabaseServer.check_cfntools() - - self.WikiServerOne.wait_for_provisioning() - self.LBInstance.wait_for_provisioning() - self.MySqlDatabaseServer.wait_for_provisioning() - - self.assertTrue(self.WikiServerOne.file_present - ('/etc/wordpress/wp-config.php')) - print 'Wordpress installation detected.' - - # Verify the output URL parses as expected, ie check that - # thewordpress installation is operational - stack_url = self.stack.get_stack_output("WebsiteURL") - print "Verifying stack output from WebsiteUrl=(%s)." % stack_url - ver = verify.VerifyStack() - self.assertTrue(ver.verify_wordpress(stack_url)) diff --git a/heat/tests/functional/test_WordPress_With_RDS.py b/heat/tests/functional/test_WordPress_With_RDS.py deleted file mode 100644 index 7c66ef8f..00000000 --- a/heat/tests/functional/test_WordPress_With_RDS.py +++ /dev/null @@ -1,69 +0,0 @@ -# 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 -from nose.plugins.attrib import attr -import unittest - - -@attr(speed='slow') -@attr(tag=['func', 'wordpress', 'RDS', 'F17', - 'WordPress_With_RDS.template']) -class WordPressRDSFunctionalTest(unittest.TestCase): - def setUp(self): - template = 'WordPress_With_RDS.template' - stack_paramstr = ';'.join(['InstanceType=m1.xlarge', - 'DBUsername=dbuser', - 'DBPassword=' + os.environ['OS_PASSWORD']]) - - self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools', - stack_paramstr) - self.WebServer = util.Instance(self, 'WebServer') - - def tearDown(self): - self.stack.cleanup() - - def test_instance(self): - self.stack.create() - self.WebServer.wait_for_boot() - self.WebServer.check_cfntools() - self.WebServer.wait_for_provisioning() - - # ensure wordpress was installed by checking for expected - # configuration file over ssh - wp_file = '/usr/share/wordpress/wp-config.php' - self.assertTrue(self.WebServer.file_present(wp_file)) - 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)) - - # Check the DB_HOST value in the wordpress config is sane - # ie not localhost, we don't have any way to get the IP of - # the RDS nested-stack instance so we can't do a proper verification - # Note there are two wp-config.php files, one under /etc and - # one under /usr/share, the template only seds the RDS instance - # IP into the /usr/share one, which seems to work but could be a - # template bug.. - stdin, stdout, sterr =\ - self.WebServer.ssh.exec_command('grep DB_HOST ' + wp_file) - result = stdout.readlines().pop().rstrip().split('\'') - print "Checking wordpress DB_HOST, got %s" % result[3] - self.assertTrue("localhost" != result[3]) diff --git a/heat/tests/functional/util.py b/heat/tests/functional/util.py deleted file mode 100644 index 48a3da9a..00000000 --- a/heat/tests/functional/util.py +++ /dev/null @@ -1,642 +0,0 @@ -# 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 -# under the License. - - -import os -import optparse -import paramiko -import subprocess -import hashlib -import email -import time # for sleep -import errno -import tempfile -import stat -import re -from pkg_resources import resource_string -from lxml import etree - -from nose.exc import SkipTest - -from glanceclient import client as glance_client -from keystoneclient.v2_0 import client as keystone_client -from novaclient.v1_1 import client as nova_client -import heat -from heat.common import template_format -from heat.engine import parser -from heat.cfn_client import client as heat_client -from heat.cfn_client import boto_client as heat_client_boto -from keystoneclient.v2_0 import client - -DEFAULT_STACKNAME = 'teststack' - - -# this test is in heat/tests/functional, so go up 3 dirs -basepath = os.path.join(heat.__path__[0], os.path.pardir) - - -class Instance(object): - def __init__(self, testcase, instance_name, stackname=DEFAULT_STACKNAME): - self.testcase = testcase - self.name = '%s.%s' % (stackname, instance_name) - - # during nose test execution this file will be imported even if - # the unit tag was specified - try: - os.environ['OS_AUTH_STRATEGY'] - except KeyError: - raise SkipTest('OS_AUTH_STRATEGY unset, skipping functional test') - - self.testcase.assertEqual(os.environ['OS_AUTH_STRATEGY'], - 'keystone', - 'keystone authentication required') - - self.creds = dict(username=os.environ['OS_USERNAME'], - password=os.environ['OS_PASSWORD'], - tenant=os.environ['OS_TENANT_NAME'], - auth_url=os.environ['OS_AUTH_URL'], - strategy=os.environ['OS_AUTH_STRATEGY']) - dbusername = 'testuser' - - self.novaclient = nova_client.Client(self.creds['username'], - self.creds['password'], - self.creds['tenant'], - self.creds['auth_url'], - service_type='compute') - - self.ssh = paramiko.SSHClient() - self.sftp = None - - self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - - self.ip = None - - def wait_for_boot(self): - tries = 0 - while self.ip is None: - servers = self.novaclient.servers.list() - for server in servers: - if server.name == self.name: - address = server.addresses - if address: - self.ip = address.items()[0][1][0]['addr'] - tries += 1 - self.testcase.assertTrue(tries < 150, 'Timed out') - time.sleep(10) - print 'Instance (%s) ip (%s) status (%s)' % (self.name, self.ip, - server.status) - - tries = 0 - while True: - try: - subprocess.check_output(['nc', '-z', self.ip, '22']) - except Exception: - print('Instance (%s) ip (%s) SSH not up yet, waiting...' % - (self.name, self.ip)) - tries += 1 - self.testcase.assertTrue(tries < 50, 'Timed out') - time.sleep(10) - else: - print 'Instance (%s) ip (%s) SSH detected.' % (self.name, - self.ip) - break - - tries = 0 - while True: - try: - tries += 1 - self.testcase.assertTrue(tries < 50, 'Timed out') - self.ssh.connect(self.ip, - username='ec2-user', - allow_agent=True, - look_for_keys=True, - password='password') - except paramiko.AuthenticationException: - print 'Authentication error' - time.sleep(2) - except Exception as e: - if e.errno != errno.EHOSTUNREACH: - raise - print('Instance (%s) ip (%s) connecting via SSH.' % - (self.name, self.ip)) - time.sleep(2) - else: - print('Instance (%s) ip (%s) connected via SSH.' % - (self.name, self.ip)) - break - self.sftp = self.ssh.open_sftp() - - tries = 0 - while True: - try: - self.sftp.stat('/var/lib/heat-cfntools/boot-finished') - except IOError as e: - tries += 1 - if e.errno == errno.ENOENT: - self.testcase.assertTrue(tries < 50, 'Timed out') - print("Instance (%s) ip (%s) not booted, waiting..." % - (self.name, self.ip)) - time.sleep(15) - else: - print e.errno - raise - else: - print("Instance (%s) ip (%s) finished booting." % - (self.name, self.ip)) - break - - def exec_sudo_command(self, cmd): - # Force a tty or sudo commands fail - channel = self.ssh.invoke_shell() - channel.sendall("sudo %s\n" % cmd) - channel.sendall('exit\n') - time.sleep(1) # necessary for sendall to complete - stdin = channel.makefile('wb') - stdout = channel.makefile('rb') - stderr = channel.makefile_stderr('rb') - return stdin, stdout, stderr - - def exec_command(self, cmd): - return self.ssh.exec_command(cmd) - - def exists(self): - servers = self.novaclient.servers.list() - for server in servers: - if server.name == self.name: - return True - return False - - def file_present(self, path): - print "Verifying file '%s' exists" % path - stdin, stdout, sterr = self.ssh.exec_command('ls "%s"' % path) - lines = stdout.readlines() - self.testcase.assertEqual(len(lines), 1) - result = lines.pop().rstrip() - return result == path - - def floating_ip_present(self): - floating_ips = self.novaclient.floating_ips.list() - for eip in floating_ips: - if self.ip == eip.fixed_ip: - return True - return False - - def check_cfntools(self): - stdin, stdout, stderr = \ - self.ssh.exec_command('cd /opt/aws/bin; sha1sum *') - files = stdout.readlines() - - cfn_tools_files = ['cfn-init', 'cfn-hup', 'cfn-signal', - 'cfn-get-metadata', 'cfn_helper.py'] - - cfntools = {} - for file in cfn_tools_files: - file_data = resource_string('heat_jeos', 'cfntools/' + file) - sha = hashlib.sha1(file_data).hexdigest() - cfntools[file] = sha - - # 1. make sure installed cfntools SHA match VM's version - for x in range(len(files)): - data = files.pop().split(' ') - cur_file = data[1].rstrip() - if cur_file in cfn_tools_files: - self.testcase.assertEqual(data[0], cfntools[cur_file]) - print 'Instance (%s) cfntools integrity verified.' % self.name - - def wait_for_provisioning(self): - print "Instance (%s) waiting for provisioning to complete." % self.name - tries = 0 - while True: - try: - self.sftp.stat('/var/lib/heat-cfntools/provision-finished') - except paramiko.SSHException as e: - print e - except IOError as e: - if e.errno != errno.ENOENT: - raise - else: - print "Instance (%s) provisioning completed." % self.name - return - - tries += 1 - self.testcase.assertTrue(tries < 50, 'Timed out') - print("Instance (%s) provisioning incomplete, waiting..." % - self.name) - time.sleep(15) - - def check_user_data(self, template_file): - return # until TODO is fixed - -# transport = self.ssh.get_transport() -# channel = transport.open_session() -# channel.get_pty() -# channel.invoke_shell() # sudo requires tty -# channel.sendall('sudo chmod 777 \ -# sudo chmod 777 /var/lib/cloud/instance/user-data.txt.i\n') -# time.sleep(1) # necessary for sendall to complete - - f = open(basepath + '/templates/' + template_file) - t = template_format.parse(f.read()) - f.close() - - template = parser.Template(t) - params = parser.Parameters('test', t, - {'KeyName': 'required_parameter', - 'DBUsername': self.dbusername, - 'DBPassword': self.creds['password']}) - - stack = parser.Stack(None, 'test', template, params) - parsed_t = stack.resolve_static_data(t) - remote_file = self.sftp.open('/var/lib/heat-cfntools/cfn-userdata') - remote_file_list = remote_file.read().split('\n') - remote_file_list_u = map(unicode, remote_file_list) - remote_file.close() - - # TODO: make server name generic - t_data = parsed_t['Resources']['WikiDatabase']['Properties'] - t_data = t_data['UserData']['Fn::Base64']['Fn::Join'].pop() - joined_t_data = ''.join(t_data) - t_data_list = joined_t_data.split('\n') - - self.testcase.assertEqual(t_data_list, remote_file_list_u) - - remote_file = self.sftp.open('/var/lib/cloud/instance/user-data.txt.i') - msg = email.message_from_file(remote_file) - remote_file.close() - - filepaths = { - 'cloud-config': basepath + '/heat/cloudinit/config', - 'part-handler.py': basepath + - '/heat/cloudinit/part-handler.py' - } - - # check multipart mime accuracy - for part in msg.walk(): - # multipart/* are just containers - if part.get_content_maintype() == 'multipart': - continue - - file = part.get_filename() - data = part.get_payload() - - if file in filepaths.keys(): - with open(filepaths[file]) as f: - self.testcase.assertEqual(data, f.read()) - - def close_ssh_client(self): - self.ssh.close() - - -class Stack(object): - - def __init__(self, testcase, template_file, distribution, arch, jeos_type, - stack_paramstr, stackname=DEFAULT_STACKNAME): - - self.testcase = testcase - self.stackname = stackname - self.template_file = template_file - self.distribution = distribution - self.stack_paramstr = stack_paramstr - - self.stack_id_re = re.compile("^arn:openstack:heat::[0-9a-z]{32}:" + - "stacks/" + self.stackname + - # Stack ID UUID in standard form - # as returned by uuid.uuid4() - "/[0-9a-f]{8}-" + - "[0-9a-f]{4}-" + - "[0-9a-f]{4}-" + - "[0-9a-f]{4}-" + - "[0-9a-f]{12}$") - - self.creds = dict(username=os.environ['OS_USERNAME'], - password=os.environ['OS_PASSWORD'], - tenant=os.environ['OS_TENANT_NAME'], - auth_url=os.environ['OS_AUTH_URL'], - strategy=os.environ['OS_AUTH_STRATEGY']) - - self.dbusername = 'testuser' - - self.testcase.assertEqual(os.environ['OS_AUTH_STRATEGY'], - 'keystone', - 'keystone authentication required') - - kc_creds = dict(username=os.environ['OS_USERNAME'], - password=os.environ['OS_PASSWORD'], - tenant_name=os.environ['OS_TENANT_NAME'], - auth_url=os.environ['OS_AUTH_URL']) - kc = keystone_client.Client(**kc_creds) - glance_url = kc.service_catalog.url_for(service_type='image', - endpoint_type='publicURL') - - version_string = '/v1' - if glance_url.endswith(version_string): - glance_url = glance_url[:-len(version_string)] - - auth_token = kc.auth_token - self.glanceclient = glance_client.Client(1, glance_url, - token=auth_token) - - self.prepare_jeos(distribution, arch, jeos_type) - - self.novaclient = nova_client.Client(self.creds['username'], - self.creds['password'], - self.creds['tenant'], - self.creds['auth_url'], - service_type='compute') - - self.heatclient = self._create_heat_client() - - def format_parameters(self): - self.keyname = self.novaclient.keypairs.list().pop().name - - self.testcase.assertTrue(self.heatclient) - - full_paramstr = ';'.join([self.stack_paramstr, - 'KeyName=' + self.keyname, - 'LinuxDistribution=' + self.distribution]) - template_params = optparse.Values({'parameters': full_paramstr}) - - # Format parameters and create the stack - parameters = {} - parameters['StackName'] = self.stackname - template_path = os.path.join(basepath, - 'templates', - self.template_file) - parameters['TemplateBody'] = open(template_path).read() - parameters.update(self.heatclient.format_parameters(template_params)) - return parameters - - def create(self): - parameters = self.format_parameters() - result = self.heatclient.create_stack(**parameters) - - self._check_create_result(result) - - alist = None - tries = 0 - - print 'Waiting for stack creation to be completed' - while self.get_state() == 'CREATE_IN_PROGRESS': - tries += 1 - self.testcase.assertTrue(tries < 150, 'Timed out') - time.sleep(10) - - self.testcase.assertEqual(self.get_state(), 'CREATE_COMPLETE') - - def update(self): - parameters = self.format_parameters() - result = self.heatclient.update_stack(**parameters) - - self._check_update_result(result) - - alist = None - tries = 0 - - print 'Waiting for stack update to be completed' - while self.get_state() == 'UPDATE_IN_PROGRESS': - tries += 1 - self.testcase.assertTrue(tries < 150, 'Timed out') - time.sleep(10) - - self.testcase.assertEqual(self.get_state(), 'UPDATE_COMPLETE') - - def _check_create_result(self, result): - # Check result looks OK - root = etree.fromstring(result) - create_list = root.xpath('/CreateStackResponse/CreateStackResult') - self.testcase.assertTrue(create_list) - self.testcase.assertEqual(len(create_list), 1) - stack_id = create_list[0].findtext('StackId') - self.testcase.assertTrue(stack_id is not None) - self.check_stackid(stack_id) - - def _check_update_result(self, result): - # Check result looks OK - root = etree.fromstring(result) - update_list = root.xpath('/UpdateStackResponse/UpdateStackResult') - self.testcase.assertTrue(update_list) - self.testcase.assertEqual(len(update_list), 1) - stack_id = update_list[0].findtext('StackId') - self.testcase.assertTrue(stack_id is not None) - self.check_stackid(stack_id) - - def check_stackid(self, stack_id): - print "Checking %s matches expected format" % (stack_id) - self.testcase.assertTrue(self.stack_id_re.match(stack_id) is not None) - - def _create_heat_client(self): - return heat_client.get_client('0.0.0.0', 8000, - self.creds['username'], - self.creds['password'], - self.creds['tenant'], - self.creds['auth_url'], - self.creds['strategy'], - None, None, False) - - def get_state(self): - stack_list = self.heatclient.list_stacks(StackName=self.stackname) - root = etree.fromstring(stack_list) - xpq = '//member[StackName="%s"]' - alist = root.xpath(xpq % (self.stackname)) - result = None - if len(alist): - item = alist.pop() - result = item.findtext("StackStatus") - if result and result.find('FAILED') >= 0: - print stack_list - return result - - def cleanup(self): - parameters = {'StackName': self.stackname} - self.heatclient.delete_stack(**parameters) - - print 'Waiting for stack deletion to be completed' - tries = 0 - while self.get_state() == 'DELETE_IN_PROGRESS': - tries += 1 - self.testcase.assertTrue(tries < 50, 'Timed out') - time.sleep(10) - - # final state for all stacks is DELETE_COMPLETE, but then they - # dissappear hence no result from list_stacks/get_state - # depending on timing, we could get either result here - end_state = self.get_state() - if end_state is not None: - self.testcase.assertEqual(end_state, 'DELETE_COMPLETE') - - def prepare_jeos(self, p_os, arch, type): - imagename = p_os + '-' + arch + '-' + type - - # skip creating jeos if image already available - if not self.poll_glance(imagename, False): - self.testcase.assertEqual(os.geteuid(), 0, - 'No JEOS found - run as root to create') - - # -d: debug, -G: register with glance - subprocess.call(['heat-jeos', '-d', '-G', 'create', imagename]) - - # Nose seems to change the behavior of the subprocess call to be - # asynchronous. So poll glance until image is registered. - self.poll_glance(self.glanceclient, imagename, True) - - def poll_glance(self, imagename, block): - image = None - tries = 0 - while image is None: - tries += 1 - self.testcase.assertTrue(tries < 50, 'Timed out') - if block: - time.sleep(15) - print "Checking glance for image registration" - imageslist = self.glanceclient.images.list( - filters={'name': imagename}) - image = next(imageslist, None) - if image: - print "Found image registration for %s" % imagename - # technically not necessary, but glance registers image - # before completely through with its operations - time.sleep(10) - return True - if not block: - break - return False - - def get_stack_output(self, output_key): - ''' - Extract a specified output from the DescribeStacks details - ''' - # Get the DescribeStacks result for this stack - parameters = {'StackName': self.stackname} - result = self.heatclient.describe_stacks(**parameters) - return self._find_stack_output(result, output_key) - - def _find_stack_output(self, result, output_key): - # Extract the OutputValue for the specified OutputKey - root = etree.fromstring(result) - output_list = root.xpath('//member[OutputKey="' + output_key + '"]') - output = output_list.pop() - value = output.findtext('OutputValue') - return value - - def instance_phys_ids(self): - events = self.heatclient.list_stack_events(StackName=self.stackname) - root = etree.fromstring(events) - xpq = ('//member[StackName="%s" and ' - 'ResourceStatus="CREATE_COMPLETE" and ' - 'ResourceType="AWS::EC2::Instance"]') - alist = root.xpath(xpq % self.stackname) - - return [elem.findtext('PhysicalResourceId') for elem in alist] - - def response_xml_item(self, response, prefix, key): - ''' - Extract response item via xpath prefix and key name - we expect the prefix to map to a single Element item - ''' - root = etree.fromstring(response) - output_list = root.xpath(prefix) - self.testcase.assertTrue(output_list) - self.testcase.assertEqual(len(output_list), 1) - output = output_list.pop() - value = output.findtext(key) - return value - - -class StackBoto(Stack): - ''' - Version of the Stack class which uses the boto client (hence AWS auth and - the CFN API). - ''' - def _check_create_result(self, result): - self.check_stackid(result) - - def _check_update_result(self, result): - self.check_stackid(result) - - def _create_heat_client(self): - # Connect to the keystone client with the supplied credentials - # and extract the ec2-credentials, so we can pass them into the - # boto client - keystone = client.Client(username=self.creds['username'], - password=self.creds['password'], - tenant_name=self.creds['tenant'], - auth_url=self.creds['auth_url']) - ksusers = keystone.users.list() - ksuid = [u.id for u in ksusers if u.name == self.creds['username']] - self.testcase.assertEqual(len(ksuid), 1) - - ec2creds = keystone.ec2.list(ksuid[0]) - self.testcase.assertEqual(len(ec2creds), 1) - self.testcase.assertTrue(ec2creds[0].access) - self.testcase.assertTrue(ec2creds[0].secret) - print "Got EC2 credentials from keystone" - - # most of the arguments passed to heat_client_boto are for - # compatibility with the non-boto client wrapper, and are - # actually ignored, only the port and credentials are used - return heat_client_boto.get_client('0.0.0.0', 8000, - self.creds['username'], - self.creds['password'], - self.creds['tenant'], - self.creds['auth_url'], - self.creds['strategy'], - None, None, False, - aws_access_key=ec2creds[0].access, - aws_secret_key=ec2creds[0].secret) - - def get_state(self): - stack_list = self.heatclient.list_stacks() - this = [s for s in stack_list if s.stack_name == self.stackname] - result = None - if len(this): - result = this[0].stack_status - return result - - def instance_phys_ids(self): - events = self.heatclient.list_stack_events(StackName=self.stackname) - - def match(e): - return (e.stack_name == self.stackname and - e.resource_status == "CREATE_COMPLETE" and - e.resource_type == "AWS::EC2::Instance") - - return [e.physical_resource_id for e in events if match(e)] - - def _find_stack_output(self, result, output_key): - self.testcase.assertEqual(len(result), 1) - - for o in result[0].outputs: - if o.key == output_key: - return o.value - - -def add_host(ip, hostname): - with open('/etc/hosts', 'a') as hostfile: - hostfile.write(ip + '\t' + hostname) - - -def remove_host(ip, hostname): - data = None - with open('/etc/hosts', 'r') as hostfile: - data = hostfile.readlines() - - perms = stat.S_IMODE(os.stat('/etc/hosts').st_mode) - - with tempfile.NamedTemporaryFile('w', dir='/etc', delete=False) as tmp: - for line in data: - if line.rstrip() == ip + '\t' + hostname: - continue - tmp.write(line) - os.chmod(tmp.name, perms) - os.rename(tmp.name, '/etc/hosts') diff --git a/heat/tests/functional/verify.py b/heat/tests/functional/verify.py deleted file mode 100644 index eab10700..00000000 --- a/heat/tests/functional/verify.py +++ /dev/null @@ -1,86 +0,0 @@ -# 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 -# under the License. - - -import urllib2 -import re -import time - -# Default retry timeout (seconds) and sleep-time -VERIFY_TIMEOUT = 300 -SLEEP_INTERVAL = 10 - - -class VerifyRetry(object): - def __init__(self, timeout=VERIFY_TIMEOUT, interval=SLEEP_INTERVAL): - """ - Decorator to wrap verify operations to retry until timeout - """ - self.timeout = timeout - self.interval = interval - - def __call__(self, f): - def fn(*args, **kwargs): - elapsed = 0 - result = False - while elapsed < self.timeout and not result: - result = f(*args, **kwargs) - if not result: - print "Failed verify, sleeping for %ss (%s/%s)" %\ - (self.interval, elapsed, self.timeout) - time.sleep(self.interval) - elapsed += self.interval - - return result - - return fn - - -class VerifyStack: - ''' - Class containing helper-functions to prove a stack resource or service - has been created correctly, e.g by accessing the service and checking - the result is as expected - ''' - - @VerifyRetry() - def verify_url(self, url, timeout, regex): - - print "Reading html from %s" % url - try: - content = urllib2.urlopen(url).read() - except IOError: - return False - - matches = re.findall(regex, content) - if len(matches): - print "VERIFY : looks OK!" - return True - else: - return False - - @VerifyRetry() - def verify_wordpress(self, url, timeout=VERIFY_TIMEOUT): - ''' - Verify the url provided has a functional wordpress installation - for now we simply scrape the page and do a regex for an expected - string - ''' - WORDPRESS_REGEX = "

Welcome to the famous five minute WordPress" - - return self.verify_url(url, timeout, WORDPRESS_REGEX) - - def verify_openshift(self, url, timeout=VERIFY_TIMEOUT): - OPENSHIFT_REGEX = "Welcome to OpenShift" - return self.verify_url(url, timeout, OPENSHIFT_REGEX) -- 2.45.2