# pylint: disable=missing-docstring
import logging
-import os
import time
import unittest
import mock
import pkg_resources
-from six.moves import urllib
from functest.utils import functest_utils
self.file_yaml = {'general': {'openstack': {'image_name':
'test_image_name'}}}
- @mock.patch('six.moves.urllib.request.urlopen',
- side_effect=urllib.error.URLError('no host given'))
- def test_check_internet_connectivity_failed(self, mock_method):
- self.assertFalse(functest_utils.check_internet_connectivity())
- mock_method.assert_called_once_with(self.url, timeout=self.timeout)
-
- @mock.patch('six.moves.urllib.request.urlopen')
- def test_check_internet_connectivity_default(self, mock_method):
- self.assertTrue(functest_utils.check_internet_connectivity())
- mock_method.assert_called_once_with(self.url, timeout=self.timeout)
-
- @mock.patch('six.moves.urllib.request.urlopen')
- def test_check_internet_connectivity_debian(self, mock_method):
- self.url = "https://www.debian.org/"
- self.assertTrue(functest_utils.check_internet_connectivity(self.url))
- mock_method.assert_called_once_with(self.url, timeout=self.timeout)
-
- @mock.patch('six.moves.urllib.request.urlopen',
- side_effect=urllib.error.URLError('no host given'))
- def test_download_url_failed(self, mock_url):
- self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
-
- @mock.patch('six.moves.urllib.request.urlopen')
- def test_download_url_default(self, mock_url):
- with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
- mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
- as mock_sh:
- name = self.url.rsplit('/')[-1]
- dest = self.dest_path + "/" + name
- self.assertTrue(functest_utils.download_url(self.url,
- self.dest_path))
- m.assert_called_once_with(dest, 'wb')
- self.assertTrue(mock_sh.called)
-
def _get_env_dict(self, var):
dic = {'INSTALLER_TYPE': self.installer,
'DEPLOY_SCENARIO': self.scenario,
FunctestUtilsTesting.readline += 1
return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
- # TODO: get_resolvconf_ns
- @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
- def test_get_resolvconf_ns_default(self, mock_dns_resolve):
- attrs = {'query.return_value': ["test"]}
- mock_dns_resolve.configure_mock(**attrs)
-
- m = mock.Mock()
- attrs = {'readline.side_effect': self.readline_side}
- m.configure_mock(**attrs)
-
- with mock.patch("six.moves.builtins.open") as mo:
- mo.return_value = m
- self.assertEqual(functest_utils.get_resolvconf_ns(),
- self.test_ip[1:])
-
def _get_environ(self, var, *args): # pylint: disable=unused-argument
if var == 'INSTALLER_TYPE':
return self.installer
return self.scenario
return var
- def test_get_ci_envvars_default(self):
- with mock.patch('os.environ.get',
- side_effect=self._get_environ):
- dic = {"installer": self.installer,
- "scenario": self.scenario}
- self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
-
- def cmd_readline(self):
+ @staticmethod
+ def cmd_readline():
return 'test_value\n'
@mock.patch('functest.utils.functest_utils.LOGGER.error')
output_file=None)
self.assertEqual(resp, 1)
- def _get_functest_config(self, var):
- return var
-
def test_get_parameter_from_yaml_failed(self):
self.file_yaml['general'] = None
with mock.patch('six.moves.builtins.open', mock.mock_open()), \
self.test_file),
'test_image_name')
- @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
- def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
- with mock.patch.dict(os.environ,
- {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
- functest_utils.get_functest_config(self.parameter)
- mock_get_parameter_from_yaml. \
- assert_called_once_with(self.parameter,
- self.config_yaml)
-
- def test_get_functest_yaml(self):
- with mock.patch('six.moves.builtins.open', mock.mock_open()), \
- mock.patch('functest.utils.functest_utils.yaml.safe_load') \
- as mock_yaml:
- mock_yaml.return_value = self.file_yaml
- resp = functest_utils.get_functest_yaml()
- self.assertEqual(resp, self.file_yaml)
-
- @mock.patch('functest.utils.functest_utils.LOGGER.info')
- def test_print_separator(self, mock_logger_info):
- functest_utils.print_separator()
- mock_logger_info.assert_called_once_with("======================="
- "=======================")
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
from __future__ import print_function
import logging
-import re
-import shutil
import subprocess
import sys
-
-import dns.resolver
-from six.moves import urllib
import yaml
-from functest.utils import constants
-from functest.utils import env
-
LOGGER = logging.getLogger(__name__)
-# ----------------------------------------------------------
-#
-# INTERNET UTILS
-#
-# -----------------------------------------------------------
-def check_internet_connectivity(url='http://www.opnfv.org/'):
- """
- Check if there is access to the internet
- """
- try:
- urllib.request.urlopen(url, timeout=5)
- return True
- except urllib.error.URLError:
- return False
-
-
-def download_url(url, dest_path):
- """
- Download a file to a destination path given a URL
- """
- name = url.rsplit('/')[-1]
- dest = dest_path + "/" + name
- try:
- response = urllib.request.urlopen(url)
- except (urllib.error.HTTPError, urllib.error.URLError):
- return False
-
- with open(dest, 'wb') as lfile:
- shutil.copyfileobj(response, lfile)
- return True
-
-
-# ----------------------------------------------------------
-#
-# CI UTILS
-#
-# -----------------------------------------------------------
-def get_resolvconf_ns():
- """
- Get nameservers from current resolv.conf
- """
- nameservers = []
- rconf = open("/etc/resolv.conf", "r")
- line = rconf.readline()
- resolver = dns.resolver.Resolver()
- while line:
- addr_ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line)
- if addr_ip:
- resolver.nameservers = [addr_ip.group(0)]
- try:
- result = resolver.query('opnfv.org')[0]
- if result != "":
- nameservers.append(addr_ip.group())
- except dns.exception.Timeout:
- pass
- line = rconf.readline()
- return nameservers
-
-
-def get_ci_envvars():
- """
- Get the CI env variables
- """
- ci_env_var = {
- "installer": env.get('INSTALLER_TYPE'),
- "scenario": env.get('DEPLOY_SCENARIO')}
- return ci_env_var
-
-
def execute_command_raise(cmd, info=False, error_msg="",
verbose=True, output_file=None):
ret = execute_command(cmd, info, error_msg, verbose, output_file)
return returncode
-# ----------------------------------------------------------
-#
-# YAML UTILS
-#
-# -----------------------------------------------------------
def get_parameter_from_yaml(parameter, yfile):
"""
Returns the value of a given parameter in file.yaml
raise ValueError("The parameter %s is not defined in"
" %s" % (parameter, yfile))
return value
-
-
-def get_functest_config(parameter):
- yaml_ = constants.CONFIG_FUNCTEST_YAML
- return get_parameter_from_yaml(parameter, yaml_)
-
-
-def get_functest_yaml():
- # pylint: disable=bad-continuation
- with open(constants.CONFIG_FUNCTEST_YAML) as yaml_fd:
- functest_yaml = yaml.safe_load(yaml_fd)
- return functest_yaml
-
-
-def print_separator():
- LOGGER.info("==============================================")
Code Review / functest.git / commitdiff