testcases/functest_utils.py

   1 #!/usr/bin/env python
   2 #
   3 # jose.lausuch@ericsson.com
   4 # valentin.boucher@orange.com
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10
  11 import re, json, os, urllib2, shutil, subprocess, sys
  12
  13 ############# CREDENTIALS OPENSTACK #############
  14 def check_credentials():
  15     """
  16     Check if the OpenStack credentials (openrc) are sourced
  17     """
  18     #TODO: there must be a short way to do this, doing if os.environ["something"] == "" throws an error
  19     try:
  20        os.environ['OS_AUTH_URL']
  21     except KeyError:
  22         return False
  23     try:
  24        os.environ['OS_USERNAME']
  25     except KeyError:
  26         return False
  27     try:
  28        os.environ['OS_PASSWORD']
  29     except KeyError:
  30         return False
  31     try:
  32        os.environ['OS_TENANT_NAME']
  33     except KeyError:
  34         return False
  35     return True
  36
  37
  38 def get_credentials(service):
  39     """Returns a creds dictionary filled with the following keys:
  40     * username
  41     * password/api_key (depending on the service)
  42     * tenant_name/project_id (depending on the service)
  43     * auth_url
  44     :param service: a string indicating the name of the service
  45                     requesting the credentials.
  46     """
  47     #TODO: get credentials from the openrc file
  48     creds = {}
  49     # Unfortunately, each of the OpenStack client will request slightly
  50     # different entries in their credentials dict.
  51     if service.lower() in ("nova", "cinder"):
  52         password = "api_key"
  53         tenant = "project_id"
  54     else:
  55         password = "password"
  56         tenant = "tenant_name"
  57
  58     # The most common way to pass these info to the script is to do it through
  59     # environment variables.
  60     creds.update({
  61         "username": os.environ.get('OS_USERNAME', "admin"),                                # add your cloud username details
  62         password: os.environ.get("OS_PASSWORD", 'admin'),                                # add password
  63         "auth_url": os.environ.get("OS_AUTH_URL","http://192.168.20.71:5000/v2.0"),        # Auth URL
  64         tenant: os.environ.get("OS_TENANT_NAME", "admin"),
  65     })
  66
  67     return creds
  68
  69
  70 ################# NOVA #################
  71 def get_instance_status(nova_client,instance):
  72     try:
  73         instance = nova_client.servers.get(instance.id)
  74         return instance.status
  75     except:
  76         return None
  77
  78
  79 def get_instance_by_name(nova_client, instance_name):
  80     try:
  81         instance = nova_client.servers.find(name=instance_name)
  82         return instance
  83     except:
  84         return None
  85
  86
  87 def get_flavor_id(nova_client, flavor_name):
  88     flavors = nova_client.flavors.list(detailed=True)
  89     id = ''
  90     for f in flavors:
  91         if f.name == flavor_name:
  92             id = f.id
  93             break
  94     return id
  95
  96
  97 def get_flavor_id_by_ram_range(nova_client, min_ram, max_ram):
  98     flavors = nova_client.flavors.list(detailed=True)
  99     id = ''
 100     for f in flavors:
 101         if min_ram <= f.ram and f.ram <= max_ram:
 102             id = f.id
 103             break
 104     return id
 105
 106
 107 ################# NEUTRON #################
 108 def create_neutron_net(neutron_client, name):
 109     json_body = {'network': {'name': name,
 110                     'admin_state_up': True}}
 111     try:
 112         network = neutron_client.create_network(body=json_body)
 113         network_dict = network['network']
 114         return network_dict['id']
 115     except:
 116         print "Error:", sys.exc_info()[0]
 117         return False
 118
 119 def delete_neutron_net(neutron_client, network_id):
 120     try:
 121         neutron_client.delete_network(network_id)
 122         return True
 123     except:
 124         print "Error:", sys.exc_info()[0]
 125         return False
 126
 127 def create_neutron_subnet(neutron_client, name, cidr, net_id):
 128     json_body = {'subnets': [{'name': name, 'cidr': cidr,
 129                            'ip_version': 4, 'network_id': net_id}]}
 130     try:
 131         subnet = neutron_client.create_subnet(body=json_body)
 132         return subnet['subnets'][0]['id']
 133     except:
 134         print "Error:", sys.exc_info()[0]
 135         return False
 136
 137 def delete_neutron_subnet(neutron_client, subnet_id):
 138     try:
 139         neutron_client.delete_subnet(subnet_id)
 140         return True
 141     except:
 142         print "Error:", sys.exc_info()[0]
 143         return False
 144
 145 def create_neutron_router(neutron_client, name):
 146     json_body = {'router': {'name': name, 'admin_state_up': True}}
 147     try:
 148         router = neutron_client.create_router(json_body)
 149         return router['router']['id']
 150     except:
 151         print "Error:", sys.exc_info()[0]
 152         return False
 153
 154 def delete_neutron_router(neutron_client, router_id):
 155     json_body = {'router': {'id': router_id}}
 156     try:
 157         neutron_client.delete_router(router=router_id)
 158         return True
 159     except:
 160         print "Error:", sys.exc_info()[0]
 161         return False
 162
 163
 164 def add_interface_router(neutron_client, router_id, subnet_id):
 165     json_body = {"subnet_id": subnet_id}
 166     try:
 167         neutron_client.add_interface_router(router=router_id, body=json_body)
 168         return True
 169     except:
 170         print "Error:", sys.exc_info()[0]
 171         return False
 172
 173
 174 def remove_interface_router(neutron_client, router_id, subnet_id):
 175     json_body = {"subnet_id": subnet_id}
 176     try:
 177         neutron_client.remove_interface_router(router=router_id, body=json_body)
 178         return True
 179     except:
 180         print "Error:", sys.exc_info()[0]
 181         return False
 182
 183 def create_neutron_port(neutron_client, name, network_id, ip):
 184     json_body = {'port': {
 185     'admin_state_up': True,
 186     'name': name,
 187     'network_id': network_id,
 188     'fixed_ips': [{"ip_address": ip}]
 189     }}
 190     try:
 191         port = neutron_client.create_port(body=json_body)
 192         return port['port']['id']
 193     except:
 194         print "Error:", sys.exc_info()[0]
 195         return False
 196
 197
 198 def get_network_id(neutron_client, network_name):
 199     networks = neutron_client.list_networks()['networks']
 200     id  = ''
 201     for n in networks:
 202         if n['name'] == network_name:
 203             id = n['id']
 204             break
 205     return id
 206
 207 def check_neutron_net(neutron_client, net_name):
 208     for network in neutron_client.list_networks()['networks']:
 209         if network['name'] == net_name :
 210             for subnet in network['subnets']:
 211                 return True
 212     return False
 213
 214 def get_network_list(neutron_client):
 215     network_list = neutron_client.list_networks()['networks']
 216     if len(network_list) == 0 :
 217         return None
 218     else :
 219         return network_list
 220
 221 def get_external_net(neutron_client):
 222     for network in neutron_client.list_networks()['networks']:
 223         if network['router:external']:
 224             return network['name']
 225     return False
 226
 227 def update_sg_quota(neutron_client, tenant_id, sg_quota, sg_rule_quota):
 228     json_body = {"quota": {
 229         "security_group": sg_quota,
 230         "security_group_rule": sg_rule_quota
 231     }}
 232
 233     try:
 234         quota = neutron_client.update_quota(tenant_id=tenant_id, body=json_body)
 235         return True
 236     except:
 237         print "Error:", sys.exc_info()[0]
 238         return False
 239
 240 ################# GLANCE #################
 241
 242 def get_image_id(glance_client, image_name):
 243     images = glance_client.images.list()
 244     id = ''
 245     for i in images:
 246         if i.name == image_name:
 247             id = i.id
 248             break
 249     return id
 250
 251 def create_glance_image(glance_client, image_name, file_path, is_public=True):
 252     try:
 253         with open(file_path) as fimage:
 254             image = glance_client.images.create(name=image_name, is_public=is_public, disk_format="qcow2",
 255                              container_format="bare", data=fimage)
 256         return image.id
 257     except:
 258         return False
 259
 260
 261 ################# KEYSTONE #################
 262 def get_tenant_id(keystone_client, tenant_name):
 263     tenants = keystone_client.tenants.list()
 264     id = ''
 265     for t in tenants:
 266         if t.name == tenant_name:
 267             id = t.id
 268             break
 269     return id
 270
 271 def get_role_id(keystone_client, role_name):
 272     roles = keystone_client.roles.list()
 273     id = ''
 274     for r in roles:
 275         if r.name == role_name:
 276             id = r.id
 277             break
 278     return id
 279
 280 def get_user_id(keystone_client, user_name):
 281     users = keystone_client.users.list()
 282     id = ''
 283     for u in users:
 284         if u.name == user_name:
 285             id = u.id
 286             break
 287     return id
 288
 289 def create_tenant(keystone_client, tenant_name, tenant_description):
 290     try:
 291         tenant = keystone_client.tenants.create(tenant_name, tenant_description, enabled=True)
 292         return tenant.id
 293     except:
 294         print "Error:", sys.exc_info()[0]
 295         return False
 296
 297 def delete_tenant(keystone_client, tenant_id):
 298     try:
 299         tenant = keystone_client.tenants.delete(tenant_id)
 300         return True
 301     except:
 302         print "Error:", sys.exc_info()[0]
 303         return False
 304
 305 def create_user(keystone_client, user_name, user_password, user_email, tenant_id):
 306     try:
 307         user = keystone_client.users.create(user_name, user_password, user_email, tenant_id, enabled=True)
 308         return user.id
 309     except:
 310         print "Error:", sys.exc_info()[0]
 311         return False
 312
 313 def delete_user(keystone_client, user_id):
 314     try:
 315         tenant = keystone_client.users.delete(user_id)
 316         return True
 317     except:
 318         print "Error:", sys.exc_info()[0]
 319         return False
 320
 321 def add_role_user(keystone_client, user_id, role_id, tenant_id):
 322     try:
 323         keystone_client.roles.add_user_role(user_id, role_id, tenant_id)
 324         return True
 325     except:
 326         print "Error:", sys.exc_info()[0]
 327         return False
 328
 329
 330 ################# UTILS #################
 331 def check_internet_connectivity(url='http://www.opnfv.org/'):
 332     """
 333     Check if there is access to the internet
 334     """
 335     try:
 336         urllib2.urlopen(url, timeout=5)
 337         return True
 338     except urllib.URLError:
 339         return False
 340
 341
 342 def download_url(url, dest_path):
 343     """
 344     Download a file to a destination path given a URL
 345     """
 346     name = url.rsplit('/')[-1]
 347     dest = dest_path + name
 348     try:
 349         response = urllib2.urlopen(url)
 350     except (urllib2.HTTPError, urllib2.URLError):
 351         return False
 352
 353     with open(dest, 'wb') as f:
 354         f.write(response.read())
 355     return True
 356
 357
 358 def execute_command(cmd, logger=None):
 359     """
 360     Execute Linux command
 361     """
 362     if logger:
 363         logger.debug('Executing command : {}'.format(cmd))
 364     output_file = "output.txt"
 365     f = open(output_file, 'w+')
 366     p = subprocess.call(cmd,shell=True, stdout=f, stderr=subprocess.STDOUT)
 367     f.close()
 368     f = open(output_file, 'r')
 369     result = f.read()
 370     if result != "" and logger:
 371         logger.debug(result)
 372     if p == 0 :
 373         return True
 374     else:
 375         if logger:
 376             logger.error("Error when executing command %s" %cmd)
 377         exit(-1)