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