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