functest/utils/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 # pylint: disable=missing-docstring
  11
  12 from __future__ import print_function
  13 import logging
  14 import os
  15 import re
  16 import shutil
  17 import subprocess
  18 import sys
  19
  20 import pkg_resources
  21 import dns.resolver
  22 from six.moves import urllib
  23 import yaml
  24
  25 from functest.utils import constants
  26
  27 LOGGER = logging.getLogger(__name__)
  28
  29
  30 # ----------------------------------------------------------
  31 #
  32 #               INTERNET UTILS
  33 #
  34 # -----------------------------------------------------------
  35 def check_internet_connectivity(url='http://www.opnfv.org/'):
  36     """
  37     Check if there is access to the internet
  38     """
  39     try:
  40         urllib.request.urlopen(url, timeout=5)
  41         return True
  42     except urllib.error.URLError:
  43         return False
  44
  45
  46 def download_url(url, dest_path):
  47     """
  48     Download a file to a destination path given a URL
  49     """
  50     name = url.rsplit('/')[-1]
  51     dest = dest_path + "/" + name
  52     try:
  53         response = urllib.request.urlopen(url)
  54     except (urllib.error.HTTPError, urllib.error.URLError):
  55         return False
  56
  57     with open(dest, 'wb') as lfile:
  58         shutil.copyfileobj(response, lfile)
  59     return True
  60
  61
  62 # ----------------------------------------------------------
  63 #
  64 #               CI UTILS
  65 #
  66 # -----------------------------------------------------------
  67 def get_resolvconf_ns():
  68     """
  69     Get nameservers from current resolv.conf
  70     """
  71     nameservers = []
  72     rconf = open("/etc/resolv.conf", "r")
  73     line = rconf.readline()
  74     resolver = dns.resolver.Resolver()
  75     while line:
  76         addr_ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line)
  77         if addr_ip:
  78             resolver.nameservers = [addr_ip.group(0)]
  79             try:
  80                 result = resolver.query('opnfv.org')[0]
  81                 if result != "":
  82                     nameservers.append(addr_ip.group())
  83             except dns.exception.Timeout:
  84                 pass
  85         line = rconf.readline()
  86     return nameservers
  87
  88
  89 def get_ci_envvars():
  90     """
  91     Get the CI env variables
  92     """
  93     ci_env_var = {
  94         "installer": os.environ.get('INSTALLER_TYPE'),
  95         "scenario": os.environ.get('DEPLOY_SCENARIO')}
  96     return ci_env_var
  97
  98
  99 def execute_command_raise(cmd, info=False, error_msg="",
 100                           verbose=True, output_file=None):
 101     ret = execute_command(cmd, info, error_msg, verbose, output_file)
 102     if ret != 0:
 103         raise Exception(error_msg)
 104
 105
 106 def execute_command(cmd, info=False, error_msg="",
 107                     verbose=True, output_file=None):
 108     if not error_msg:
 109         error_msg = ("The command '%s' failed." % cmd)
 110     msg_exec = ("Executing command: '%s'" % cmd)
 111     if verbose:
 112         if info:
 113             LOGGER.info(msg_exec)
 114         else:
 115             LOGGER.debug(msg_exec)
 116     popen = subprocess.Popen(
 117         cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 118     if output_file:
 119         ofd = open(output_file, "w")
 120     for line in iter(popen.stdout.readline, b''):
 121         if output_file:
 122             ofd.write(line)
 123         else:
 124             line = line.replace('\n', '')
 125             print (line)
 126             sys.stdout.flush()
 127     if output_file:
 128         ofd.close()
 129     popen.stdout.close()
 130     returncode = popen.wait()
 131     if returncode != 0:
 132         if verbose:
 133             LOGGER.error(error_msg)
 134
 135     return returncode
 136
 137
 138 def get_dict_by_test(testname):
 139     # pylint: disable=bad-continuation
 140     with open(pkg_resources.resource_filename(
 141             'functest', 'ci/testcases.yaml')) as tyaml:
 142         testcases_yaml = yaml.safe_load(tyaml)
 143
 144     for dic_tier in testcases_yaml.get("tiers"):
 145         for dic_testcase in dic_tier['testcases']:
 146             if dic_testcase['case_name'] == testname:
 147                 return dic_testcase
 148
 149     LOGGER.error('Project %s is not defined in testcases.yaml', testname)
 150     return None
 151
 152
 153 def get_criteria_by_test(testname):
 154     tdict = get_dict_by_test(testname)
 155     if tdict:
 156         return tdict['criteria']
 157     return None
 158
 159
 160 # ----------------------------------------------------------
 161 #
 162 #               YAML UTILS
 163 #
 164 # -----------------------------------------------------------
 165 def get_parameter_from_yaml(parameter, yfile):
 166     """
 167     Returns the value of a given parameter in file.yaml
 168     parameter must be given in string format with dots
 169     Example: general.openstack.image_name
 170     """
 171     with open(yfile) as yfd:
 172         file_yaml = yaml.safe_load(yfd)
 173     value = file_yaml
 174     for element in parameter.split("."):
 175         value = value.get(element)
 176         if value is None:
 177             raise ValueError("The parameter %s is not defined in"
 178                              " %s" % (parameter, yfile))
 179     return value
 180
 181
 182 def get_functest_config(parameter):
 183     yaml_ = constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')
 184     return get_parameter_from_yaml(parameter, yaml_)
 185
 186
 187 def get_functest_yaml():
 188     # pylint: disable=bad-continuation
 189     with open(constants.CONST.__getattribute__(
 190             'CONFIG_FUNCTEST_YAML')) as yaml_fd:
 191         functest_yaml = yaml.safe_load(yaml_fd)
 192     return functest_yaml
 193
 194
 195 def print_separator():
 196     LOGGER.info("==============================================")