diff options
Diffstat (limited to 'testcases/functest_utils.py')
| -rw-r--r-- | testcases/functest_utils.py | 692 |
1 files changed, 409 insertions, 283 deletions
diff --git a/testcases/functest_utils.py b/testcases/functest_utils.py index ed9e36c89..da79c2c9a 100644 --- a/testcases/functest_utils.py +++ b/testcases/functest_utils.py @@ -8,22 +8,30 @@ # http://www.apache.org/licenses/LICENSE-2.0 # - +import json import os import os.path -import urllib2 -import subprocess -import sys +import re import requests -import json import shutil -import re -import yaml import socket +import subprocess +import sys +import urllib2 +import yaml from git import Repo -# ############ CREDENTIALS OPENSTACK ############# +# ---------------------------------------------------------- +# +# OPENSTACK UTILS +# +# ----------------------------------------------------------- + + +#********************************************* +# CREDENTIALS +#********************************************* def check_credentials(): """ Check if the OpenStack credentials (openrc) are sourced @@ -31,7 +39,6 @@ def check_credentials(): env_vars = ['OS_AUTH_URL', 'OS_USERNAME', 'OS_PASSWORD', 'OS_TENANT_NAME'] return all(map(lambda v: v in os.environ and os.environ[v], env_vars)) - def get_credentials(service): """Returns a creds dictionary filled with the following keys: * username @@ -60,16 +67,19 @@ def get_credentials(service): "http://192.168.20.71:5000/v2.0"), tenant: os.environ.get("OS_TENANT_NAME", "admin"), }) - return creds -# ################ NOVA ################# + +#********************************************* +# NOVA +#********************************************* def get_instances(nova_client): try: instances = nova_client.servers.list(search_opts={'all_tenants': 1}) return instances - except: + except Exception, e: + print "Error [get_instances(nova_client)]:", e return None @@ -77,7 +87,9 @@ def get_instance_status(nova_client, instance): try: instance = nova_client.servers.get(instance.id) return instance.status - except: + except Exception, e: + print "Error [get_instance_status(nova_client, '%s')]:" % \ + str(instance), e return None @@ -85,15 +97,11 @@ def get_instance_by_name(nova_client, instance_name): try: instance = nova_client.servers.find(name=instance_name) return instance - except: + except Exception, e: + print "Error [get_instance_by_name(nova_client, '%s')]:" % \ + instance_name, e return None -def create_flavor(nova_client, flavor_name, ram, disk, vcpus): - try: - flavor = nova_client.flavors.create(flavor_name,ram,vcpus,disk) - except: - return None - return flavor.id def get_flavor_id(nova_client, flavor_name): flavors = nova_client.flavors.list(detailed=True) @@ -115,13 +123,24 @@ def get_flavor_id_by_ram_range(nova_client, min_ram, max_ram): return id -def delete_instance(nova_client, instance_id): +def get_floating_ips(nova_client): try: - nova_client.servers.force_delete(instance_id) - return True - except: - print "Error:", sys.exc_info()[0] - return False + floating_ips = nova_client.floating_ips.list() + return floating_ips + except Exception, e: + print "Error [get_floating_ips(nova_client)]:", e + return None + + +def create_flavor(nova_client, flavor_name, ram, disk, vcpus): + 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 + return None + return flavor.id + def create_floating_ip(neutron_client): extnet_id = get_external_net_id(neutron_client) @@ -129,35 +148,112 @@ def create_floating_ip(neutron_client): try: ip_json = neutron_client.create_floatingip({'floatingip': props}) floating_ip = ip_json['floatingip']['floating_ip_address'] - except: + except Exception, e: + print "Error [create_floating_ip(neutron_client)]:", e return None return floating_ip -def get_floating_ips(nova_client): - try: - floating_ips = nova_client.floating_ips.list() - return floating_ips - except: - return None def add_floating_ip(nova_client, server_id, floatingip_id): try: nova_client.servers.add_floating_ip(server_id,floatingip_id) return True - except: - print "Error:", sys.exc_info()[0] - return None + except Exception, e: + print "Error [add_floating_ip(nova_client, '%s', '%s')]:" % \ + (server_id, floatingip_id), e + return False + + +def delete_instance(nova_client, instance_id): + try: + nova_client.servers.force_delete(instance_id) + return True + except Exception, e: + print "Error [delete_instance(nova_client, '%s')]:" % instance_id, e + return False + def delete_floating_ip(nova_client, floatingip_id): try: nova_client.floating_ips.delete(floatingip_id) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [delete_floating_ip(nova_client, '%s')]:" % floatingip_id, e + return False + + + + +#********************************************* +# NEUTRON +#********************************************* +def get_network_list(neutron_client): + network_list = neutron_client.list_networks()['networks'] + if len(network_list) == 0: + return None + else: + return network_list + + +def get_router_list(neutron_client): + router_list = neutron_client.list_routers()['routers'] + if len(router_list) == 0: + return None + else: + return router_list + + +def get_port_list(neutron_client): + port_list = neutron_client.list_ports()['ports'] + if len(port_list) == 0: + return None + else: + return port_list + + +def get_network_id(neutron_client, network_name): + networks = neutron_client.list_networks()['networks'] + id = '' + for n in networks: + if n['name'] == network_name: + id = n['id'] + break + return id + + +def get_private_net(neutron_client): + # Checks if there is an existing shared private network + networks = neutron_client.list_networks()['networks'] + if len(networks) == 0: return None + for net in networks: + if (net['router:external'] is False) and (net['shared'] is True): + return net + return None + + +def get_external_net(neutron_client): + for network in neutron_client.list_networks()['networks']: + if network['router:external']: + return network['name'] + return False + + +def get_external_net_id(neutron_client): + for network in neutron_client.list_networks()['networks']: + if network['router:external']: + return network['id'] + return False + + +def check_neutron_net(neutron_client, net_name): + for network in neutron_client.list_networks()['networks']: + if network['name'] == net_name: + for subnet in network['subnets']: + return True + return False -# ################ NEUTRON ################# def create_neutron_net(neutron_client, name): json_body = {'network': {'name': name, 'admin_state_up': True}} @@ -165,27 +261,8 @@ def create_neutron_net(neutron_client, name): network = neutron_client.create_network(body=json_body) network_dict = network['network'] return network_dict['id'] - except: - print "Error:", sys.exc_info()[0] - return False - - -def update_neutron_net(neutron_client, network_id, shared=False): - json_body = {'network': {'shared': shared}} - try: - neutron_client.update_network(network_id, body=json_body) - return True - except: - print "Error:", sys.exc_info()[0] - return False - - -def delete_neutron_net(neutron_client, network_id): - try: - neutron_client.delete_network(network_id) - return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [create_neutron_net(neutron_client, '%s')]:" % name, e return False @@ -195,17 +272,9 @@ def create_neutron_subnet(neutron_client, name, cidr, net_id): try: subnet = neutron_client.create_subnet(body=json_body) return subnet['subnets'][0]['id'] - except: - print "Error:", sys.exc_info()[0] - return False - - -def delete_neutron_subnet(neutron_client, subnet_id): - try: - neutron_client.delete_subnet(subnet_id) - return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [create_neutron_subnet(neutron_client, '%s', '%s', "\ + "'%s')]:" %(name,cidr, net_id), e return False @@ -214,39 +283,60 @@ def create_neutron_router(neutron_client, name): try: router = neutron_client.create_router(json_body) return router['router']['id'] - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [create_neutron_router(neutron_client, '%s')]:" % name, e return False -def delete_neutron_router(neutron_client, router_id): - json_body = {'router': {'id': router_id}} +def create_neutron_port(neutron_client, name, network_id, ip): + json_body = {'port': { + 'admin_state_up': True, + 'name': name, + 'network_id': network_id, + 'fixed_ips': [{"ip_address": ip}] + }} try: - neutron_client.delete_router(router=router_id) - return True - except: - print "Error:", sys.exc_info()[0] + 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 return False -def add_interface_router(neutron_client, router_id, subnet_id): - json_body = {"subnet_id": subnet_id} +def update_neutron_net(neutron_client, network_id, shared=False): + json_body = {'network': {'shared': shared}} try: - neutron_client.add_interface_router(router=router_id, body=json_body) + neutron_client.update_network(network_id, body=json_body) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [update_neutron_net(neutron_client, '%s', '%s')]:" % \ + (network_id,str(shared)), e return False -def remove_interface_router(neutron_client, router_id, subnet_id): +def update_neutron_port(neutron_client, port_id, device_owner): + json_body = {'port': { + 'device_owner': device_owner, + }} + try: + port = neutron_client.update_port(port=port_id, + body=json_body) + return port['port']['id'] + except Exception, e: + print "Error [update_neutron_port(neutron_client, '%s', '%s')]:" % \ + (port_id,device_owner), e + return False + + +def add_interface_router(neutron_client, router_id, subnet_id): json_body = {"subnet_id": subnet_id} try: - neutron_client.remove_interface_router(router=router_id, - body=json_body) + neutron_client.add_interface_router(router=router_id, body=json_body) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [add_interface_router(neutron_client, '%s', '%s')]:" % \ + (router_id,subnet_id), e return False def add_gateway_router(neutron_client, router_id): @@ -255,44 +345,37 @@ def add_gateway_router(neutron_client, router_id): try: neutron_client.add_gateway_router(router_id,router_dict) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [add_gateway_router(neutron_client, '%s')]:" % router_id, e return False -def remove_gateway_router(neutron_client, router_id): + +def delete_neutron_net(neutron_client, network_id): try: - neutron_client.remove_gateway_router(router_id) + neutron_client.delete_network(network_id) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [delete_neutron_net(neutron_client, '%s')]:" % network_id, e return False -def create_neutron_port(neutron_client, name, network_id, ip): - json_body = {'port': { - 'admin_state_up': True, - 'name': name, - 'network_id': network_id, - 'fixed_ips': [{"ip_address": ip}] - }} +def delete_neutron_subnet(neutron_client, subnet_id): try: - port = neutron_client.create_port(body=json_body) - return port['port']['id'] - except: - print "Error:", sys.exc_info()[0] + neutron_client.delete_subnet(subnet_id) + return True + except Exception, e: + print "Error [delete_neutron_subnet(neutron_client, '%s')]:" % subnet_id, e return False -def update_neutron_port(neutron_client, port_id, device_owner): - json_body = {'port': { - 'device_owner': device_owner, - }} +def delete_neutron_router(neutron_client, router_id): + json_body = {'router': {'id': router_id}} try: - port = neutron_client.update_port(port=port_id, - body=json_body) - return port['port']['id'] - except: - print "Error:", sys.exc_info()[0] + neutron_client.delete_router(router=router_id) + return True + except Exception, e: + print "Error [delete_neutron_router(neutron_client, '%s')]:" % \ + router_id, e return False @@ -300,65 +383,44 @@ def delete_neutron_port(neutron_client, port_id): try: neutron_client.delete_port(port_id) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [delete_neutron_port(neutron_client, '%s')]:" % port_id, e return False -def get_network_id(neutron_client, network_name): - networks = neutron_client.list_networks()['networks'] - id = '' - for n in networks: - if n['name'] == network_name: - id = n['id'] - break - return id - - -def check_neutron_net(neutron_client, net_name): - for network in neutron_client.list_networks()['networks']: - if network['name'] == net_name: - for subnet in network['subnets']: - return True - return False - +def remove_interface_router(neutron_client, router_id, subnet_id): + json_body = {"subnet_id": subnet_id} + try: + neutron_client.remove_interface_router(router=router_id, + body=json_body) + return True + except Exception, e: + print "Error [remove_interface_router(neutron_client, '%s', '%s')]:" % \ + (router_id,subnet_id), e + return False -def get_network_list(neutron_client): - network_list = neutron_client.list_networks()['networks'] - if len(network_list) == 0: - return None - else: - return network_list +def remove_gateway_router(neutron_client, router_id): + try: + neutron_client.remove_gateway_router(router_id) + return True + except Exception, e: + print "Error [remove_gateway_router(neutron_client, '%s')]:" % router_id, e + return False -def get_router_list(neutron_client): - router_list = neutron_client.list_routers()['routers'] - if len(router_list) == 0: - return None - else: - return router_list -def get_port_list(neutron_client): - port_list = neutron_client.list_ports()['ports'] - if len(port_list) == 0: +#********************************************* +# SEC GROUPS +#********************************************* +def get_security_groups(neutron_client): + try: + security_groups = neutron_client.list_security_groups()[ + 'security_groups'] + return security_groups + except Exception, e: + print "Error [get_security_groups(neutron_client)]:", e return None - else: - return port_list - - -def get_external_net(neutron_client): - for network in neutron_client.list_networks()['networks']: - if network['router:external']: - return network['name'] - return False - - -def get_external_net_id(neutron_client): - for network in neutron_client.list_networks()['networks']: - if network['router:external']: - return network['id'] - return False def update_sg_quota(neutron_client, tenant_id, sg_quota, sg_rule_quota): @@ -371,43 +433,31 @@ def update_sg_quota(neutron_client, tenant_id, sg_quota, sg_rule_quota): quota = neutron_client.update_quota(tenant_id=tenant_id, body=json_body) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [update_sg_quota(neutron_client, '%s', '%s', "\ + "'%s')]:" %(tenant_id, sg_quota, sg_rule_quota), e return False -def update_cinder_quota(cinder_client, tenant_id, vols_quota, - snapshots_quota, gigabytes_quota): - quotas_values = {"volumes": vols_quota, - "snapshots": snapshots_quota, - "gigabytes": gigabytes_quota} - +def delete_security_group(neutron_client, secgroup_id): try: - quotas_default = cinder_client.quotas.update(tenant_id, - **quotas_values) + neutron_client.delete_security_group(secgroup_id) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [delete_security_group(neutron_client, '%s')]:" % secgroup_id, e return False -def get_private_net(neutron_client): - # Checks if there is an existing shared private network - networks = neutron_client.list_networks()['networks'] - if len(networks) == 0: - return None - for net in networks: - if (net['router:external'] is False) and (net['shared'] is True): - return net - return None - -# ################ GLANCE ################# +#********************************************* +# GLANCE +#********************************************* def get_images(nova_client): try: images = nova_client.images.list() return images - except: + except Exception, e: + print "Error [get_images]:", e return None @@ -433,8 +483,9 @@ def create_glance_image(glance_client, image_name, file_path, public=True): container_format="bare", data=fimage) return image.id - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [create_glance_image(glance_client, '%s', '%s', "\ + "'%s')]:" %(image_name, file_path, str(public)), e return False @@ -442,20 +493,62 @@ def delete_glance_image(nova_client, image_id): try: nova_client.images.delete(image_id) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [delete_glance_image(nova_client, '%s')]:" % image_id, e return False -# ################ CINDER ################# + +#********************************************* +# CINDER +#********************************************* def get_volumes(cinder_client): try: volumes = cinder_client.volumes.list(search_opts={'all_tenants': 1}) return volumes - except: + except Exception, e: + print "Error [get_volumes(cinder_client)]:", e + return None + + +def list_volume_types(cinder_client, public=True, private=True): + try: + volume_types = cinder_client.volume_types.list() + if not public: + volume_types = [vt for vt in volume_types if not vt.is_public] + if not private: + volume_types = [vt for vt in volume_types if vt.is_public] + return volume_types + except Exception, e: + print "Error [list_volume_types(cinder_client)]:", e return None +def create_volume_type(cinder_client, name): + try: + volume_type = cinder_client.volume_types.create(name) + return volume_type + except Exception, e: + print "Error [create_volume_type(cinder_client, '%s')]:" % name, e + return None + + +def update_cinder_quota(cinder_client, tenant_id, vols_quota, + snapshots_quota, gigabytes_quota): + quotas_values = {"volumes": vols_quota, + "snapshots": snapshots_quota, + "gigabytes": gigabytes_quota} + + try: + quotas_default = 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 + return False + + def delete_volume(cinder_client, volume_id, forced=False): try: if forced: @@ -467,63 +560,49 @@ def delete_volume(cinder_client, volume_id, forced=False): else: cinder_client.volumes.delete(volume_id) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [delete_volume(cinder_client, '%s', '%s')]:" % \ + (volume_id, str(forced)), e return False -# ################ CINDER ################# -def get_security_groups(neutron_client): +def delete_volume_type(cinder_client, volume_type): try: - security_groups = neutron_client.list_security_groups()[ - 'security_groups'] - return security_groups - except: - return None - - -def delete_security_group(neutron_client, secgroup_id): - try: - neutron_client.delete_security_group(secgroup_id) + cinder_client.volume_types.delete(volume_type) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [delete_volume_type(cinder_client, '%s')]:" % volume_type, e return False -# ################ KEYSTONE ################# + +#********************************************* +# KEYSTONE +#********************************************* def get_tenants(keystone_client): try: tenants = keystone_client.tenants.list() return tenants - except: + except Exception, e: + print "Error [get_tenants(keystone_client)]:", e return None -def get_tenant_id(keystone_client, tenant_name): - tenants = keystone_client.tenants.list() - id = '' - for t in tenants: - if t.name == tenant_name: - id = t.id - break - return id - - def get_users(keystone_client): try: users = keystone_client.users.list() return users - except: + except Exception, e: + print "Error [get_users(keystone_client)]:", e return None -def get_role_id(keystone_client, role_name): - roles = keystone_client.roles.list() +def get_tenant_id(keystone_client, tenant_name): + tenants = keystone_client.tenants.list() id = '' - for r in roles: - if r.name == role_name: - id = r.id + for t in tenants: + if t.name == tenant_name: + id = t.id break return id @@ -538,23 +617,25 @@ def get_user_id(keystone_client, user_name): return id +def get_role_id(keystone_client, role_name): + roles = keystone_client.roles.list() + id = '' + for r in roles: + if r.name == role_name: + id = r.id + break + return id + + def create_tenant(keystone_client, tenant_name, tenant_description): try: tenant = keystone_client.tenants.create(tenant_name, tenant_description, enabled=True) return tenant.id - except: - print "Error:", sys.exc_info()[0] - return False - - -def delete_tenant(keystone_client, tenant_id): - try: - tenant = keystone_client.tenants.delete(tenant_id) - return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [create_tenant(cinder_client, '%s', '%s')]:" % \ + (tenant_name, tenant_description), e return False @@ -565,30 +646,50 @@ def create_user(keystone_client, user_name, user_password, user_email, tenant_id, enabled=True) return user.id - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [create_user(keystone_client, '%s', '%s', '%s'" \ + "'%s')]:" %(user_name, user_password, user_email, tenant_id), e return False -def delete_user(keystone_client, user_id): +def add_role_user(keystone_client, user_id, role_id, tenant_id): try: - tenant = keystone_client.users.delete(user_id) + keystone_client.roles.add_user_role(user_id, role_id, tenant_id) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [add_role_user(keystone_client, '%s', '%s'" \ + "'%s')]:" %(user_id, role_id, tenant_id), e return False -def add_role_user(keystone_client, user_id, role_id, tenant_id): +def delete_tenant(keystone_client, tenant_id): try: - keystone_client.roles.add_user_role(user_id, role_id, tenant_id) + tenant = keystone_client.tenants.delete(tenant_id) + return True + except Exception, e: + print "Error [delete_tenant(keystone_client, '%s')]:" % tenant_id, e + return False + + +def delete_user(keystone_client, user_id): + try: + tenant = keystone_client.users.delete(user_id) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [delete_user(keystone_client, '%s')]:" % user_id, e return False -# ################ UTILS ################# + + + + + +# ---------------------------------------------------------- +# +# INTERNET UTILS +# +# ----------------------------------------------------------- def check_internet_connectivity(url='http://www.opnfv.org/'): """ Check if there is access to the internet @@ -616,30 +717,13 @@ def download_url(url, dest_path): return True -def execute_command(cmd, logger=None, exit_on_error=True): - """ - Execute Linux command - """ - if logger: - logger.debug('Executing command : {}'.format(cmd)) - output_file = "output.txt" - f = open(output_file, 'w+') - p = subprocess.call(cmd, shell=True, stdout=f, stderr=subprocess.STDOUT) - f.close() - f = open(output_file, 'r') - result = f.read() - if result != "" and logger: - logger.debug(result) - if p == 0: - return True - else: - if logger: - logger.error("Error when executing command %s" % cmd) - if exit_on_error: - exit(-1) - return False +# ---------------------------------------------------------- +# +# CI UTILS +# +# ----------------------------------------------------------- def get_git_branch(repo_path): """ Get git branch name @@ -692,6 +776,9 @@ def get_pod_name(logger=None): def push_results_to_db(db_url, case_name, logger, pod_name, version, payload): + """ + POST results to the Result target DB + """ url = db_url + "/results" installer = get_installer_type(logger) params = {"project_name": "functest", "case_name": case_name, @@ -704,12 +791,16 @@ def push_results_to_db(db_url, case_name, logger, pod_name, if logger: logger.debug(r) return True - except: - print "Error:", sys.exc_info()[0] + except Exception, e: + print "Error [push_results_to_db('%s', '%s', '%s', '%s', '%s')]:" \ + % (db_url, case_name, pod_name, version, payload), e return False def get_resolvconf_ns(): + """ + Get nameservers from current resolv.conf + """ nameservers = [] rconf = open("/etc/resolv.conf", "r") line = rconf.readline() @@ -723,12 +814,15 @@ def get_resolvconf_ns(): line = rconf.readline() return nameservers + def getTestEnv(test, functest_yaml): - # get the config of the testcase based on functest_config.yaml - # 2 options - # - test = test project e.g; ovno - # - test = testcase e.g. functest/odl - # look for the / to see if it is a test project or a testcase + """ + Get the config of the testcase based on functest_config.yaml + 2 options + - test = test project e.g; ovno + - test = testcase e.g. functest/odl + look for the / to see if it is a test project or a testcase + """ try: TEST_ENV = functest_yaml.get("test-dependencies") @@ -742,8 +836,8 @@ def getTestEnv(test, functest_yaml): except KeyError: # if not defined in dependencies => no dependencies config_test = "" - except: - print "Error getTestEnv:", sys.exc_info()[0] + except Exception, e: + print "Error [getTestEnv]:", e return config_test @@ -759,6 +853,9 @@ def get_ci_envvars(): def isTestRunnable(test, functest_yaml): + """ + Return True if the test is runnable in the current scenario + """ # By default we assume that all the tests are always runnable... is_runnable = True # Retrieve CI environment @@ -779,6 +876,10 @@ def isTestRunnable(test, functest_yaml): def generateTestcaseList(functest_yaml): + """ + Generate a test file with the runnable test according to + the current scenario + """ test_list = "" # get testcases testcase_list = functest_yaml.get("test-dependencies") @@ -786,7 +887,7 @@ def generateTestcaseList(functest_yaml): for project in projects: testcases = testcase_list[project] - # 1 or 2 levels for testcases project[/case] + # 1 or 2 levels for testcases project[/case]l # if only project name without controller or scenario # => shall be runnable on any controller/scenario if testcases is None: @@ -820,3 +921,28 @@ def generateTestcaseList(functest_yaml): return test_sorted_list + +def execute_command(cmd, logger=None, exit_on_error=True): + """ + Execute Linux command + prints stdout to a file and depending on if there + is a logger defined, it will print it or not. + """ + if logger: + logger.debug('Executing command : {}'.format(cmd)) + output_file = "output.txt" + f = open(output_file, 'w+') + p = subprocess.call(cmd, shell=True, stdout=f, stderr=subprocess.STDOUT) + f.close() + f = open(output_file, 'r') + result = f.read() + if result != "" and logger: + logger.debug(result) + if p == 0: + return True + else: + if logger: + logger.error("Error when executing command %s" % cmd) + if exit_on_error: + exit(-1) + return False |