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 os.path
  14 import urllib2
  15 import subprocess
  16 import sys
  17 import requests
  18 import json
  19 import shutil
  20 import re
  21 import yaml
  22 from git import Repo
  23
  24
  25 # ############ CREDENTIALS OPENSTACK #############
  26 def check_credentials():
  27     """
  28     Check if the OpenStack credentials (openrc) are sourced
  29     """
  30     env_vars = ['OS_AUTH_URL', 'OS_USERNAME', 'OS_PASSWORD', 'OS_TENANT_NAME']
  31     return all(map(lambda v: v in os.environ and os.environ[v], env_vars))
  32
  33
  34 def get_credentials(service):
  35     """Returns a creds dictionary filled with the following keys:
  36     * username
  37     * password/api_key (depending on the service)
  38     * tenant_name/project_id (depending on the service)
  39     * auth_url
  40     :param service: a string indicating the name of the service
  41                     requesting the credentials.
  42     """
  43     creds = {}
  44     # Unfortunately, each of the OpenStack client will request slightly
  45     # different entries in their credentials dict.
  46     if service.lower() in ("nova", "cinder"):
  47         password = "api_key"
  48         tenant = "project_id"
  49     else:
  50         password = "password"
  51         tenant = "tenant_name"
  52
  53     # The most common way to pass these info to the script is to do it through
  54     # environment variables.
  55     creds.update({
  56         "username": os.environ.get('OS_USERNAME', "admin"),
  57         password: os.environ.get("OS_PASSWORD", 'admin'),
  58         "auth_url": os.environ.get("OS_AUTH_URL",
  59                                    "http://192.168.20.71:5000/v2.0"),
  60         tenant: os.environ.get("OS_TENANT_NAME", "admin"),
  61     })
  62
  63     return creds
  64
  65
  66 # ################ NOVA #################
  67 def get_instances(nova_client):
  68     try:
  69         instances = nova_client.servers.list(search_opts={'all_tenants': 1})
  70         return instances
  71     except:
  72         return None
  73
  74
  75 def get_instance_status(nova_client, instance):
  76     try:
  77         instance = nova_client.servers.get(instance.id)
  78         return instance.status
  79     except:
  80         return None
  81
  82
  83 def get_instance_by_name(nova_client, instance_name):
  84     try:
  85         instance = nova_client.servers.find(name=instance_name)
  86         return instance
  87     except:
  88         return None
  89
  90
  91 def get_flavor_id(nova_client, flavor_name):
  92     flavors = nova_client.flavors.list(detailed=True)
  93     id = ''
  94     for f in flavors:
  95         if f.name == flavor_name:
  96             id = f.id
  97             break
  98     return id
  99
 100
 101 def get_flavor_id_by_ram_range(nova_client, min_ram, max_ram):
 102     flavors = nova_client.flavors.list(detailed=True)
 103     id = ''
 104     for f in flavors:
 105         if min_ram <= f.ram and f.ram <= max_ram:
 106             id = f.id
 107             break
 108     return id
 109
 110
 111 def delete_instance(nova_client, instance_id):
 112     try:
 113         nova_client.servers.force_delete(instance_id)
 114         return True
 115     except:
 116         print "Error:", sys.exc_info()[0]
 117         return False
 118
 119 def create_floating_ip(neutron_client):
 120     extnet_id = get_external_net_id(neutron_client)
 121     props = {'floating_network_id': extnet_id}
 122     try:
 123         ip_json = neutron_client.create_floatingip({'floatingip': props})
 124         floating_ip = ip_json['floatingip']['floating_ip_address']
 125     except:
 126         return None
 127     return floating_ip
 128
 129 def get_floating_ips(nova_client):
 130     try:
 131         floating_ips = nova_client.floating_ips.list()
 132         return floating_ips
 133     except:
 134         return None
 135
 136 def add_floating_ip(nova_client, server_id, floatingip_id):
 137     try:
 138         nova_client.servers.add_floating_ip(server_id,floatingip_id)
 139         return True
 140     except:
 141         print "Error:", sys.exc_info()[0]
 142         return None
 143
 144 def delete_floating_ip(nova_client, floatingip_id):
 145     try:
 146         nova_client.floating_ips.delete(floatingip_id)
 147         return True
 148     except:
 149         print "Error:", sys.exc_info()[0]
 150         return None
 151
 152
 153 # ################ NEUTRON #################
 154 def create_neutron_net(neutron_client, name):
 155     json_body = {'network': {'name': name,
 156                              'admin_state_up': True}}
 157     try:
 158         network = neutron_client.create_network(body=json_body)
 159         network_dict = network['network']
 160         return network_dict['id']
 161     except:
 162         print "Error:", sys.exc_info()[0]
 163         return False
 164
 165
 166 def update_neutron_net(neutron_client, network_id, shared=False):
 167     json_body = {'network': {'shared': shared}}
 168     try:
 169         neutron_client.update_network(network_id, body=json_body)
 170         return True
 171     except:
 172         print "Error:", sys.exc_info()[0]
 173         return False
 174
 175
 176 def delete_neutron_net(neutron_client, network_id):
 177     try:
 178         neutron_client.delete_network(network_id)
 179         return True
 180     except:
 181         print "Error:", sys.exc_info()[0]
 182         return False
 183
 184
 185 def create_neutron_subnet(neutron_client, name, cidr, net_id):
 186     json_body = {'subnets': [{'name': name, 'cidr': cidr,
 187                               'ip_version': 4, 'network_id': net_id}]}
 188     try:
 189         subnet = neutron_client.create_subnet(body=json_body)
 190         return subnet['subnets'][0]['id']
 191     except:
 192         print "Error:", sys.exc_info()[0]
 193         return False
 194
 195
 196 def delete_neutron_subnet(neutron_client, subnet_id):
 197     try:
 198         neutron_client.delete_subnet(subnet_id)
 199         return True
 200     except:
 201         print "Error:", sys.exc_info()[0]
 202         return False
 203
 204
 205 def create_neutron_router(neutron_client, name):
 206     json_body = {'router': {'name': name, 'admin_state_up': True}}
 207     try:
 208         router = neutron_client.create_router(json_body)
 209         return router['router']['id']
 210     except:
 211         print "Error:", sys.exc_info()[0]
 212         return False
 213
 214
 215 def delete_neutron_router(neutron_client, router_id):
 216     json_body = {'router': {'id': router_id}}
 217     try:
 218         neutron_client.delete_router(router=router_id)
 219         return True
 220     except:
 221         print "Error:", sys.exc_info()[0]
 222         return False
 223
 224
 225 def add_interface_router(neutron_client, router_id, subnet_id):
 226     json_body = {"subnet_id": subnet_id}
 227     try:
 228         neutron_client.add_interface_router(router=router_id, body=json_body)
 229         return True
 230     except:
 231         print "Error:", sys.exc_info()[0]
 232         return False
 233
 234
 235 def remove_interface_router(neutron_client, router_id, subnet_id):
 236     json_body = {"subnet_id": subnet_id}
 237     try:
 238         neutron_client.remove_interface_router(router=router_id,
 239                                                body=json_body)
 240         return True
 241     except:
 242         print "Error:", sys.exc_info()[0]
 243         return False
 244
 245 def add_gateway_router(neutron_client, router_id):
 246     ext_net_id = get_external_net_id(neutron_client)
 247     router_dict = {'network_id': ext_net_id}
 248     try:
 249         neutron_client.add_gateway_router(router_id,router_dict)
 250         return True
 251     except:
 252         print "Error:", sys.exc_info()[0]
 253         return False
 254
 255 def remove_gateway_router(neutron_client, router_id):
 256     try:
 257         neutron_client.remove_gateway_router(router_id)
 258         return True
 259     except:
 260         print "Error:", sys.exc_info()[0]
 261         return False
 262
 263
 264 def create_neutron_port(neutron_client, name, network_id, ip):
 265     json_body = {'port': {
 266                  'admin_state_up': True,
 267                  'name': name,
 268                  'network_id': network_id,
 269                  'fixed_ips': [{"ip_address": ip}]
 270                  }}
 271     try:
 272         port = neutron_client.create_port(body=json_body)
 273         return port['port']['id']
 274     except:
 275         print "Error:", sys.exc_info()[0]
 276         return False
 277
 278
 279 def update_neutron_port(neutron_client, port_id, device_owner):
 280     json_body = {'port': {
 281                  'device_owner': device_owner,
 282                  }}
 283     try:
 284         port = neutron_client.update_port(port=port_id,
 285                                           body=json_body)
 286         return port['port']['id']
 287     except:
 288         print "Error:", sys.exc_info()[0]
 289         return False
 290
 291
 292 def delete_neutron_port(neutron_client, port_id):
 293     try:
 294         neutron_client.delete_port(port_id)
 295         return True
 296     except:
 297         print "Error:", sys.exc_info()[0]
 298         return False
 299
 300
 301 def get_network_id(neutron_client, network_name):
 302     networks = neutron_client.list_networks()['networks']
 303     id = ''
 304     for n in networks:
 305         if n['name'] == network_name:
 306             id = n['id']
 307             break
 308     return id
 309
 310
 311 def check_neutron_net(neutron_client, net_name):
 312     for network in neutron_client.list_networks()['networks']:
 313         if network['name'] == net_name:
 314             for subnet in network['subnets']:
 315                 return True
 316     return False
 317
 318
 319 def get_network_list(neutron_client):
 320     network_list = neutron_client.list_networks()['networks']
 321     if len(network_list) == 0:
 322         return None
 323     else:
 324         return network_list
 325
 326
 327 def get_router_list(neutron_client):
 328     router_list = neutron_client.list_routers()['routers']
 329     if len(router_list) == 0:
 330         return None
 331     else:
 332         return router_list
 333
 334
 335 def get_port_list(neutron_client):
 336     port_list = neutron_client.list_ports()['ports']
 337     if len(port_list) == 0:
 338         return None
 339     else:
 340         return port_list
 341
 342
 343 def get_external_net(neutron_client):
 344     for network in neutron_client.list_networks()['networks']:
 345         if network['router:external']:
 346             return network['name']
 347     return False
 348
 349
 350 def get_external_net_id(neutron_client):
 351     for network in neutron_client.list_networks()['networks']:
 352         if network['router:external']:
 353             return network['id']
 354     return False
 355
 356
 357 def update_sg_quota(neutron_client, tenant_id, sg_quota, sg_rule_quota):
 358     json_body = {"quota": {
 359         "security_group": sg_quota,
 360         "security_group_rule": sg_rule_quota
 361     }}
 362
 363     try:
 364         quota = neutron_client.update_quota(tenant_id=tenant_id,
 365                                             body=json_body)
 366         return True
 367     except:
 368         print "Error:", sys.exc_info()[0]
 369         return False
 370
 371
 372 def update_cinder_quota(cinder_client, tenant_id, vols_quota,
 373                         snapshots_quota, gigabytes_quota):
 374     quotas_values = {"volumes": vols_quota,
 375                      "snapshots": snapshots_quota,
 376                      "gigabytes": gigabytes_quota}
 377
 378     try:
 379         quotas_default = cinder_client.quotas.update(tenant_id,
 380                                                       **quotas_values)
 381         return True
 382     except:
 383         print "Error:", sys.exc_info()[0]
 384         return False
 385
 386
 387 def get_private_net(neutron_client):
 388     # Checks if there is an existing shared private network
 389     networks = neutron_client.list_networks()['networks']
 390     if len(networks) == 0:
 391         return None
 392     for net in networks:
 393         if (net['router:external'] is False) and (net['shared'] is True):
 394             return net
 395     return None
 396
 397
 398 # ################ GLANCE #################
 399 def get_images(nova_client):
 400     try:
 401         images = nova_client.images.list()
 402         return images
 403     except:
 404         return None
 405
 406
 407 def get_image_id(glance_client, image_name):
 408     images = glance_client.images.list()
 409     id = ''
 410     for i in images:
 411         if i.name == image_name:
 412             id = i.id
 413             break
 414     return id
 415
 416
 417 def create_glance_image(glance_client, image_name, file_path, public=True):
 418     if not os.path.isfile(file_path):
 419         print "Error: file " + file_path + " does not exist."
 420         return False
 421     try:
 422         with open(file_path) as fimage:
 423             image = glance_client.images.create(name=image_name,
 424                                                 is_public=public,
 425                                                 disk_format="qcow2",
 426                                                 container_format="bare",
 427                                                 data=fimage)
 428         return image.id
 429     except:
 430         print "Error:", sys.exc_info()[0]
 431         return False
 432
 433
 434 def delete_glance_image(nova_client, image_id):
 435     try:
 436         nova_client.images.delete(image_id)
 437         return True
 438     except:
 439         print "Error:", sys.exc_info()[0]
 440         return False
 441
 442
 443 # ################ CINDER #################
 444 def get_volumes(cinder_client):
 445     try:
 446         volumes = cinder_client.volumes.list(search_opts={'all_tenants': 1})
 447         return volumes
 448     except:
 449         return None
 450
 451
 452 def delete_volume(cinder_client, volume_id, forced=False):
 453     try:
 454         if forced:
 455             try:
 456                 cinder_client.volumes.detach(volume_id)
 457             except:
 458                 print "Error:", sys.exc_info()[0]
 459             cinder_client.volumes.force_delete(volume_id)
 460         else:
 461             cinder_client.volumes.delete(volume_id)
 462         return True
 463     except:
 464         print "Error:", sys.exc_info()[0]
 465         return False
 466
 467
 468 # ################ CINDER #################
 469 def get_security_groups(neutron_client):
 470     try:
 471         security_groups = neutron_client.list_security_groups()[
 472             'security_groups']
 473         return security_groups
 474     except:
 475         return None
 476
 477
 478 def delete_security_group(neutron_client, secgroup_id):
 479     try:
 480         neutron_client.delete_security_group(secgroup_id)
 481         return True
 482     except:
 483         print "Error:", sys.exc_info()[0]
 484         return False
 485
 486
 487 # ################ KEYSTONE #################
 488 def get_tenants(keystone_client):
 489     try:
 490         tenants = keystone_client.tenants.list()
 491         return tenants
 492     except:
 493         return None
 494
 495
 496 def get_tenant_id(keystone_client, tenant_name):
 497     tenants = keystone_client.tenants.list()
 498     id = ''
 499     for t in tenants:
 500         if t.name == tenant_name:
 501             id = t.id
 502             break
 503     return id
 504
 505
 506 def get_users(keystone_client):
 507     try:
 508         users = keystone_client.users.list()
 509         return users
 510     except:
 511         return None
 512
 513
 514 def get_role_id(keystone_client, role_name):
 515     roles = keystone_client.roles.list()
 516     id = ''
 517     for r in roles:
 518         if r.name == role_name:
 519             id = r.id
 520             break
 521     return id
 522
 523
 524 def get_user_id(keystone_client, user_name):
 525     users = keystone_client.users.list()
 526     id = ''
 527     for u in users:
 528         if u.name == user_name:
 529             id = u.id
 530             break
 531     return id
 532
 533
 534 def create_tenant(keystone_client, tenant_name, tenant_description):
 535     try:
 536         tenant = keystone_client.tenants.create(tenant_name,
 537                                                 tenant_description,
 538                                                 enabled=True)
 539         return tenant.id
 540     except:
 541         print "Error:", sys.exc_info()[0]
 542         return False
 543
 544
 545 def delete_tenant(keystone_client, tenant_id):
 546     try:
 547         tenant = keystone_client.tenants.delete(tenant_id)
 548         return True
 549     except:
 550         print "Error:", sys.exc_info()[0]
 551         return False
 552
 553
 554 def create_user(keystone_client, user_name, user_password,
 555                 user_email, tenant_id):
 556     try:
 557         user = keystone_client.users.create(user_name, user_password,
 558                                             user_email, tenant_id,
 559                                             enabled=True)
 560         return user.id
 561     except:
 562         print "Error:", sys.exc_info()[0]
 563         return False
 564
 565
 566 def delete_user(keystone_client, user_id):
 567     try:
 568         tenant = keystone_client.users.delete(user_id)
 569         return True
 570     except:
 571         print "Error:", sys.exc_info()[0]
 572         return False
 573
 574
 575 def add_role_user(keystone_client, user_id, role_id, tenant_id):
 576     try:
 577         keystone_client.roles.add_user_role(user_id, role_id, tenant_id)
 578         return True
 579     except:
 580         print "Error:", sys.exc_info()[0]
 581         return False
 582
 583
 584 # ################ UTILS #################
 585 def check_internet_connectivity(url='http://www.opnfv.org/'):
 586     """
 587     Check if there is access to the internet
 588     """
 589     try:
 590         urllib2.urlopen(url, timeout=5)
 591         return True
 592     except urllib2.URLError:
 593         return False
 594
 595
 596 def download_url(url, dest_path):
 597     """
 598     Download a file to a destination path given a URL
 599     """
 600     name = url.rsplit('/')[-1]
 601     dest = dest_path + "/" + name
 602     try:
 603         response = urllib2.urlopen(url)
 604     except (urllib2.HTTPError, urllib2.URLError):
 605         return False
 606
 607     with open(dest, 'wb') as f:
 608         shutil.copyfileobj(response, f)
 609     return True
 610
 611
 612 def execute_command(cmd, logger=None, exit_on_error=True):
 613     """
 614     Execute Linux command
 615     """
 616     if logger:
 617         logger.debug('Executing command : {}'.format(cmd))
 618     output_file = "output.txt"
 619     f = open(output_file, 'w+')
 620     p = subprocess.call(cmd, shell=True, stdout=f, stderr=subprocess.STDOUT)
 621     f.close()
 622     f = open(output_file, 'r')
 623     result = f.read()
 624     if result != "" and logger:
 625         logger.debug(result)
 626     if p == 0:
 627         return True
 628     else:
 629         if logger:
 630             logger.error("Error when executing command %s" % cmd)
 631         if exit_on_error:
 632             exit(-1)
 633         return False
 634
 635
 636 def get_git_branch(repo_path):
 637     """
 638     Get git branch name
 639     """
 640     repo = Repo(repo_path)
 641     branch = repo.active_branch
 642     return branch.name
 643
 644
 645 def get_installer_type(logger=None):
 646     """
 647     Get installer type (fuel, apex, joid, compass)
 648     """
 649     try:
 650         installer = os.environ['INSTALLER_TYPE']
 651     except KeyError:
 652         if logger:
 653             logger.error("Impossible to retrieve the installer type")
 654         installer = "Unkown"
 655
 656     return installer
 657
 658
 659 def get_pod_name(logger=None):
 660     """
 661     Get PoD Name from env variable NODE_NAME
 662     """
 663     try:
 664         return os.environ['NODE_NAME']
 665     except KeyError:
 666         if logger:
 667             logger.error(
 668                 "Unable to retrieve the POD name from environment.Using pod name 'unknown-pod'")
 669         return "unknown-pod"
 670
 671
 672 def push_results_to_db(db_url, case_name, logger, pod_name,
 673                        git_version, payload):
 674     url = db_url + "/results"
 675     installer = get_installer_type(logger)
 676     params = {"project_name": "functest", "case_name": case_name,
 677               "pod_name": pod_name, "installer": installer,
 678               "version": git_version, "details": payload}
 679
 680     headers = {'Content-Type': 'application/json'}
 681     try:
 682         r = requests.post(url, data=json.dumps(params), headers=headers)
 683         if logger:
 684             logger.debug(r)
 685         return True
 686     except:
 687         print "Error:", sys.exc_info()[0]
 688         return False
 689
 690
 691 def get_resolvconf_ns():
 692     nameservers = []
 693     rconf = open("/etc/resolv.conf", "r")
 694     line = rconf.readline()
 695     while line:
 696         ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line)
 697         if ip:
 698             nameservers.append(ip.group())
 699         line = rconf.readline()
 700     return nameservers
 701
 702 def getTestEnv(test, functest_yaml):
 703     # get the config of the testcase based on functest_config.yaml
 704     # 2 options
 705     # - test = test project e.g; ovno
 706     # - test = testcase e.g. functest/odl
 707     # look for the / to see if it is a test project or a testcase
 708     try:
 709         TEST_ENV = functest_yaml.get("test-dependencies")
 710
 711         if test.find("/") < 0:
 712             config_test = TEST_ENV[test]
 713         else:
 714             test_split = test.split("/")
 715             testproject = test_split[0]
 716             testcase = test_split[1]
 717             config_test = TEST_ENV[testproject][testcase]
 718     except KeyError:
 719         # if not defined in dependencies => no dependencies
 720         config_test = ""
 721     except:
 722         print "Error getTestEnv:", sys.exc_info()[0]
 723
 724     return config_test
 725
 726
 727 def get_ci_envvars():
 728     """
 729     Get the CI env variables
 730     """
 731     ci_env_var = {
 732         "installer": os.environ.get('INSTALLER_TYPE'),
 733         "scenario": os.environ.get('DEPLOY_SCENARIO')}
 734     return ci_env_var
 735
 736
 737 def isTestRunnable(test, functest_yaml):
 738     # By default we assume that all the tests are always runnable...
 739     is_runnable = True
 740     # Retrieve CI environment
 741     ci_env = get_ci_envvars()
 742     # Retrieve test environement from config file
 743     test_env = getTestEnv(test, functest_yaml)
 744
 745     # if test_env not empty => dependencies to be checked
 746     if test_env is not None and len(test_env) > 0:
 747         # possible criteria = ["installer", "scenario"]
 748         # consider test criteria from config file
 749         # compare towards CI env through CI en variable
 750         for criteria in test_env:
 751             if re.search(test_env[criteria], ci_env[criteria]) is None:
 752                 # print "Test "+ test + " cannot be run on the environment"
 753                 is_runnable = False
 754     return is_runnable
 755
 756
 757 def generateTestcaseList(functest_yaml):
 758     test_list = ""
 759     # get testcases
 760     testcase_list = functest_yaml.get("test-dependencies")
 761     projects = testcase_list.keys()
 762
 763     for project in projects:
 764         testcases = testcase_list[project]
 765         # 1 or 2 levels for testcases project[/case]
 766         # if only project name without controller or scenario
 767         # => shall be runnable on any controller/scenario
 768         if testcases is None:
 769             test_list += project + " "
 770         else:
 771             for testcase in testcases:
 772                 if testcase == "installer" or testcase == "scenario":
 773                     # project (1 level)
 774                     if isTestRunnable(project, functest_yaml):
 775                         test_list += project + " "
 776                 else:
 777                     # project/testcase (2 levels)
 778                     thetest = project + "/" + testcase
 779                     if isTestRunnable(thetest, functest_yaml):
 780                         test_list += testcase + " "
 781
 782     # sort the list to execute the test in the right order
 783     test_order_list = functest_yaml.get("test_exec_priority")
 784     test_sorted_list = ""
 785     for test in test_order_list:
 786         if test_order_list[test] in test_list:
 787             test_sorted_list += test_order_list[test] + " "
 788
 789     # create a file that could be consumed by run-test.sh
 790     # this method is used only for CI
 791     # so it can be run only in container
 792     # reuse default conf directory to store the list of runnable tests
 793     file = open("/home/opnfv/functest/conf/testcase-list.txt", 'w')
 794     file.write(test_sorted_list)
 795     file.close()
 796
 797     return test_sorted_list
 798