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