functest/tests/unit/utils/test_functest_utils.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 Orange and others.
   4 #
   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 import logging
  13 import os
  14 import time
  15 import unittest
  16
  17 import mock
  18 import pkg_resources
  19 from six.moves import urllib
  20
  21 from functest.utils import functest_utils
  22
  23
  24 class FunctestUtilsTesting(unittest.TestCase):
  25
  26     readline = 0
  27     test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
  28
  29     def setUp(self):
  30         self.url = 'http://www.opnfv.org/'
  31         self.timeout = 5
  32         self.dest_path = 'test_path'
  33         self.repo_path = 'test_repo_path'
  34         self.installer = 'test_installer'
  35         self.scenario = 'test_scenario'
  36         self.build_tag = 'jenkins-functest-fuel-opnfv-jump-2-daily-master-190'
  37         self.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
  38         self.version = 'master'
  39         self.node_name = 'test_node_name'
  40         self.project = 'test_project'
  41         self.case_name = 'test_case_name'
  42         self.status = 'test_status'
  43         self.details = 'test_details'
  44         self.db_url = 'test_db_url'
  45         self.criteria = 50
  46         self.result = 75
  47         self.start_date = 1482624000
  48         self.stop_date = 1482624000
  49         self.start_time = time.time()
  50         self.stop_time = time.time()
  51         self.readline = -1
  52         self.test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
  53         self.test_file = 'test_file'
  54         self.error_msg = 'test_error_msg'
  55         self.cmd = 'test_cmd'
  56         self.output_file = 'test_output_file'
  57         self.testname = 'testname'
  58         self.testcase_dict = {'case_name': 'testname',
  59                               'criteria': self.criteria}
  60         self.parameter = 'general.openstack.image_name'
  61         self.config_yaml = pkg_resources.resource_filename(
  62             'functest', 'ci/config_functest.yaml')
  63         self.db_url_env = 'http://foo/testdb'
  64         self.testcases_yaml = "test_testcases_yaml"
  65         self.file_yaml = {'general': {'openstack': {'image_name':
  66                                                     'test_image_name'}}}
  67
  68     @mock.patch('six.moves.urllib.request.urlopen',
  69                 side_effect=urllib.error.URLError('no host given'))
  70     def test_check_internet_connectivity_failed(self, mock_method):
  71         self.assertFalse(functest_utils.check_internet_connectivity())
  72         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
  73
  74     @mock.patch('six.moves.urllib.request.urlopen')
  75     def test_check_internet_connectivity_default(self, mock_method):
  76         self.assertTrue(functest_utils.check_internet_connectivity())
  77         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
  78
  79     @mock.patch('six.moves.urllib.request.urlopen')
  80     def test_check_internet_connectivity_debian(self, mock_method):
  81         self.url = "https://www.debian.org/"
  82         self.assertTrue(functest_utils.check_internet_connectivity(self.url))
  83         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
  84
  85     @mock.patch('six.moves.urllib.request.urlopen',
  86                 side_effect=urllib.error.URLError('no host given'))
  87     def test_download_url_failed(self, mock_url):
  88         self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
  89
  90     @mock.patch('six.moves.urllib.request.urlopen')
  91     def test_download_url_default(self, mock_url):
  92         with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
  93                 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
  94                 as mock_sh:
  95             name = self.url.rsplit('/')[-1]
  96             dest = self.dest_path + "/" + name
  97             self.assertTrue(functest_utils.download_url(self.url,
  98                                                         self.dest_path))
  99             m.assert_called_once_with(dest, 'wb')
 100             self.assertTrue(mock_sh.called)
 101
 102     def _get_env_dict(self, var):
 103         dic = {'INSTALLER_TYPE': self.installer,
 104                'DEPLOY_SCENARIO': self.scenario,
 105                'NODE_NAME': self.node_name,
 106                'BUILD_TAG': self.build_tag}
 107         dic.pop(var, None)
 108         return dic
 109
 110     @staticmethod
 111     def readline_side():
 112         if FunctestUtilsTesting.readline == \
 113                 len(FunctestUtilsTesting.test_ip) - 1:
 114             return False
 115         FunctestUtilsTesting.readline += 1
 116         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
 117
 118     # TODO: get_resolvconf_ns
 119     @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
 120     def test_get_resolvconf_ns_default(self, mock_dns_resolve):
 121         attrs = {'query.return_value': ["test"]}
 122         mock_dns_resolve.configure_mock(**attrs)
 123
 124         m = mock.Mock()
 125         attrs = {'readline.side_effect': self.readline_side}
 126         m.configure_mock(**attrs)
 127
 128         with mock.patch("six.moves.builtins.open") as mo:
 129             mo.return_value = m
 130             self.assertEqual(functest_utils.get_resolvconf_ns(),
 131                              self.test_ip[1:])
 132
 133     def _get_environ(self, var):
 134         if var == 'INSTALLER_TYPE':
 135             return self.installer
 136         elif var == 'DEPLOY_SCENARIO':
 137             return self.scenario
 138         return var
 139
 140     def test_get_ci_envvars_default(self):
 141         with mock.patch('os.environ.get',
 142                         side_effect=self._get_environ):
 143             dic = {"installer": self.installer,
 144                    "scenario": self.scenario}
 145             self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
 146
 147     def cmd_readline(self):
 148         return 'test_value\n'
 149
 150     @mock.patch('functest.utils.functest_utils.LOGGER.error')
 151     @mock.patch('functest.utils.functest_utils.LOGGER.info')
 152     def test_execute_command_args_present_with_error(self, mock_logger_info,
 153                                                      mock_logger_error):
 154         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 155                 as mock_subproc_open, \
 156                 mock.patch('six.moves.builtins.open',
 157                            mock.mock_open()) as mopen:
 158
 159             FunctestUtilsTesting.readline = 0
 160
 161             mock_obj = mock.Mock()
 162             attrs = {'readline.side_effect': self.cmd_readline()}
 163             mock_obj.configure_mock(**attrs)
 164
 165             mock_obj2 = mock.Mock()
 166             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 167             mock_obj2.configure_mock(**attrs)
 168
 169             mock_subproc_open.return_value = mock_obj2
 170
 171             resp = functest_utils.execute_command(self.cmd, info=True,
 172                                                   error_msg=self.error_msg,
 173                                                   verbose=True,
 174                                                   output_file=self.output_file)
 175             self.assertEqual(resp, 1)
 176             msg_exec = ("Executing command: '%s'" % self.cmd)
 177             mock_logger_info.assert_called_once_with(msg_exec)
 178             mopen.assert_called_once_with(self.output_file, "w")
 179             mock_logger_error.assert_called_once_with(self.error_msg)
 180
 181     @mock.patch('functest.utils.functest_utils.LOGGER.info')
 182     def test_execute_command_args_present_with_success(self, mock_logger_info,
 183                                                        ):
 184         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 185                 as mock_subproc_open, \
 186                 mock.patch('six.moves.builtins.open',
 187                            mock.mock_open()) as mopen:
 188
 189             FunctestUtilsTesting.readline = 0
 190
 191             mock_obj = mock.Mock()
 192             attrs = {'readline.side_effect': self.cmd_readline()}
 193             mock_obj.configure_mock(**attrs)
 194
 195             mock_obj2 = mock.Mock()
 196             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 197             mock_obj2.configure_mock(**attrs)
 198
 199             mock_subproc_open.return_value = mock_obj2
 200
 201             resp = functest_utils.execute_command(self.cmd, info=True,
 202                                                   error_msg=self.error_msg,
 203                                                   verbose=True,
 204                                                   output_file=self.output_file)
 205             self.assertEqual(resp, 0)
 206             msg_exec = ("Executing command: '%s'" % self.cmd)
 207             mock_logger_info.assert_called_once_with(msg_exec)
 208             mopen.assert_called_once_with(self.output_file, "w")
 209
 210     @mock.patch('sys.stdout')
 211     def test_execute_command_args_missing_with_success(self, stdout=None):
 212         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 213                 as mock_subproc_open:
 214
 215             FunctestUtilsTesting.readline = 2
 216
 217             mock_obj = mock.Mock()
 218             attrs = {'readline.side_effect': self.cmd_readline()}
 219             mock_obj.configure_mock(**attrs)
 220
 221             mock_obj2 = mock.Mock()
 222             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 223             mock_obj2.configure_mock(**attrs)
 224
 225             mock_subproc_open.return_value = mock_obj2
 226
 227             resp = functest_utils.execute_command(self.cmd, info=False,
 228                                                   error_msg="",
 229                                                   verbose=False,
 230                                                   output_file=None)
 231             self.assertEqual(resp, 0)
 232
 233     @mock.patch('sys.stdout')
 234     def test_execute_command_args_missing_with_error(self, stdout=None):
 235         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 236                 as mock_subproc_open:
 237
 238             FunctestUtilsTesting.readline = 2
 239             mock_obj = mock.Mock()
 240             attrs = {'readline.side_effect': self.cmd_readline()}
 241             mock_obj.configure_mock(**attrs)
 242
 243             mock_obj2 = mock.Mock()
 244             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 245             mock_obj2.configure_mock(**attrs)
 246
 247             mock_subproc_open.return_value = mock_obj2
 248
 249             resp = functest_utils.execute_command(self.cmd, info=False,
 250                                                   error_msg="",
 251                                                   verbose=False,
 252                                                   output_file=None)
 253             self.assertEqual(resp, 1)
 254
 255     def _get_functest_config(self, var):
 256         return var
 257
 258     @mock.patch('functest.utils.functest_utils.LOGGER.error')
 259     def test_get_dict_by_test(self, mock_logger_error):
 260         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 261                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 262                 as mock_yaml:
 263             mock_obj = mock.Mock()
 264             attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
 265             mock_obj.configure_mock(**attrs)
 266
 267             mock_yaml.return_value = mock_obj
 268
 269             self.assertDictEqual(functest_utils.
 270                                  get_dict_by_test(self.testname),
 271                                  self.testcase_dict)
 272
 273     @mock.patch('functest.utils.functest_utils.get_dict_by_test')
 274     def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
 275         mock_get_dict_by_test.return_value = self.testcase_dict
 276         self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
 277                          self.criteria)
 278
 279     @mock.patch('functest.utils.functest_utils.get_dict_by_test')
 280     def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
 281         mock_get_dict_by_test.return_value = None
 282         self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
 283
 284     def test_get_parameter_from_yaml_failed(self):
 285         self.file_yaml['general'] = None
 286         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 287                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 288                 as mock_yaml, \
 289                 self.assertRaises(ValueError) as excep:
 290             mock_yaml.return_value = self.file_yaml
 291             functest_utils.get_parameter_from_yaml(self.parameter,
 292                                                    self.test_file)
 293             self.assertTrue(("The parameter %s is not"
 294                              " defined in config_functest.yaml" %
 295                              self.parameter) in excep.exception)
 296
 297     def test_get_parameter_from_yaml_default(self):
 298         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 299                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 300                 as mock_yaml:
 301             mock_yaml.return_value = self.file_yaml
 302             self.assertEqual(functest_utils.
 303                              get_parameter_from_yaml(self.parameter,
 304                                                      self.test_file),
 305                              'test_image_name')
 306
 307     @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
 308     def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
 309         with mock.patch.dict(os.environ,
 310                              {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
 311             functest_utils.get_functest_config(self.parameter)
 312             mock_get_parameter_from_yaml. \
 313                 assert_called_once_with(self.parameter,
 314                                         self.config_yaml)
 315
 316     def test_get_functest_yaml(self):
 317         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 318                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 319                 as mock_yaml:
 320             mock_yaml.return_value = self.file_yaml
 321             resp = functest_utils.get_functest_yaml()
 322             self.assertEqual(resp, self.file_yaml)
 323
 324     @mock.patch('functest.utils.functest_utils.LOGGER.info')
 325     def test_print_separator(self, mock_logger_info):
 326         functest_utils.print_separator()
 327         mock_logger_info.assert_called_once_with("======================="
 328                                                  "=======================")
 329
 330
 331 if __name__ == "__main__":
 332     logging.disable(logging.CRITICAL)
 333     unittest.main(verbosity=2)