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