'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WebServerGroup0 = util.Instance(self, 'WebServerGroup-0')
def tearDown(self):
inst = CfnApiFunctions()
cls.stack = util.Stack(inst, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
cls.WikiDatabase = util.Instance(inst, cls.logical_resource_name)
try:
update_time = self.stack.response_xml_item(response, prefix,
"LastUpdatedTime")
- self.assertTrue(self.time_re.match(update_time) != None)
+ 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) != None)
+ 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) != None)
+ self.assertTrue(self.description_re.match(description) is not None)
status_reason = self.stack.response_xml_item(response, prefix,
"StackStatusReason")
update_time = self.stack.response_xml_item(response, prefix,
"LastUpdatedTime")
- self.assertTrue(self.time_re.match(update_time) != None)
+ 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) != None)
+ 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) != None)
+ self.assertTrue(self.description_re.match(description) is not None)
status_reason = self.stack.response_xml_item(response, prefix,
"StackStatusReason")
self.assertEqual(stack_timeout, self.stack_timeout)
disable_rollback = self.stack.response_xml_item(response, prefix,
- "DisableRollback")
+ "DisableRollback")
self.assertEqual(disable_rollback, self.stack_disable_rollback)
# Create a dict to lookup the expected template parameters
template_parameters = {'DBUsername': 'dbuser',
- 'LinuxDistribution': 'F17',
- 'InstanceType': 'm1.xlarge',
- 'DBRootPassword': 'admin',
- 'KeyName': self.stack.keyname,
- 'DBPassword':
- os.environ['OS_PASSWORD'],
- 'DBName': 'wordpress'}
+ 'LinuxDistribution': 'F17',
+ 'InstanceType': 'm1.xlarge',
+ 'DBRootPassword': 'admin',
+ 'KeyName': self.stack.keyname,
+ 'DBPassword':
+ os.environ['OS_PASSWORD'],
+ 'DBName': 'wordpress'}
# We do a fully qualified xpath lookup to extract the paramter
# value for each key, then check the extracted value
lookup = '[ParameterKey="' + key + '" and ParameterValue="' +\
value + '"]'
lookup_value = self.stack.response_xml_item(response,
- param_prefix + lookup,
- "ParameterValue")
+ param_prefix + lookup,
+ "ParameterValue")
self.assertEqual(lookup_value, value)
# Then to a similar lookup to verify the Outputs section
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")
+ outputs_prefix + lookup,
+ "OutputValue")
self.assertEqual(lookup_value, expected_url)
print "DescribeStacks : OK"
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 + '"]'
+ '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) != None)
+ self.assertTrue(re.match("[0-9]*$", event_id) is not None)
resource_status = self.stack.response_xml_item(response, prefix,
- "ResourceStatus")
+ "ResourceStatus")
self.assertEqual(resource_status, self.logical_resource_status)
resource_type = self.stack.response_xml_item(response, prefix,
- "ResourceType")
+ "ResourceType")
self.assertEqual(resource_type, self.logical_resource_type)
update_time = self.stack.response_xml_item(response, prefix,
"Timestamp")
- self.assertTrue(self.time_re.match(update_time) != None)
+ self.assertTrue(self.time_re.match(update_time) is not None)
status_data = self.stack.response_xml_item(response, prefix,
- "ResourceStatusReason")
+ "ResourceStatusReason")
self.assertEqual(status_data, "state changed")
stack_name = self.stack.response_xml_item(response, prefix,
self.assertEqual(log_res_id, self.logical_resource_name)
phys_res_id = self.stack.response_xml_item(response, prefix,
- "PhysicalResourceId")
- self.assertTrue(self.phys_res_id_re.match(phys_res_id) != None)
+ "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 != None)
+ "ResourceProperties")
+ self.assertTrue(prop_json is not None)
prop = json.loads(prop_json)
self.assertEqual(prop["InstanceType"], "m1.xlarge")
template = self.stack.response_xml_item(response, prefix,
"TemplateBody")
json_load = template_format.parse(template)
- self.assertTrue(json_load != None)
+ 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) != None)
+ 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}
+ 'LogicalResourceId': self.logical_resource_name}
response = self.stack.heatclient.describe_stack_resource(**parameters)
prefix = '/DescribeStackResourceResponse/DescribeStackResourceResult'\
- + '/StackResourceDetail'
+ + '/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")
+ "ResourceStatus")
self.assertEqual(resource_status, self.logical_resource_status)
resource_type = self.stack.response_xml_item(response, prefix,
- "ResourceType")
+ "ResourceType")
self.assertEqual(resource_type, self.logical_resource_type)
update_time = self.stack.response_xml_item(response, prefix,
"LastUpdatedTimestamp")
- self.assertTrue(self.time_re.match(update_time) != None)
+ self.assertTrue(self.time_re.match(update_time) is not None)
status_reason = self.stack.response_xml_item(response, prefix,
- "ResourceStatusReason")
+ "ResourceStatusReason")
self.assertEqual(status_reason, "state changed")
stack_name = self.stack.response_xml_item(response, prefix,
self.assertEqual(log_res_id, self.logical_resource_name)
phys_res_id = self.stack.response_xml_item(response, prefix,
- "PhysicalResourceId")
- self.assertTrue(self.phys_res_id_re.match(phys_res_id) != None)
+ "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 != None)
+ 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}
+ 'LogicalResourceId': self.logical_resource_name}
response = self.stack.heatclient.describe_stack_resources(**parameters)
prefix = '/DescribeStackResourcesResponse/' +\
- 'DescribeStackResourcesResult/StackResources/member'
+ '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")
+ "ResourceStatus")
self.assertEqual(resource_status, self.logical_resource_status)
resource_type = self.stack.response_xml_item(response, prefix,
- "ResourceType")
+ "ResourceType")
self.assertEqual(resource_type, self.logical_resource_type)
update_time = self.stack.response_xml_item(response, prefix,
"Timestamp")
- self.assertTrue(self.time_re.match(update_time) != None)
+ self.assertTrue(self.time_re.match(update_time) is not None)
status_reason = self.stack.response_xml_item(response, prefix,
- "ResourceStatusReason")
+ "ResourceStatusReason")
self.assertEqual(status_reason, "state changed")
stack_name = self.stack.response_xml_item(response, prefix,
self.assertEqual(log_res_id, self.logical_resource_name)
phys_res_id = self.stack.response_xml_item(response, prefix,
- "PhysicalResourceId")
- self.assertTrue(self.phys_res_id_re.match(phys_res_id) != None)
+ "PhysicalResourceId")
+ self.assertTrue(self.phys_res_id_re.match(phys_res_id) is not None)
print "DescribeStackResources : OK"
parameters['StackName'] = self.stack.stackname
response = self.stack.heatclient.list_stack_resources(**parameters)
prefix = '/ListStackResourcesResponse/ListStackResourcesResult' +\
- '/StackResourceSummaries/member'
+ '/StackResourceSummaries/member'
resource_status = self.stack.response_xml_item(response, prefix,
- "ResourceStatus")
+ "ResourceStatus")
self.assertEqual(resource_status, self.logical_resource_status)
status_reason = self.stack.response_xml_item(response, prefix,
- "ResourceStatusReason")
+ "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) != None)
+ self.assertTrue(self.time_re.match(update_time) is not None)
resource_type = self.stack.response_xml_item(response, prefix,
- "ResourceType")
+ "ResourceType")
self.assertEqual(resource_type, self.logical_resource_type)
log_res_id = self.stack.response_xml_item(response, prefix,
self.assertEqual(log_res_id, self.logical_resource_name)
phys_res_id = self.stack.response_xml_item(response, prefix,
- "PhysicalResourceId")
- self.assertTrue(self.phys_res_id_re.match(phys_res_id) != None)
+ "PhysicalResourceId")
+ self.assertTrue(self.phys_res_id_re.match(phys_res_id) is not None)
print "ListStackResources : OK"
for param in templ_params:
lookup = '[ParameterKey="' + param + '"]'
val = self.stack.response_xml_item(response, prefix + lookup,
- 'ParameterKey')
+ 'ParameterKey')
self.assertEqual(param, val)
print "ValidateTemplate : OK"
inst = CfnApiFunctions()
cls.stack = util.StackBoto(inst, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
cls.WikiDatabase = util.Instance(inst, cls.logical_resource_name)
try:
# Extract the StackSummary for this stack
summary = [s for s in response
- if s.stack_name == self.stack.stackname]
+ if s.stack_name == self.stack.stackname]
self.assertEqual(len(summary), 1)
# Note the boto StackSummary object does not contain every item
self.assertEqual(type(summary[0].creation_time), datetime.datetime)
self.assertTrue(self.description_re.match(
- summary[0].template_description) != None)
+ summary[0].template_description) is not None)
self.assertEqual(summary[0].stack_name, self.stack.stackname)
# Extract the Stack object for this stack
stacks = [s for s in response
- if s.stack_name == self.stack.stackname]
+ 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)
- != None)
+ is not None)
self.assertEqual(stacks[0].stack_status_reason,
self.stack_status_reason)
# Create a dict to lookup the expected template parameters
template_parameters = {'DBUsername': 'dbuser',
- 'LinuxDistribution': 'F17',
- 'InstanceType': 'm1.xlarge',
- 'DBRootPassword': 'admin',
- 'KeyName': self.stack.keyname,
- 'DBPassword':
- os.environ['OS_PASSWORD'],
- 'DBName': 'wordpress'}
+ 'LinuxDistribution': 'F17',
+ 'InstanceType': 'm1.xlarge',
+ 'DBRootPassword': 'admin',
+ 'KeyName': self.stack.keyname,
+ 'DBPassword':
+ os.environ['OS_PASSWORD'],
+ 'DBName': 'wordpress'}
for key, value in template_parameters.iteritems():
# The parameters returned via the API include a couple
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]
+ 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) != None)
+ self.assertTrue(re.match("[0-9]*$", events[0].event_id) is not None)
self.assertEqual(events[0].resource_status,
self.logical_resource_status)
self.logical_resource_name)
self.assertTrue(self.phys_res_id_re.match(
- events[0].physical_resource_id) != None)
+ events[0].physical_resource_id) is not None)
# Check ResourceProperties, skip pending resolution of #245
properties = json.loads(events[0].resource_properties)
parameters = {}
parameters['StackName'] = self.stack.stackname
response = self.stack.heatclient.get_template(**parameters)
- self.assertTrue(response != None)
+ self.assertTrue(response is not None)
result = response['GetTemplateResponse']['GetTemplateResult']
- self.assertTrue(result != None)
+ self.assertTrue(result is not None)
template = result['TemplateBody']
- self.assertTrue(template != None)
+ 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) != None)
+ 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}
+ '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 != None)
+ self.assertTrue(desc_resp is not None)
desc_result = desc_resp['DescribeStackResourceResult']
- self.assertTrue(desc_result != None)
+ self.assertTrue(desc_result is not None)
res = desc_result['StackResourceDetail']
- self.assertTrue(res != None)
+ self.assertTrue(res is not None)
self.stack.check_stackid(res['StackId'])
# Note due to issue mentioned above timestamp is a string in this case
# not a datetime.datetime object
self.assertTrue(self.time_re.match(res['LastUpdatedTimestamp'])
- != None)
+ is not None)
self.assertEqual(res['ResourceStatusReason'], 'state changed')
self.assertEqual(res['LogicalResourceId'], self.logical_resource_name)
self.assertTrue(self.phys_res_id_re.match(res['PhysicalResourceId'])
- != None)
+ is not None)
self.assertTrue("AWS::CloudFormation::Init" in res['Metadata'])
def testDescribeStackResources(self):
parameters = {'NameOrPid': self.stack.stackname,
- 'LogicalResourceId': self.logical_resource_name}
+ 'LogicalResourceId': self.logical_resource_name}
response = self.stack.heatclient.describe_stack_resources(**parameters)
self.assertEqual(len(response), 1)
res = response[0]
- self.assertTrue(res != None)
+ self.assertTrue(res is not None)
self.stack.check_stackid(res.stack_id)
self.assertEqual(res.logical_resource_id, self.logical_resource_name)
self.assertTrue(self.phys_res_id_re.match(res.physical_resource_id)
- != None)
+ is not None)
print "DescribeStackResources : OK"
self.assertEqual(len(response), 1)
res = response[0]
- self.assertTrue(res != None)
+ self.assertTrue(res is not None)
self.assertEqual(res.resource_status, self.logical_resource_status)
self.assertEqual(res.logical_resource_id, self.logical_resource_name)
self.assertTrue(self.phys_res_id_re.match(res.physical_resource_id)
- != None)
+ is not None)
print "ListStackResources : OK"
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']])
+ 'DBUsername=%s' % cls.db_user,
+ 'DBPassword=' +
+ os.environ['OS_PASSWORD']])
cls.logical_resource_name = 'WikiDatabase'
cls.logical_resource_type = 'AWS::EC2::Instance'
cls.inst = CfnApiFunctions()
cls.stack = util.Stack(cls.inst, template, 'F17', 'x86_64', 'cfntools',
- cls.stack_paramstr)
+ cls.stack_paramstr)
cls.WikiDatabase = util.Instance(cls.inst, cls.logical_resource_name)
try:
self.db_user = 'anewuser'
self.stack_paramstr = ';'.join(['InstanceType=%s' % self.instance_type,
- 'DBUsername=%s' % self.db_user,
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=%s' % self.db_user,
+ 'DBPassword=' +
+ os.environ['OS_PASSWORD']])
self.stack.stack_paramstr = self.stack_paramstr
self.stack.update()
tries = 0
wp_template = 'WordPress_Single_Instance.template'
wp_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, wp_template, 'F17', 'x86_64', 'cfntools',
wp_paramstr)
stack_paramstr = ';'.join(['InstanceType=m1.xlarge'])
self.stack = util.Stack(self, template, 'F16', 'x86_64',
- 'cfntools-openshift', stack_paramstr)
+ 'cfntools-openshift', stack_paramstr)
self.Node = util.Instance(self, 'OpenShiftNodeServer')
self.Broker = util.Instance(self, 'OpenShiftBrokerServer')
template = 'WordPress_2_Instances.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.DatabaseServer = util.Instance(self, 'DatabaseServer')
self.WebServer = util.Instance(self, 'WebServer')
template = 'WordPress_2_Instances_With_EBS.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WikiDatabase = util.Instance(self, 'WikiDatabase')
self.WebServer = util.Instance(self, 'WebServer')
# Check EBS volume is present and mounted
stdin, stdout, sterr = self.WikiDatabase.exec_command(
- 'grep vdc /proc/mounts')
+ 'grep vdc /proc/mounts')
result = stdout.readlines().pop().rstrip()
self.assertTrue(len(result))
print "Checking EBS volume is attached : %s" % result
self.template = 'WordPress_2_Instances_With_EBS_EIP.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, self.template,
'F17', 'x86_64', 'cfntools',
# 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')
+ 'wp-config.php is not present')
# Check mysql is installed and running
stdin, stdout, sterr = self.database.exec_command(
self.assertTrue('/var/lib/mysql' in result)
# Check the floating IPs
- self.assertTrue(self.webserver.floating_ip_present(),
+ self.assertTrue(
+ self.webserver.floating_ip_present(),
'WebServer instance does not have a floating IP assigned')
- self.assertTrue(self.database.floating_ip_present(),
+ 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),
+ self.assertTrue(
+ ver.verify_wordpress(stack_url),
'Wordpress is not accessible at: %s' % stack_url)
template = 'WordPress_Composed_Instances.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WebServer = util.Instance(self, 'WebServer')
- self.MySqlDatabaseServer = util.Instance(self,
- 'DatabaseTemplate.MySqlDatabaseServer')
+ self.MySqlDatabaseServer = util.Instance(
+ self,
+ 'DatabaseTemplate.MySqlDatabaseServer')
def tearDown(self):
self.stack.cleanup()
'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WikiDatabase = util.Instance(self, 'WikiDatabase')
def tearDown(self):
template = 'WordPress_Single_Instance.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.StackBoto(self, template,
'F17', 'x86_64', 'cfntools',
template = 'WordPress_Single_Instance_With_EBS.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WikiDatabase = util.Instance(self, 'WikiDatabase')
def tearDown(self):
# Check EBS volume is present and mounted
stdin, stdout, sterr = self.WikiDatabase.exec_command(
- 'grep vdc /proc/mounts')
+ 'grep vdc /proc/mounts')
result = stdout.readlines().pop().rstrip()
self.assertTrue(len(result))
print "Checking EBS volume is attached : %s" % result
template = 'WordPress_Single_Instance_With_EBS_EIP.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WikiServer = util.Instance(self, 'WikiServer')
def tearDown(self):
# Check EBS volume is present and mounted
stdin, stdout, sterr = self.WikiServer.exec_command(
- 'grep vdc /proc/mounts')
+ 'grep vdc /proc/mounts')
result = stdout.readlines().pop().rstrip()
self.assertTrue(len(result))
print "Checking EBS volume is attached : %s" % result
template = 'WordPress_Single_Instance_With_EIP.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WebServer = util.Instance(self, 'WebServer')
def tearDown(self):
template = 'WordPress_Single_Instance_With_HA.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WikiDatabase = util.Instance(self, 'WikiDatabase')
def tearDown(self):
def setUp(self):
template = 'WordPress_Single_Instance_With_IHA.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WikiDatabase = util.Instance(self, 'WikiDatabase')
def tearDown(self):
template = 'WordPress_With_LB.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WikiServerOne = util.Instance(self, 'WikiServerOne')
self.LBInstance = util.Instance(self, 'LoadBalancer.LB_instance')
- self.MySqlDatabaseServer = util.Instance(self,
- 'DatabaseServer.MySqlDatabaseServer')
+ self.MySqlDatabaseServer = util.Instance(
+ self,
+ 'DatabaseServer.MySqlDatabaseServer')
def tearDown(self):
self.stack.cleanup()
def setUp(self):
template = 'WordPress_With_RDS.template'
stack_paramstr = ';'.join(['InstanceType=m1.xlarge',
- 'DBUsername=dbuser',
- 'DBPassword=' + os.environ['OS_PASSWORD']])
+ 'DBUsername=dbuser',
+ 'DBPassword=' + os.environ['OS_PASSWORD']])
self.stack = util.Stack(self, template, 'F17', 'x86_64', 'cfntools',
- stack_paramstr)
+ stack_paramstr)
self.WebServer = util.Instance(self, 'WebServer')
def tearDown(self):
'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'])
+ 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.creds['password'],
+ self.creds['tenant'],
+ self.creds['auth_url'],
+ service_type='compute')
self.ssh = paramiko.SSHClient()
self.testcase.assertTrue(tries < 150, 'Timed out')
time.sleep(10)
print 'Instance (%s) ip (%s) status (%s)' % (self.name, self.ip,
- server.status)
+ server.status)
tries = 0
while True:
time.sleep(10)
else:
print 'Instance (%s) ip (%s) SSH detected.' % (self.name,
- self.ip)
+ self.ip)
break
tries = 0
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')
+ 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)
files = stdout.readlines()
cfn_tools_files = ['cfn-init', 'cfn-hup', 'cfn-signal',
- 'cfn-get-metadata', 'cfn_helper.py']
+ 'cfn-get-metadata', 'cfn_helper.py']
cfntools = {}
for file in cfn_tools_files:
t_data_list = joined_t_data.split('\n')
# must match user data injection
t_data_list.insert(len(t_data_list) - 1,
- u'touch /var/lib/cloud/instance/provision-finished')
+ u'touch /var/lib/cloud/instance/provision-finished')
self.testcase.assertEqual(t_data_list, remote_file_list_u)
self.testcase.assertEqual(data, f.read())
def get_ssh_client(self):
- if self.ssh.get_transport() != None:
+ if self.ssh.get_transport() is not None:
return self.ssh
return None
def get_sftp_client(self):
- if self.sftp != None:
+ if self.sftp is not None:
return self.sftp
return None
class Stack(object):
def __init__(self, testcase, template_file, distribution, arch, jeos_type,
- stack_paramstr, stackname=DEFAULT_STACKNAME):
+ stack_paramstr, stackname=DEFAULT_STACKNAME):
self.testcase = testcase
self.stackname = stackname
'keystone',
'keystone authentication required')
- self.glanceclient = glance_client.Client(host="0.0.0.0", port=9292,
- use_ssl=False, auth_tok=None, creds=self.creds)
+ self.glanceclient = glance_client.Client(host="0.0.0.0",
+ port=9292,
+ use_ssl=False,
+ auth_tok=None,
+ creds=self.creds)
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.creds['password'],
+ self.creds['tenant'],
+ self.creds['auth_url'],
+ service_type='compute')
self.heatclient = self._create_heat_client()
self.testcase.assertTrue(create_list)
self.testcase.assertEqual(len(create_list), 1)
stack_id = create_list[0].findtext('StackId')
- self.testcase.assertTrue(stack_id != None)
+ self.testcase.assertTrue(stack_id is not None)
self.check_stackid(stack_id)
def _check_update_result(self, result):
self.testcase.assertTrue(update_list)
self.testcase.assertEqual(len(update_list), 1)
stack_id = update_list[0].findtext('StackId')
- self.testcase.assertTrue(stack_id != None)
+ 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) != None)
+ 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)
+ 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)
# 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'])
+ 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)
# 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)
+ 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()