self.ChangeTestCasePara(testname, self.masterusername,
self.masterpassword)
runhandle = handle
- runtest = self.home + "/OnosSystemTest/TestON/bin/cli.py run " + \
- testname
+ runtest = (self.home + "/OnosSystemTest/TestON/bin/cli.py run " +
+ testname)
runhandle.sendline(runtest)
circletime = 0
lastshowscreeninfo = ''
print "try to connect " + str(host)
result = self.CheckSshNoPasswd(host)
if not result:
- print "ssh lgin failed,try to copy master publickey" + \
- "to agent " + str(host)
+ print ("ssh lgin failed,try to copy master publickey" +
+ "to agent " + str(host))
self.CopyPublicKey(host)
self.OnosPushKeys(handle, "onos-push-keys " + self.OCT, masterpass)
self.OnosPushKeys(handle, "onos-push-keys " + self.OC1, agentpass)
filepath = '/root/'
else:
filepath = '/home/' + self.masterusername + '/'
- filepath = filepath + "OnosSystemTest/TestON/tests/" \
- + testcase + "/" + testcase + ".topo"
+ filepath = (filepath + "OnosSystemTest/TestON/tests/" +
+ testcase + "/" + testcase + ".topo")
line = open(filepath, 'r').readlines()
lenall = len(line) - 1
for i in range(lenall - 2):
if("localhost" in line[i]) or ("OCT" in line[i]):
line[i + 1] = re.sub(">\w+", ">" + user, line[i + 1])
line[i + 2] = re.sub(">\w+", ">" + password, line[i + 2])
- if "OC1" in line[i] or "OC2" in line[i] or "OC3" in line[i] or \
- "OCN" in line[i] or "OCN2" in line[i]:
- line[i + 1] = re.sub(">\w+", ">root", line[i + 1])
- line[i + 2] = re.sub(">\w+", ">root", line[i + 2])
+ if ("OC1" in line[i] or "OC2" in line[i] or "OC3" in line[i] or
+ "OCN" in line[i] or "OCN2" in line[i]):
+ line[i + 1] = re.sub(">\w+", ">root", line[i + 1])
+ line[i + 2] = re.sub(">\w+", ">root", line[i + 2])
NewFile = open(filepath, 'w')
NewFile.writelines(line)
NewFile.close
get("onoscli_username"))
self.agentpassword = str(functest_yaml.get("ONOS").get("general").
get("onoscli_password"))
- self.runtimeout = functest_yaml.get("ONOS").\
- get("general").get("runtimeout")
+ self.runtimeout = functest_yaml.get("ONOS").get("general").get(
+ "runtimeout")
self.OCT = str(functest_yaml.get("ONOS").get("environment").get("OCT"))
self.OC1 = str(functest_yaml.get("ONOS").get("environment").get("OC1"))
self.OC2 = str(functest_yaml.get("ONOS").get("environment").get("OC2"))
# onos parameters
TEST_DB = functest_yaml.get("results").get("test_db_url")
-ONOS_REPO_PATH = functest_yaml.get("general").\
- get("directories").get("dir_repos")
-ONOS_CONF_DIR = functest_yaml.get("general").\
- get("directories").get("dir_functest_conf")
+ONOS_REPO_PATH = functest_yaml.get("general").get("directories").get(
+ "dir_repos")
+ONOS_CONF_DIR = functest_yaml.get("general").get("directories").get(
+ "dir_functest_conf")
REPO_PATH = ONOS_REPO_PATH + '/functest/'
if not os.path.exists(REPO_PATH):
logger.error("Functest repository directory not found '%s'" % REPO_PATH)
parserout.password,
com)
elif args['which'] == 'oval':
- com = '{0} oval eval --results {1}/{2}' + \
- ' --report {1}/{3} {4}'.format(oscap,
- tmpdir.rstrip(),
- parserout.results,
- parserout.report,
- parserout.secpolicy)
+ com = ('{0} oval eval --results {1}/{2}' +
+ ' --report {1}/{3} {4}'.format(oscap,
+ tmpdir.rstrip(),
+ parserout.results,
+ parserout.report,
+ parserout.secpolicy))
connect = connect.connectionManager(parserout.host,
parserout.user,
parserout.password,
elif port['device_owner'] == 'network:router_interface':
logger.debug("Detaching port %s (subnet %s) from router %s ..."
% (port_id, subnet_id, router_id))
- if openstack_utils.\
- remove_interface_router(neutron_client,
- router_id,
- subnet_id):
+ if openstack_utils.remove_interface_router(
+ neutron_client, router_id, subnet_id):
time.sleep(5) # leave 5 seconds to detach
logger.debug(" > Done!")
else:
f.close()
HOME = os.environ['HOME'] + "/"
-SCENARIOS_DIR = REPO_PATH + functest_yaml.get("general"). \
- get("directories").get("dir_rally_scn")
+SCENARIOS_DIR = REPO_PATH + functest_yaml.get("general").get(
+ "directories").get("dir_rally_scn")
TEMPLATE_DIR = SCENARIOS_DIR + "scenario/templates"
SUPPORT_DIR = SCENARIOS_DIR + "scenario/support"
ITERATIONS_AMOUNT = 10
CONCURRENCY = 4
-RESULTS_DIR = functest_yaml.get("general").get("directories"). \
- get("dir_rally_res")
-TEMPEST_CONF_FILE = functest_yaml.get("general").get("directories"). \
- get("dir_results") + '/tempest/tempest.conf'
+RESULTS_DIR = functest_yaml.get("general").get("directories").get(
+ "dir_rally_res")
+TEMPEST_CONF_FILE = functest_yaml.get("general").get("directories").get(
+ "dir_results") + '/tempest/tempest.conf'
TEST_DB = functest_yaml.get("results").get("test_db_url")
-PRIVATE_NETWORK = functest_yaml.get("general"). \
- get("openstack").get("neutron_private_net_name")
-
-GLANCE_IMAGE_NAME = functest_yaml.get("general"). \
- get("openstack").get("image_name")
-GLANCE_IMAGE_FILENAME = functest_yaml.get("general"). \
- get("openstack").get("image_file_name")
-GLANCE_IMAGE_FORMAT = functest_yaml.get("general"). \
- get("openstack").get("image_disk_format")
-GLANCE_IMAGE_PATH = functest_yaml.get("general"). \
- get("directories").get("dir_functest_data") + "/" + GLANCE_IMAGE_FILENAME
+PRIVATE_NETWORK = functest_yaml.get("general").get("openstack").get(
+ "neutron_private_net_name")
+
+GLANCE_IMAGE_NAME = functest_yaml.get("general").get("openstack").get(
+ "image_name")
+GLANCE_IMAGE_FILENAME = functest_yaml.get("general").get("openstack").get(
+ "image_file_name")
+GLANCE_IMAGE_FORMAT = functest_yaml.get("general").get("openstack").get(
+ "image_disk_format")
+GLANCE_IMAGE_PATH = functest_yaml.get("general").get("directories").get(
+ "dir_functest_data") + "/" + GLANCE_IMAGE_FILENAME
CINDER_VOLUME_TYPE_NAME = "volume_test"
def live_migration_supported():
config = iniparse.ConfigParser()
- if config.read(TEMPEST_CONF_FILE) and \
- config.has_section('compute-feature-enabled') and \
- config.has_option('compute-feature-enabled', 'live_migration'):
+ if (config.read(TEMPEST_CONF_FILE) and
+ config.has_section('compute-feature-enabled') and
+ config.has_option('compute-feature-enabled', 'live_migration')):
return config.getboolean('compute-feature-enabled', 'live_migration')
return False
if args.verbose:
result += line
else:
- if "Load duration" in line or \
- "started" in line or \
- "finished" in line or \
- " Preparing" in line or \
- "+-" in line or \
- "|" in line:
+ if ("Load duration" in line or
+ "started" in line or
+ "finished" in line or
+ " Preparing" in line or
+ "+-" in line or
+ "|" in line):
result += line
elif "test scenario" in line:
result += "\n" + line
result += line + "\n\n"
# parse output for summary report
- if "| " in line and \
- "| action" not in line and \
- "| Starting" not in line and \
- "| Completed" not in line and \
- "| ITER" not in line and \
- "| " not in line and \
- "| total" not in line:
+ if ("| " in line and
+ "| action" not in line and
+ "| Starting" not in line and
+ "| Completed" not in line and
+ "| ITER" not in line and
+ "| " not in line and
+ "| total" not in line):
nb_tests += 1
elif "| total" in line:
percentage = ((line.split('|')[8]).strip(' ')).strip('%')
logger.debug('Scenario fetched from : {}'.format(test_file_name))
- cmd_line = "rally task start --abort-on-sla-failure " + \
- "--task {} ".format(task_file) + \
- "--task-args \"{}\" ".format(build_task_args(test_name))
+ cmd_line = ("rally task start --abort-on-sla-failure " +
+ "--task {} ".format(task_file) +
+ "--task-args \"{}\" ".format(build_task_args(test_name)))
logger.debug('running command line : {}'.format(cmd_line))
p = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
neutron_client = neutronclient.Client(**creds_neutron)
creds_keystone = openstack_utils.get_credentials("keystone")
keystone_client = keystoneclient.Client(**creds_keystone)
- glance_endpoint = keystone_client.\
- service_catalog.url_for(service_type='image',
- endpoint_type='publicURL')
+ glance_endpoint = keystone_client.service_catalog.url_for(
+ service_type='image', endpoint_type='publicURL')
glance_client = glanceclient.Client(1, glance_endpoint,
token=keystone_client.auth_token)
creds_cinder = openstack_utils.get_credentials("cinder")
volume_types = openstack_utils.list_volume_types(cinder_client,
private=False)
if not volume_types:
- volume_type = openstack_utils.\
- create_volume_type(cinder_client, CINDER_VOLUME_TYPE_NAME)
+ volume_type = openstack_utils.create_volume_type(
+ cinder_client, CINDER_VOLUME_TYPE_NAME)
if not volume_type:
logger.error("Failed to create volume type...")
exit(-1)
logger.debug("Test name: " + args.test_name)
run_task(args.test_name)
- report = "\n"\
- " "\
- "\n"\
- " Rally Summary Report\n"\
- "\n"\
- "+===================+============+===============+===========+"\
- "\n"\
- "| Module | Duration | nb. Test Run | Success |"\
- "\n"\
- "+===================+============+===============+===========+"\
- "\n"
+ report = ("\n"
+ " "
+ "\n"
+ " Rally Summary Report\n"
+ "\n"
+ "+===================+============+===============+===========+"
+ "\n"
+ "| Module | Duration | nb. Test Run | Success |"
+ "\n"
+ "+===================+============+===============+===========+"
+ "\n")
payload = []
# for each scenario we draw a row for the table
total_nb_tests += int(s['nb_tests'])
success = "{0:<10}".format(str(s['success']) + '%')
total_success += float(s['success'])
- report += "" + \
- "| " + name + " | " + duration + " | " + \
- nb_tests + " | " + success + "|\n" + \
- "+-------------------+------------+---------------+-----------+\n"
+ report += ("" +
+ "| " + name + " | " + duration + " | " +
+ nb_tests + " | " + success + "|\n" +
+ "+-------------------+------------"
+ "+---------------+-----------+\n")
payload.append({'module': name,
'details': {'duration': s['overall_duration'],
'nb tests': s['nb_tests'],
total_success_str = "{0:<10}".format(str(total_success) + '%')
report += "+===================+============+===============+===========+"
report += "\n"
- report += "| TOTAL: | " + total_duration_str2 + " | " + \
- total_nb_tests_str + " | " + total_success_str + "|\n"
+ report += ("| TOTAL: | " + total_duration_str2 + " | " +
+ total_nb_tests_str + " | " + total_success_str + "|\n")
report += "+===================+============+===============+===========+"
report += "\n"
TEST_DB = functest_yaml.get("results").get("test_db_url")
MODE = "smoke"
-TENANT_NAME = functest_yaml.get("tempest").\
- get("identity").get("tenant_name")
-TENANT_DESCRIPTION = functest_yaml.get("tempest").\
- get("identity").get("tenant_description")
-USER_NAME = functest_yaml.get("tempest").\
- get("identity").get("user_name")
-USER_PASSWORD = functest_yaml.get("tempest").\
- get("identity").get("user_password")
-SSH_USER_REGEX = functest_yaml.get("tempest").\
- get("input-scenario").get("ssh_user_regex")
-DEPLOYMENT_MAME = functest_yaml.get("rally").\
- get("deployment_name")
-RALLY_INSTALLATION_DIR = functest_yaml.get("general").\
- get("directories").get("dir_rally_inst")
-RESULTS_DIR = functest_yaml.get("general").\
- get("directories").get("dir_results")
+TENANT_NAME = functest_yaml.get("tempest").get("identity").get("tenant_name")
+TENANT_DESCRIPTION = functest_yaml.get("tempest").get("identity").get(
+ "tenant_description")
+USER_NAME = functest_yaml.get("tempest").get("identity").get("user_name")
+USER_PASSWORD = functest_yaml.get("tempest").get("identity").get(
+ "user_password")
+SSH_USER_REGEX = functest_yaml.get("tempest").get("input-scenario").get(
+ "ssh_user_regex")
+DEPLOYMENT_MAME = functest_yaml.get("rally").get("deployment_name")
+RALLY_INSTALLATION_DIR = functest_yaml.get("general").get("directories").get(
+ "dir_rally_inst")
+RESULTS_DIR = functest_yaml.get("general").get("directories").get(
+ "dir_results")
TEMPEST_RESULTS_DIR = RESULTS_DIR + '/tempest'
-TEST_LIST_DIR = functest_yaml.get("general").\
- get("directories").get("dir_tempest_cases")
+TEST_LIST_DIR = functest_yaml.get("general").get("directories").get(
+ "dir_tempest_cases")
TEMPEST_LIST_FILE = REPO_PATH + TEST_LIST_DIR + 'test_list.txt'
TEMPEST_DEFCORE = REPO_PATH + TEST_LIST_DIR + 'defcore_req.txt'
if deployment_uuid == "":
logger.debug(" Rally deployment NOT found")
return False
- deployment_dir = RALLY_INSTALLATION_DIR + "/tempest/for-deployment-" + \
- deployment_uuid
+ deployment_dir = (RALLY_INSTALLATION_DIR + "/tempest/for-deployment-" +
+ deployment_uuid)
logger.debug("Finding tempest.conf file...")
tempest_conf_file = deployment_dir + "/tempest.conf"
if CI_DEBUG == "true" or CI_DEBUG == "True":
subprocess.call(cmd_line, shell=True, stderr=subprocess.STDOUT)
else:
- header = "Tempest environment:\n"\
- " Installer: %s\n Scenario: %s\n Node: %s\n Date: %s\n" % \
- (os.getenv('INSTALLER_TYPE', 'Unknown'),
- os.getenv('DEPLOY_SCENARIO', 'Unknown'),
- os.getenv('NODE_NAME', 'Unknown'),
- time.strftime("%a %b %d %H:%M:%S %Z %Y"))
+ header = ("Tempest environment:\n"
+ " Installer: %s\n Scenario: %s\n Node: %s\n Date: %s\n" %
+ (os.getenv('INSTALLER_TYPE', 'Unknown'),
+ os.getenv('DEPLOY_SCENARIO', 'Unknown'),
+ os.getenv('NODE_NAME', 'Unknown'),
+ time.strftime("%a %b %d %H:%M:%S %Z %Y")))
f_stdout = open(TEMPEST_RESULTS_DIR + "/tempest.log", 'w+')
f_stderr = open(TEMPEST_RESULTS_DIR + "/tempest-error.log", 'w+')
actions = ['start', 'check', 'clean']
parser = argparse.ArgumentParser()
-parser.add_argument("action", help="Possible actions are: \
- '{d[0]}|{d[1]}|{d[2]}' ".format(d=actions))
+parser.add_argument("action", help="Possible actions are: "
+ "'{d[0]}|{d[1]}|{d[2]}' ".format(d=actions))
parser.add_argument("-d", "--debug", help="Debug mode", action="store_true")
parser.add_argument("-f", "--force", help="Force", action="store_true")
args = parser.parse_args()
""" global variables """
# Directories
-RALLY_DIR = FUNCTEST_REPO + functest_yaml.get("general").\
- get("directories").get("dir_rally")
-RALLY_REPO_DIR = functest_yaml.get("general").\
- get("directories").get("dir_repo_rally")
-RALLY_INSTALLATION_DIR = functest_yaml.get("general").\
- get("directories").get("dir_rally_inst")
-RALLY_RESULT_DIR = functest_yaml.get("general").\
- get("directories").get("dir_rally_res")
-TEMPEST_REPO_DIR = functest_yaml.get("general").\
- get("directories").get("dir_repo_tempest")
-VPING_DIR = FUNCTEST_REPO + functest_yaml.\
- get("general").get("directories").get("dir_vping")
-ODL_DIR = FUNCTEST_REPO + functest_yaml.\
- get("general").get("directories").get("dir_odl")
-DATA_DIR = functest_yaml.get("general").\
- get("directories").get("dir_functest_data")
+RALLY_DIR = FUNCTEST_REPO + functest_yaml.get("general").get(
+ "directories").get("dir_rally")
+RALLY_REPO_DIR = functest_yaml.get("general").get(
+ "directories").get("dir_repo_rally")
+RALLY_INSTALLATION_DIR = functest_yaml.get("general").get(
+ "directories").get("dir_rally_inst")
+RALLY_RESULT_DIR = functest_yaml.get("general").get(
+ "directories").get("dir_rally_res")
+TEMPEST_REPO_DIR = functest_yaml.get("general").get(
+ "directories").get("dir_repo_tempest")
+VPING_DIR = FUNCTEST_REPO + functest_yaml.get("general").get(
+ "directories").get("dir_vping")
+ODL_DIR = FUNCTEST_REPO + functest_yaml.get("general").get(
+ "directories").get("dir_odl")
+DATA_DIR = functest_yaml.get("general").get(
+ "directories").get("dir_functest_data")
# Tempest/Rally configuration details
DEPLOYMENT_MAME = functest_yaml.get("rally").get("deployment_name")
# Image (cirros)
-IMAGE_FILE_NAME = functest_yaml.get("general").\
- get("openstack").get("image_file_name")
+IMAGE_FILE_NAME = functest_yaml.get("general").get("openstack").get(
+ "image_file_name")
IMAGE_PATH = DATA_DIR + "/" + IMAGE_FILE_NAME
# NEUTRON Private Network parameters
-NEUTRON_PRIVATE_NET_NAME = functest_yaml.get("general"). \
- get("openstack").get("neutron_private_net_name")
-NEUTRON_PRIVATE_SUBNET_NAME = functest_yaml.get("general"). \
- get("openstack").get("neutron_private_subnet_name")
-NEUTRON_PRIVATE_SUBNET_CIDR = functest_yaml.get("general"). \
- get("openstack").get("neutron_private_subnet_cidr")
-NEUTRON_ROUTER_NAME = functest_yaml.get("general"). \
- get("openstack").get("neutron_router_name")
+NEUTRON_PRIVATE_NET_NAME = functest_yaml.get("general").get(
+ "openstack").get("neutron_private_net_name")
+NEUTRON_PRIVATE_SUBNET_NAME = functest_yaml.get("general").get(
+ "openstack").get("neutron_private_subnet_name")
+NEUTRON_PRIVATE_SUBNET_CIDR = functest_yaml.get("general").get(
+ "openstack").get("neutron_private_subnet_cidr")
+NEUTRON_ROUTER_NAME = functest_yaml.get("general").get(
+ "openstack").get("neutron_router_name")
creds_neutron = openstack_utils.get_credentials("neutron")
neutron_client = neutronclient.Client(**creds_neutron)
functest_utils.execute_command(cmd, logger)
logger.debug("Installing tempest from existing repo...")
- cmd = "rally verify install --source " + TEMPEST_REPO_DIR + \
- " --system-wide"
+ cmd = ("rally verify install --source " + TEMPEST_REPO_DIR +
+ " --system-wide")
functest_utils.execute_command(cmd, logger)
cmd = "rally deployment check"
def create_private_neutron_net(neutron):
neutron.format = 'json'
logger.info("Creating network '%s'..." % NEUTRON_PRIVATE_NET_NAME)
- network_id = openstack_utils. \
- create_neutron_net(neutron, NEUTRON_PRIVATE_NET_NAME)
+ network_id = openstack_utils.create_neutron_net(
+ neutron, NEUTRON_PRIVATE_NET_NAME)
if not network_id:
return False
logger.info("Updating neutron network '%s' failed" % network_id)
logger.info("Creating Subnet....")
- subnet_id = openstack_utils. \
- create_neutron_subnet(neutron,
- NEUTRON_PRIVATE_SUBNET_NAME,
- NEUTRON_PRIVATE_SUBNET_CIDR,
- network_id)
+ subnet_id = openstack_utils.create_neutron_subnet(
+ neutron, NEUTRON_PRIVATE_SUBNET_NAME, NEUTRON_PRIVATE_SUBNET_CIDR,
+ network_id)
if not subnet_id:
return False
logger.debug("Subnet '%s' created successfully." % subnet_id)
logger.info("Creating Router...")
- router_id = openstack_utils. \
- create_neutron_router(neutron, NEUTRON_ROUTER_NAME)
+ router_id = openstack_utils.create_neutron_router(neutron,
+ NEUTRON_ROUTER_NAME)
if not router_id:
return False
FLAVOR_DISK = functest_yaml.get('promise').get('general').get('flavor_disk')
-GLANCE_IMAGE_FILENAME = functest_yaml.get('general'). \
- get('openstack').get('image_file_name')
-GLANCE_IMAGE_FORMAT = functest_yaml.get('general'). \
- get('openstack').get('image_disk_format')
-GLANCE_IMAGE_PATH = functest_yaml.get('general'). \
- get('directories').get('dir_functest_data') + "/" + GLANCE_IMAGE_FILENAME
+GLANCE_IMAGE_FILENAME = functest_yaml.get('general').get('openstack').get(
+ 'image_file_name')
+GLANCE_IMAGE_FORMAT = functest_yaml.get('general').get('openstack').get(
+ 'image_disk_format')
+GLANCE_IMAGE_PATH = functest_yaml.get('general').get('directories').get(
+ 'dir_functest_data') + "/" + GLANCE_IMAGE_FILENAME
""" logging configuration """
logger = logging.getLogger('Promise')
"project_id": TENANT_NAME,
})
- glance_endpoint = keystone.\
- service_catalog.url_for(service_type='image',
- endpoint_type='publicURL')
+ glance_endpoint = keystone.service_catalog.url_for(
+ service_type='image', endpoint_type='publicURL')
glance = glclient.Client(1, glance_endpoint, token=keystone.auth_token)
nova = nvclient.Client("2", **nv_creds)
logger.debug(r)
return True
except Exception, e:
- print "Error [push_results_to_db('%s', '%s', '%s', " + \
- "'%s', '%s', '%s', '%s', '%s', '%s')]:" \
- % (db_url, project, case_name, pod_name, version,
- scenario, criteria, build_tag, payload), e
+ print ("Error [push_results_to_db('%s', '%s', '%s', " +
+ "'%s', '%s', '%s', '%s', '%s', '%s')]:" %
+ (db_url, project, case_name, pod_name, version,
+ scenario, criteria, build_tag, payload), e)
return False
"ca_file": cacert})
creds.update({"insecure": "True", "https_insecure": "True"})
if not os.path.isfile(cacert):
- print "WARNING: The 'OS_CACERT' environment variable is " + \
- "set to %s but the file does not exist." % cacert
+ print ("WARNING: The 'OS_CACERT' environment variable is " +
+ "set to %s but the file does not exist." % cacert)
return creds
instance = nova_client.servers.get(instance.id)
return instance.status
except:
- # print "Error [get_instance_status(nova_client, '%s')]:" % \
- # str(instance), e
+ # print ("Error [get_instance_status(nova_client, '%s')]:" %
+ # str(instance), e)
return None
instance = nova_client.servers.find(name=instance_name)
return instance
except Exception, e:
- print "Error [get_instance_by_name(nova_client, '%s')]:" % \
- instance_name, e
+ print ("Error [get_instance_by_name(nova_client, '%s')]:" %
+ instance_name, e)
return None
try:
flavor = nova_client.flavors.create(flavor_name, ram, vcpus, disk)
except Exception, e:
- print "Error [create_flavor(nova_client, '%s', '%s', '%s', "\
- "'%s')]:" % (flavor_name, ram, disk, vcpus), e
+ print ("Error [create_flavor(nova_client, '%s', '%s', '%s', "
+ "'%s')]:" % (flavor_name, ram, disk, vcpus), e)
return None
return flavor.id
nova_client.servers.add_floating_ip(server_id, floatingip_id)
return True
except Exception, e:
- print "Error [add_floating_ip(nova_client, '%s', '%s')]:" % \
- (server_id, floatingip_id), e
+ print ("Error [add_floating_ip(nova_client, '%s', '%s')]:" %
+ (server_id, floatingip_id), e)
return False
nova_client.floating_ips.delete(floatingip_id)
return True
except Exception, e:
- print "Error [delete_floating_ip(nova_client, '%s')]:" \
- % floatingip_id, e
+ print ("Error [delete_floating_ip(nova_client, '%s')]:" %
+ floatingip_id, e)
return False
subnet = neutron_client.create_subnet(body=json_body)
return subnet['subnets'][0]['id']
except Exception, e:
- print "Error [create_neutron_subnet(neutron_client, '%s', '%s', "\
- "'%s')]:" % (name, cidr, net_id), e
+ print ("Error [create_neutron_subnet(neutron_client, '%s', '%s', "
+ "'%s')]:" % (name, cidr, net_id), e)
return False
port = neutron_client.create_port(body=json_body)
return port['port']['id']
except Exception, e:
- print "Error [create_neutron_port(neutron_client, '%s', '%s', "\
- "'%s')]:" % (name, network_id, ip), e
+ print ("Error [create_neutron_port(neutron_client, '%s', '%s', "
+ "'%s')]:" % (name, network_id, ip), e)
return False
neutron_client.update_network(network_id, body=json_body)
return True
except Exception, e:
- print "Error [update_neutron_net(neutron_client, '%s', '%s')]:" % \
- (network_id, str(shared)), e
+ print ("Error [update_neutron_net(neutron_client, '%s', '%s')]:" %
+ (network_id, str(shared)), e)
return False
body=json_body)
return port['port']['id']
except Exception, e:
- print "Error [update_neutron_port(neutron_client, '%s', '%s')]:" % \
- (port_id, device_owner), e
+ print ("Error [update_neutron_port(neutron_client, '%s', '%s')]:" %
+ (port_id, device_owner), e)
return False
neutron_client.add_interface_router(router=router_id, body=json_body)
return True
except Exception, e:
- print "Error [add_interface_router(neutron_client, '%s', '%s')]:" % \
- (router_id, subnet_id), e
+ print ("Error [add_interface_router(neutron_client, '%s', '%s')]:" %
+ (router_id, subnet_id), e)
return False
neutron_client.add_gateway_router(router_id, router_dict)
return True
except Exception, e:
- print "Error [add_gateway_router(neutron_client, '%s')]:" \
- % router_id, e
+ print ("Error [add_gateway_router(neutron_client, '%s')]:" %
+ router_id, e)
return False
neutron_client.delete_network(network_id)
return True
except Exception, e:
- print "Error [delete_neutron_net(neutron_client, '%s')]:" \
- % network_id, e
+ print ("Error [delete_neutron_net(neutron_client, '%s')]:" %
+ network_id, e)
return False
neutron_client.delete_subnet(subnet_id)
return True
except Exception, e:
- print "Error [delete_neutron_subnet(neutron_client, '%s')]:" \
- % subnet_id, e
+ print ("Error [delete_neutron_subnet(neutron_client, '%s')]:" %
+ subnet_id, e)
return False
neutron_client.delete_router(router=router_id)
return True
except Exception, e:
- print "Error [delete_neutron_router(neutron_client, '%s')]:" % \
- router_id, e
+ print ("Error [delete_neutron_router(neutron_client, '%s')]:" %
+ router_id, e)
return False
body=json_body)
return True
except Exception, e:
- print "Error [remove_interface_router(neutron_client, '%s', '%s')]:" \
- % (router_id, subnet_id), e
+ print ("Error [remove_interface_router(neutron_client, '%s', '%s')]:" %
+ (router_id, subnet_id), e)
return False
neutron_client.remove_gateway_router(router_id)
return True
except Exception, e:
- print "Error [remove_gateway_router(neutron_client, '%s')]:" \
- % router_id, e
+ print ("Error [remove_gateway_router(neutron_client, '%s')]:" %
+ router_id, e)
return False
secgroup = neutron_client.create_security_group(json_body)
return secgroup['security_group']
except Exception, e:
- print "Error [create_security_group(neutron_client, '%s', '%s')]:" % \
- (sg_name, sg_description), e
+ print ("Error [create_security_group(neutron_client, '%s', '%s')]:" %
+ (sg_name, sg_description), e)
return False
'port_range_max': port_range_max,
'protocol': protocol}}
else:
- print "Error [create_secgroup_rule(neutron_client, '%s', '%s', "\
- "'%s', '%s', '%s', '%s')]:" % (neutron_client, sg_id, direction,
- port_range_min, port_range_max,
- protocol),\
- " Invalid values for port_range_min, port_range_max"
+ print ("Error [create_secgroup_rule(neutron_client, '%s', '%s', "
+ "'%s', '%s', '%s', '%s')]:" % (neutron_client, sg_id, direction,
+ port_range_min, port_range_max,
+ protocol),
+ " Invalid values for port_range_min, port_range_max")
return False
try:
neutron_client.create_security_group_rule(json_body)
return True
except Exception, e:
- print "Error [create_secgroup_rule(neutron_client, '%s', '%s', "\
- "'%s', '%s', '%s', '%s')]:" % (neutron_client, sg_id, direction,
- port_range_min, port_range_max,
- protocol), e
+ print ("Error [create_secgroup_rule(neutron_client, '%s', '%s', "
+ "'%s', '%s', '%s', '%s')]:" % (neutron_client, sg_id, direction,
+ port_range_min, port_range_max,
+ protocol), e)
return False
nova_client.servers.add_security_group(instance_id, secgroup_id)
return True
except Exception, e:
- print "Error [add_secgroup_to_instance(nova_client, '%s', '%s')]: " % \
- (instance_id, secgroup_id), e
+ print ("Error [add_secgroup_to_instance(nova_client, '%s', '%s')]: " %
+ (instance_id, secgroup_id), e)
return False
body=json_body)
return True
except Exception, e:
- print "Error [update_sg_quota(neutron_client, '%s', '%s', " \
- "'%s')]:" % (tenant_id, sg_quota, sg_rule_quota), e
+ print ("Error [update_sg_quota(neutron_client, '%s', '%s', "
+ "'%s')]:" % (tenant_id, sg_quota, sg_rule_quota), e)
return False
neutron_client.delete_security_group(secgroup_id)
return True
except Exception, e:
- print "Error [delete_security_group(neutron_client, '%s')]:" \
- % secgroup_id, e
+ print ("Error [delete_security_group(neutron_client, '%s')]:" %
+ secgroup_id, e)
return False
data=fimage)
return image.id
except Exception, e:
- print "Error [create_glance_image(glance_client, '%s', '%s', "\
- "'%s')]:" % (image_name, file_path, str(public)), e
+ print ("Error [create_glance_image(glance_client, '%s', '%s', "
+ "'%s')]:" % (image_name, file_path, str(public)), e)
return False
nova_client.images.delete(image_id)
return True
except Exception, e:
- print "Error [delete_glance_image(nova_client, '%s')]:" % image_id, e
+ print ("Error [delete_glance_image(nova_client, '%s')]:" % image_id, e)
return False
cinder_client.quotas.update(tenant_id, **quotas_values)
return True
except Exception, e:
- print "Error [update_cinder_quota(cinder_client, '%s', '%s', '%s'" \
- "'%s')]:" % (tenant_id, vols_quota,
- snapshots_quota, gigabytes_quota), e
+ print ("Error [update_cinder_quota(cinder_client, '%s', '%s', '%s'"
+ "'%s')]:" % (tenant_id, vols_quota,
+ snapshots_quota, gigabytes_quota), e)
return False
cinder_client.volumes.delete(volume_id)
return True
except Exception, e:
- print "Error [delete_volume(cinder_client, '%s', '%s')]:" % \
- (volume_id, str(forced)), e
+ print ("Error [delete_volume(cinder_client, '%s', '%s')]:" %
+ (volume_id, str(forced)), e)
return False
cinder_client.volume_types.delete(volume_type)
return True
except Exception, e:
- print "Error [delete_volume_type(cinder_client, '%s')]:" \
- % volume_type, e
+ print ("Error [delete_volume_type(cinder_client, '%s')]:" %
+ volume_type, e)
return False
enabled=True)
return tenant.id
except Exception, e:
- print "Error [create_tenant(cinder_client, '%s', '%s')]:" % \
- (tenant_name, tenant_description), e
+ print ("Error [create_tenant(cinder_client, '%s', '%s')]:" %
+ (tenant_name, tenant_description), e)
return False
enabled=True)
return user.id
except Exception, e:
- print "Error [create_user(keystone_client, '%s', '%s', '%s'" \
- "'%s')]:" % (user_name, user_password, user_email, tenant_id), e
+ print ("Error [create_user(keystone_client, '%s', '%s', '%s'"
+ "'%s')]:" % (user_name, user_password, user_email, tenant_id),
+ e)
return False
keystone_client.roles.add_user_role(user_id, role_id, tenant_id)
return True
except Exception, e:
- print "Error [add_role_user(keystone_client, '%s', '%s'" \
- "'%s')]:" % (user_id, role_id, tenant_id), e
+ print ("Error [add_role_user(keystone_client, '%s', '%s'"
+ "'%s')]:" % (user_id, role_id, tenant_id), e)
return False
test = generateTestcaseList(functest_yaml)
- expected_list = "vping_ssh vping_userdata tempest odl doctor " + \
- "promise policy-test odl-vpn_service-tests vims rally "
+ expected_list = ("vping_ssh vping_userdata tempest odl doctor " +
+ "promise policy-test odl-vpn_service-tests vims " +
+ "rally ")
self.assertEqual(test, expected_list)
def tearDown(self):
if self.logger:
self.logger.info("Launching the cloudify-manager deployment")
script = "set -e; "
- script += "source " + self.testcase_dir + \
- "venv_cloudify/bin/activate; "
+ script += ("source " + self.testcase_dir +
+ "venv_cloudify/bin/activate; ")
script += "cd " + self.testcase_dir + "; "
script += "cfy init -r; "
script += "cd cloudify-manager-blueprint; "
- script += "cfy local create-requirements -o requirements.txt " + \
- "-p openstack-manager-blueprint.yaml; "
+ script += ("cfy local create-requirements -o requirements.txt " +
+ "-p openstack-manager-blueprint.yaml; ")
script += "pip install -r requirements.txt; "
- script += "timeout 1800 cfy bootstrap --install-plugins " + \
- "-p openstack-manager-blueprint.yaml -i inputs.yaml; "
+ script += ("timeout 1800 cfy bootstrap --install-plugins " +
+ "-p openstack-manager-blueprint.yaml -i inputs.yaml; ")
cmd = "/bin/bash -c '" + script + "'"
error = execute_command(cmd, self.logger)
if error:
if self.logger:
self.logger.info("Launching the {0} deployment".format(bp_name))
script = "source " + self.testcase_dir + "venv_cloudify/bin/activate; "
- script += "cd " + self.testcase_dir + \
- blueprint['destination_folder'] + "; "
- script += "cfy blueprints upload -b " + \
- bp_name + " -p openstack-blueprint.yaml; "
- script += "cfy deployments create -b " + bp_name + \
- " -d " + dep_name + " --inputs inputs.yaml; "
- script += "cfy executions start -w install -d " \
- + dep_name + " --timeout 1800; "
+ script += ("cd " + self.testcase_dir +
+ blueprint['destination_folder'] + "; ")
+ script += ("cfy blueprints upload -b " +
+ bp_name + " -p openstack-blueprint.yaml; ")
+ script += ("cfy deployments create -b " + bp_name +
+ " -d " + dep_name + " --inputs inputs.yaml; ")
+ script += ("cfy executions start -w install -d " +
+ dep_name + " --timeout 1800; ")
cmd = "/bin/bash -c '" + script + "'"
error = execute_command(cmd, self.logger)
self.logger.info("Launching the {0} undeployment".format(dep_name))
script = "source " + self.testcase_dir + "venv_cloudify/bin/activate; "
script += "cd " + self.testcase_dir + "; "
- script += "cfy executions start -w uninstall -d " + dep_name \
- + " --timeout 1800 ; "
+ script += ("cfy executions start -w uninstall -d " + dep_name +
+ " --timeout 1800 ; ")
script += "cfy deployments delete -d " + dep_name + "; "
cmd = "/bin/bash -c '" + script + "'"
f.close()
# Cloudify parameters
-VIMS_DIR = REPO_PATH + \
- functest_yaml.get("general").get("directories").get("dir_vIMS")
+VIMS_DIR = (REPO_PATH +
+ functest_yaml.get("general").get("directories").get("dir_vIMS"))
VIMS_DATA_DIR = functest_yaml.get("general").get(
"directories").get("dir_vIMS_data") + "/"
VIMS_TEST_DIR = functest_yaml.get("general").get(
resolvconf += "\nnameserver " + ns
if dns_ip != "":
- script = 'echo -e "nameserver ' + dns_ip + resolvconf + \
- '" > /etc/resolv.conf; '
+ script = ('echo -e "nameserver ' + dns_ip + resolvconf +
+ '" > /etc/resolv.conf; ')
script += 'source /etc/profile.d/rvm.sh; '
script += 'cd ' + VIMS_TEST_DIR + '; '
- script += 'rake test[' + \
- CW_INPUTS["public_domain"] + '] SIGNUP_CODE="secret"'
+ script += ('rake test[' + CW_INPUTS["public_domain"] +
+ '] SIGNUP_CODE="secret"')
cmd = "/bin/bash -c '" + script + "'"
output_file = "output.txt"
})
logger.info("Upload some OS images if it doesn't exist")
- glance_endpoint = keystone.\
- service_catalog.url_for(service_type='image',
- endpoint_type='publicURL')
+ glance_endpoint = keystone.service_catalog.url_for(
+ service_type='image', endpoint_type='publicURL')
glance = glclient.Client(1, glance_endpoint, token=keystone.auth_token)
for img in IMAGES.keys():
logger.error(
"Failed to find %s flavor. "
"Try with ram range default requirement !" % flavor_name)
- flavor_id = openstack_utils.\
- get_flavor_id_by_ram_range(nova, 4000, 8196)
+ flavor_id = openstack_utils.get_flavor_id_by_ram_range(
+ nova, 4000, 8196)
if flavor_id == '':
step_failure("orchestrator",
logger.error(
"Failed to find %s flavor. Try with ram range "
"default requirement !" % flavor_name)
- flavor_id = openstack_utils.\
- get_flavor_id_by_ram_range(nova, 4000, 8196)
+ flavor_id = openstack_utils.get_flavor_id_by_ram_range(
+ nova, 4000, 8196)
if flavor_id == '':
step_failure(
NAME_VM_1 = functest_yaml.get("vping").get("vm_name_1")
NAME_VM_2 = functest_yaml.get("vping").get("vm_name_2")
GLANCE_IMAGE_NAME = functest_yaml.get("vping").get("image_name")
-GLANCE_IMAGE_FILENAME = functest_yaml.get("general"). \
- get("openstack").get("image_file_name")
-GLANCE_IMAGE_FORMAT = functest_yaml.get("general"). \
- get("openstack").get("image_disk_format")
-GLANCE_IMAGE_PATH = functest_yaml.get("general"). \
- get("directories").get("dir_functest_data") + "/" + GLANCE_IMAGE_FILENAME
-
+GLANCE_IMAGE_FILENAME = functest_yaml.get("general").get("openstack").get(
+ "image_file_name")
+GLANCE_IMAGE_FORMAT = functest_yaml.get("general").get("openstack").get(
+ "image_disk_format")
+GLANCE_IMAGE_PATH = functest_yaml.get("general").get("directories").get(
+ "dir_functest_data") + "/" + GLANCE_IMAGE_FILENAME
FLAVOR = functest_yaml.get("vping").get("vm_flavor")
# NEUTRON Private Network parameters
-NEUTRON_PRIVATE_NET_NAME = functest_yaml.get("vping"). \
- get("vping_private_net_name")
-NEUTRON_PRIVATE_SUBNET_NAME = functest_yaml.get("vping"). \
- get("vping_private_subnet_name")
-NEUTRON_PRIVATE_SUBNET_CIDR = functest_yaml.get("vping"). \
- get("vping_private_subnet_cidr")
-NEUTRON_ROUTER_NAME = functest_yaml.get("vping"). \
- get("vping_router_name")
+NEUTRON_PRIVATE_NET_NAME = functest_yaml.get("vping").get(
+ "vping_private_net_name")
+NEUTRON_PRIVATE_SUBNET_NAME = functest_yaml.get("vping").get(
+ "vping_private_subnet_name")
+NEUTRON_PRIVATE_SUBNET_CIDR = functest_yaml.get("vping").get(
+ "vping_private_subnet_cidr")
+NEUTRON_ROUTER_NAME = functest_yaml.get("vping").get(
+ "vping_router_name")
-SECGROUP_NAME = functest_yaml.get("vping"). \
- get("vping_sg_name")
-SECGROUP_DESCR = functest_yaml.get("vping"). \
- get("vping_sg_descr")
+SECGROUP_NAME = functest_yaml.get("vping").get("vping_sg_name")
+SECGROUP_DESCR = functest_yaml.get("vping").get("vping_sg_descr")
def pMsg(value):
neutron.format = 'json'
logger.info('Creating neutron network %s...'
% NEUTRON_PRIVATE_NET_NAME)
- network_id = openstack_utils. \
- create_neutron_net(neutron, NEUTRON_PRIVATE_NET_NAME)
+ network_id = openstack_utils.create_neutron_net(
+ neutron, NEUTRON_PRIVATE_NET_NAME)
if not network_id:
return False
logger.debug("Network '%s' created successfully" % network_id)
logger.debug('Creating Subnet....')
- subnet_id = openstack_utils. \
- create_neutron_subnet(neutron,
- NEUTRON_PRIVATE_SUBNET_NAME,
- NEUTRON_PRIVATE_SUBNET_CIDR,
- network_id)
+ subnet_id = openstack_utils.create_neutron_subnet(
+ neutron, NEUTRON_PRIVATE_SUBNET_NAME, NEUTRON_PRIVATE_SUBNET_CIDR,
+ network_id)
if not subnet_id:
return False
logger.debug("Subnet '%s' created successfully" % subnet_id)
logger.debug('Creating Router...')
- router_id = openstack_utils. \
- create_neutron_router(neutron, NEUTRON_ROUTER_NAME)
+ router_id = openstack_utils.create_neutron_router(
+ neutron, NEUTRON_ROUTER_NAME)
if not router_id:
return False
logger.debug("Adding SSH rules in security group '%s'..."
% SECGROUP_NAME)
- if not openstack_utils.\
- create_secgroup_rule(neutron_client, sg_id,
- 'ingress', 'tcp', '22', '22'):
+ if not openstack_utils.create_secgroup_rule(
+ neutron_client, sg_id, 'ingress', 'tcp', '22', '22'):
logger.error("Failed to create the security group rule...")
return False
- if not openstack_utils.\
- create_secgroup_rule(neutron_client, sg_id,
- 'egress', 'tcp', '22', '22'):
+ if not openstack_utils.create_secgroup_rule(
+ neutron_client, sg_id, 'egress', 'tcp', '22', '22'):
logger.error("Failed to create the security group rule...")
return False
return sg_id
neutron_client = neutronclient.Client(**creds_neutron)
creds_keystone = openstack_utils.get_credentials("keystone")
keystone_client = keystoneclient.Client(**creds_keystone)
- glance_endpoint = keystone_client.\
- service_catalog.url_for(service_type='image',
- endpoint_type='publicURL')
+ glance_endpoint = keystone_client.service_catalog.url_for(
+ service_type='image', endpoint_type='publicURL')
glance_client = glanceclient.Client(1, glance_endpoint,
token=keystone_client.auth_token)
EXIT_CODE = -1
console_log = vm2.get_console_output()
# print each "Sending discover" captured on the console log
- if len(re.findall("Sending discover", console_log)) > discover_count \
- and not got_ip:
+ if (len(re.findall("Sending discover", console_log)) >
+ discover_count and not got_ip):
discover_count += 1
logger.debug("Console-log '%s': Sending discover..."
% NAME_VM_2)
NAME_VM_1 = functest_yaml.get("vping").get("vm_name_1")
NAME_VM_2 = functest_yaml.get("vping").get("vm_name_2")
GLANCE_IMAGE_NAME = functest_yaml.get("vping").get("image_name")
-GLANCE_IMAGE_FILENAME = functest_yaml.get("general"). \
- get("openstack").get("image_file_name")
-GLANCE_IMAGE_FORMAT = functest_yaml.get("general"). \
- get("openstack").get("image_disk_format")
-GLANCE_IMAGE_PATH = functest_yaml.get("general"). \
- get("directories").get("dir_functest_data") + "/" + GLANCE_IMAGE_FILENAME
+GLANCE_IMAGE_FILENAME = functest_yaml.get("general").get(
+ "openstack").get("image_file_name")
+GLANCE_IMAGE_FORMAT = functest_yaml.get("general").get(
+ "openstack").get("image_disk_format")
+GLANCE_IMAGE_PATH = functest_yaml.get("general").get("directories").get(
+ "dir_functest_data") + "/" + GLANCE_IMAGE_FILENAME
FLAVOR = functest_yaml.get("vping").get("vm_flavor")
# NEUTRON Private Network parameters
-NEUTRON_PRIVATE_NET_NAME = functest_yaml.get("vping"). \
- get("vping_private_net_name")
-NEUTRON_PRIVATE_SUBNET_NAME = functest_yaml.get("vping"). \
- get("vping_private_subnet_name")
-NEUTRON_PRIVATE_SUBNET_CIDR = functest_yaml.get("vping"). \
- get("vping_private_subnet_cidr")
-NEUTRON_ROUTER_NAME = functest_yaml.get("vping"). \
- get("vping_router_name")
+NEUTRON_PRIVATE_NET_NAME = functest_yaml.get("vping").get(
+ "vping_private_net_name")
+NEUTRON_PRIVATE_SUBNET_NAME = functest_yaml.get("vping").get(
+ "vping_private_subnet_name")
+NEUTRON_PRIVATE_SUBNET_CIDR = functest_yaml.get("vping").get(
+ "vping_private_subnet_cidr")
+NEUTRON_ROUTER_NAME = functest_yaml.get("vping").get("vping_router_name")
-SECGROUP_NAME = functest_yaml.get("vping"). \
- get("vping_sg_name")
-SECGROUP_DESCR = functest_yaml.get("vping"). \
- get("vping_sg_descr")
+SECGROUP_NAME = functest_yaml.get("vping").get("vping_sg_name")
+SECGROUP_DESCR = functest_yaml.get("vping").get("vping_sg_descr")
def pMsg(value):
else:
neutron.format = 'json'
logger.info('Creating neutron network %s..' % NEUTRON_PRIVATE_NET_NAME)
- network_id = openstack_utils. \
- create_neutron_net(neutron, NEUTRON_PRIVATE_NET_NAME)
+ network_id = openstack_utils.create_neutron_net(
+ neutron, NEUTRON_PRIVATE_NET_NAME)
if not network_id:
return False
logger.debug("Network '%s' created successfully" % network_id)
logger.debug('Creating Subnet....')
- subnet_id = openstack_utils. \
- create_neutron_subnet(neutron,
- NEUTRON_PRIVATE_SUBNET_NAME,
- NEUTRON_PRIVATE_SUBNET_CIDR,
- network_id)
+ subnet_id = openstack_utils.create_neutron_subnet(
+ neutron, NEUTRON_PRIVATE_SUBNET_NAME, NEUTRON_PRIVATE_SUBNET_CIDR,
+ network_id)
if not subnet_id:
return False
logger.debug("Subnet '%s' created successfully" % subnet_id)
logger.debug('Creating Router...')
- router_id = openstack_utils. \
- create_neutron_router(neutron, NEUTRON_ROUTER_NAME)
+ router_id = openstack_utils.create_neutron_router(
+ neutron, NEUTRON_ROUTER_NAME)
if not router_id:
return False
neutron_client = neutronclient.Client(**creds_neutron)
creds_keystone = openstack_utils.get_credentials("keystone")
keystone_client = keystoneclient.Client(**creds_keystone)
- glance_endpoint = keystone_client.\
- service_catalog.url_for(service_type='image',
- endpoint_type='publicURL')
+ glance_endpoint = keystone_client.service_catalog.url_for(
+ service_type='image', endpoint_type='publicURL')
glance_client = glanceclient.Client(1, glance_endpoint,
token=keystone_client.auth_token)
EXIT_CODE = -1
# the long chain corresponds to the ping procedure converted with base 64
# tune (e.g. flavor, images, network) to your specific openstack
# configuration here
- u = "#!/bin/sh\n\nwhile true; do\n ping -c 1 %s 2>&1 >/dev/null\n " \
- "RES=$?\n if [ \"Z$RES\" = \"Z0\" ] ; then\n echo 'vPing OK'\n " \
- "break\n else\n echo 'vPing KO'\n fi\n sleep 1\ndone\n" % test_ip
+ u = ("#!/bin/sh\n\nwhile true; do\n ping -c 1 %s 2>&1 >/dev/null\n "
+ "RES=$?\n if [ \"Z$RES\" = \"Z0\" ] ; then\n echo 'vPing OK'\n "
+ "break\n else\n echo 'vPing KO'\n fi\n sleep 1\ndone\n" % test_ip)
# create VM
logger.info("Creating instance '%s'..." % NAME_VM_2)
logger.info("vPing duration:'%s'" % duration)
EXIT_CODE = 0
break
- elif "failed to read iid from metadata" in console_log or \
- metadata_tries > 5:
+ elif ("failed to read iid from metadata" in console_log or
+ metadata_tries > 5):
EXIT_CODE = -2
break
elif sec == PING_TIMEOUT:
Code Review / functest-xtesting.git / commitdiff