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 import logging
  11 import pkg_resources
  12 import os
  13 import time
  14 import unittest
  15
  16 import mock
  17 import requests
  18 from six.moves import urllib
  19
  20 from functest.tests.unit import test_utils
  21 from functest.utils import functest_utils
  22 from functest.utils.constants import CONST
  23
  24
  25 class FunctestUtilsTesting(unittest.TestCase):
  26
  27     def setUp(self):
  28         self.url = 'http://www.opnfv.org/'
  29         self.timeout = 5
  30         self.dest_path = 'test_path'
  31         self.repo_path = 'test_repo_path'
  32         self.installer = 'test_installer'
  33         self.scenario = 'test_scenario'
  34         self.build_tag = 'jenkins-functest-fuel-opnfv-jump-2-daily-master-190'
  35         self.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
  36         self.version = 'master'
  37         self.node_name = 'test_node_name'
  38         self.project = 'test_project'
  39         self.case_name = 'test_case_name'
  40         self.status = 'test_status'
  41         self.details = 'test_details'
  42         self.db_url = 'test_db_url'
  43         self.criteria = 50
  44         self.result = 75
  45         self.start_date = 1482624000
  46         self.stop_date = 1482624000
  47         self.start_time = time.time()
  48         self.stop_time = time.time()
  49         self.readline = -1
  50         self.test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
  51         self.test_file = 'test_file'
  52         self.error_msg = 'test_error_msg'
  53         self.cmd = 'test_cmd'
  54         self.output_file = 'test_output_file'
  55         self.testname = 'testname'
  56         self.testcase_dict = {'case_name': 'testname',
  57                               'criteria': self.criteria}
  58         self.parameter = 'general.openstack.image_name'
  59         self.config_yaml = pkg_resources.resource_filename(
  60             'functest', 'ci/config_functest.yaml')
  61         self.db_url_env = 'http://foo/testdb'
  62         self.testcases_yaml = "test_testcases_yaml"
  63         self.file_yaml = {'general': {'openstack': {'image_name':
  64                                                     'test_image_name'}}}
  65
  66     @mock.patch('six.moves.urllib.request.urlopen',
  67                 side_effect=urllib.error.URLError('no host given'))
  68     def test_check_internet_connectivity_failed(self, mock_method):
  69         self.assertFalse(functest_utils.check_internet_connectivity())
  70         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
  71
  72     @mock.patch('six.moves.urllib.request.urlopen')
  73     def test_check_internet_connectivity_default(self, mock_method):
  74         self.assertTrue(functest_utils.check_internet_connectivity())
  75         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
  76
  77     @mock.patch('six.moves.urllib.request.urlopen')
  78     def test_check_internet_connectivity_debian(self, mock_method):
  79         self.url = "https://www.debian.org/"
  80         self.assertTrue(functest_utils.check_internet_connectivity(self.url))
  81         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
  82
  83     @mock.patch('six.moves.urllib.request.urlopen',
  84                 side_effect=urllib.error.URLError('no host given'))
  85     def test_download_url_failed(self, mock_url):
  86         self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
  87
  88     @mock.patch('six.moves.urllib.request.urlopen')
  89     def test_download_url_default(self, mock_url):
  90         with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
  91                 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
  92                 as mock_sh:
  93             name = self.url.rsplit('/')[-1]
  94             dest = self.dest_path + "/" + name
  95             self.assertTrue(functest_utils.download_url(self.url,
  96                                                         self.dest_path))
  97             m.assert_called_once_with(dest, 'wb')
  98             self.assertTrue(mock_sh.called)
  99
 100     def test_get_version_daily_job(self):
 101         CONST.__setattr__('BUILD_TAG', self.build_tag)
 102         self.assertEqual(functest_utils.get_version(), self.version)
 103
 104     def test_get_version_weekly_job(self):
 105         CONST.__setattr__('BUILD_TAG', self.build_tag_week)
 106         self.assertEqual(functest_utils.get_version(), self.version)
 107
 108     def test_get_version_with_dummy_build_tag(self):
 109         CONST.__setattr__('BUILD_TAG', 'whatever')
 110         self.assertEqual(functest_utils.get_version(), 'unknown')
 111
 112     def test_get_version_unknown(self):
 113         CONST.__setattr__('BUILD_TAG', 'unknown_build_tag')
 114         self.assertEqual(functest_utils.get_version(), "unknown")
 115
 116     def _get_env_dict(self, var):
 117         dic = {'INSTALLER_TYPE': self.installer,
 118                'DEPLOY_SCENARIO': self.scenario,
 119                'NODE_NAME': self.node_name,
 120                'BUILD_TAG': self.build_tag}
 121         dic.pop(var, None)
 122         return dic
 123
 124     def _test_push_results_to_db_missing_env(self, env_var):
 125         dic = self._get_env_dict(env_var)
 126         CONST.__setattr__('results_test_db_url', self.db_url)
 127         with mock.patch.dict(os.environ,
 128                              dic,
 129                              clear=True), \
 130                 mock.patch('functest.utils.functest_utils.logger.error') \
 131                 as mock_logger_error:
 132             functest_utils.push_results_to_db(self.project, self.case_name,
 133                                               self.start_date, self.stop_date,
 134                                               self.result, self.details)
 135             mock_logger_error.assert_called_once_with("Please set env var: " +
 136                                                       str("\'" + env_var +
 137                                                           "\'"))
 138
 139     def test_push_results_to_db_missing_installer(self):
 140         self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
 141
 142     def test_push_results_to_db_missing_scenario(self):
 143         self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
 144
 145     def test_push_results_to_db_missing_nodename(self):
 146         self._test_push_results_to_db_missing_env('NODE_NAME')
 147
 148     def test_push_results_to_db_missing_buildtag(self):
 149         self._test_push_results_to_db_missing_env('BUILD_TAG')
 150
 151     def test_push_results_to_db_request_post_failed(self):
 152         dic = self._get_env_dict(None)
 153         CONST.__setattr__('results_test_db_url', self.db_url)
 154         with mock.patch.dict(os.environ,
 155                              dic,
 156                              clear=True), \
 157                 mock.patch('functest.utils.functest_utils.logger.error') \
 158                 as mock_logger_error, \
 159                 mock.patch('functest.utils.functest_utils.requests.post',
 160                            side_effect=requests.RequestException):
 161             self.assertFalse(functest_utils.
 162                              push_results_to_db(self.project, self.case_name,
 163                                                 self.start_date,
 164                                                 self.stop_date,
 165                                                 self.result, self.details))
 166             mock_logger_error.assert_called_once_with(test_utils.
 167                                                       RegexMatch("Pushing "
 168                                                                  "Result to"
 169                                                                  " DB"
 170                                                                  "(\S+\s*) "
 171                                                                  "failed:"))
 172
 173     def test_push_results_to_db_request_post_exception(self):
 174         dic = self._get_env_dict(None)
 175         CONST.__setattr__('results_test_db_url', self.db_url)
 176         with mock.patch.dict(os.environ,
 177                              dic,
 178                              clear=True), \
 179                 mock.patch('functest.utils.functest_utils.logger.error') \
 180                 as mock_logger_error, \
 181                 mock.patch('functest.utils.functest_utils.requests.post',
 182                            side_effect=Exception):
 183             self.assertFalse(functest_utils.
 184                              push_results_to_db(self.project, self.case_name,
 185                                                 self.start_date,
 186                                                 self.stop_date,
 187                                                 self.result, self.details))
 188             self.assertTrue(mock_logger_error.called)
 189
 190     def test_push_results_to_db_default(self):
 191         dic = self._get_env_dict(None)
 192         CONST.__setattr__('results_test_db_url', self.db_url)
 193         with mock.patch.dict(os.environ,
 194                              dic,
 195                              clear=True), \
 196                 mock.patch('functest.utils.functest_utils.requests.post'):
 197             self.assertTrue(functest_utils.
 198                             push_results_to_db(self.project, self.case_name,
 199                                                self.start_date,
 200                                                self.stop_date,
 201                                                self.result, self.details))
 202     readline = 0
 203     test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
 204
 205     @staticmethod
 206     def readline_side():
 207         if FunctestUtilsTesting.readline == \
 208                 len(FunctestUtilsTesting.test_ip) - 1:
 209             return False
 210         FunctestUtilsTesting.readline += 1
 211         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
 212
 213     # TODO: get_resolvconf_ns
 214     @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
 215     def test_get_resolvconf_ns_default(self, mock_dns_resolve):
 216         attrs = {'query.return_value': ["test"]}
 217         mock_dns_resolve.configure_mock(**attrs)
 218
 219         m = mock.Mock()
 220         attrs = {'readline.side_effect': self.readline_side}
 221         m.configure_mock(**attrs)
 222
 223         with mock.patch("six.moves.builtins.open") as mo:
 224             mo.return_value = m
 225             self.assertEqual(functest_utils.get_resolvconf_ns(),
 226                              self.test_ip[1:])
 227
 228     def _get_environ(self, var):
 229         if var == 'INSTALLER_TYPE':
 230             return self.installer
 231         elif var == 'DEPLOY_SCENARIO':
 232             return self.scenario
 233         return var
 234
 235     def test_get_ci_envvars_default(self):
 236         with mock.patch('os.environ.get',
 237                         side_effect=self._get_environ):
 238             dic = {"installer": self.installer,
 239                    "scenario": self.scenario}
 240             self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
 241
 242     def cmd_readline(self):
 243         return 'test_value\n'
 244
 245     @mock.patch('functest.utils.functest_utils.logger.error')
 246     @mock.patch('functest.utils.functest_utils.logger.info')
 247     def test_execute_command_args_present_with_error(self, mock_logger_info,
 248                                                      mock_logger_error):
 249         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 250                 as mock_subproc_open, \
 251                 mock.patch('six.moves.builtins.open',
 252                            mock.mock_open()) as mopen:
 253
 254             FunctestUtilsTesting.readline = 0
 255
 256             mock_obj = mock.Mock()
 257             attrs = {'readline.side_effect': self.cmd_readline()}
 258             mock_obj.configure_mock(**attrs)
 259
 260             mock_obj2 = mock.Mock()
 261             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 262             mock_obj2.configure_mock(**attrs)
 263
 264             mock_subproc_open.return_value = mock_obj2
 265
 266             resp = functest_utils.execute_command(self.cmd, info=True,
 267                                                   error_msg=self.error_msg,
 268                                                   verbose=True,
 269                                                   output_file=self.output_file)
 270             self.assertEqual(resp, 1)
 271             msg_exec = ("Executing command: '%s'" % self.cmd)
 272             mock_logger_info.assert_called_once_with(msg_exec)
 273             mopen.assert_called_once_with(self.output_file, "w")
 274             mock_logger_error.assert_called_once_with(self.error_msg)
 275
 276     @mock.patch('functest.utils.functest_utils.logger.info')
 277     def test_execute_command_args_present_with_success(self, mock_logger_info,
 278                                                        ):
 279         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 280                 as mock_subproc_open, \
 281                 mock.patch('six.moves.builtins.open',
 282                            mock.mock_open()) as mopen:
 283
 284             FunctestUtilsTesting.readline = 0
 285
 286             mock_obj = mock.Mock()
 287             attrs = {'readline.side_effect': self.cmd_readline()}
 288             mock_obj.configure_mock(**attrs)
 289
 290             mock_obj2 = mock.Mock()
 291             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 292             mock_obj2.configure_mock(**attrs)
 293
 294             mock_subproc_open.return_value = mock_obj2
 295
 296             resp = functest_utils.execute_command(self.cmd, info=True,
 297                                                   error_msg=self.error_msg,
 298                                                   verbose=True,
 299                                                   output_file=self.output_file)
 300             self.assertEqual(resp, 0)
 301             msg_exec = ("Executing command: '%s'" % self.cmd)
 302             mock_logger_info.assert_called_once_with(msg_exec)
 303             mopen.assert_called_once_with(self.output_file, "w")
 304
 305     @mock.patch('sys.stdout')
 306     def test_execute_command_args_missing_with_success(self, stdout=None):
 307         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 308                 as mock_subproc_open:
 309
 310             FunctestUtilsTesting.readline = 2
 311
 312             mock_obj = mock.Mock()
 313             attrs = {'readline.side_effect': self.cmd_readline()}
 314             mock_obj.configure_mock(**attrs)
 315
 316             mock_obj2 = mock.Mock()
 317             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 318             mock_obj2.configure_mock(**attrs)
 319
 320             mock_subproc_open.return_value = mock_obj2
 321
 322             resp = functest_utils.execute_command(self.cmd, info=False,
 323                                                   error_msg="",
 324                                                   verbose=False,
 325                                                   output_file=None)
 326             self.assertEqual(resp, 0)
 327
 328     @mock.patch('sys.stdout')
 329     def test_execute_command_args_missing_with_error(self, stdout=None):
 330         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 331                 as mock_subproc_open:
 332
 333             FunctestUtilsTesting.readline = 2
 334             mock_obj = mock.Mock()
 335             attrs = {'readline.side_effect': self.cmd_readline()}
 336             mock_obj.configure_mock(**attrs)
 337
 338             mock_obj2 = mock.Mock()
 339             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 340             mock_obj2.configure_mock(**attrs)
 341
 342             mock_subproc_open.return_value = mock_obj2
 343
 344             resp = functest_utils.execute_command(self.cmd, info=False,
 345                                                   error_msg="",
 346                                                   verbose=False,
 347                                                   output_file=None)
 348             self.assertEqual(resp, 1)
 349
 350     def _get_functest_config(self, var):
 351         return var
 352
 353     @mock.patch('functest.utils.functest_utils.logger.error')
 354     def test_get_dict_by_test(self, mock_logger_error):
 355         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 356                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 357                 as mock_yaml:
 358             mock_obj = mock.Mock()
 359             attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
 360             mock_obj.configure_mock(**attrs)
 361
 362             mock_yaml.return_value = mock_obj
 363
 364             self.assertDictEqual(functest_utils.
 365                                  get_dict_by_test(self.testname),
 366                                  self.testcase_dict)
 367
 368     @mock.patch('functest.utils.functest_utils.get_dict_by_test')
 369     def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
 370         mock_get_dict_by_test.return_value = self.testcase_dict
 371         self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
 372                          self.criteria)
 373
 374     @mock.patch('functest.utils.functest_utils.get_dict_by_test')
 375     def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
 376         mock_get_dict_by_test.return_value = None
 377         self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
 378
 379     def test_get_parameter_from_yaml_failed(self):
 380         self.file_yaml['general'] = None
 381         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 382                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 383                 as mock_yaml, \
 384                 self.assertRaises(ValueError) as excep:
 385             mock_yaml.return_value = self.file_yaml
 386             functest_utils.get_parameter_from_yaml(self.parameter,
 387                                                    self.test_file)
 388             self.assertTrue(("The parameter %s is not"
 389                              " defined in config_functest.yaml" %
 390                              self.parameter) in excep.exception)
 391
 392     def test_get_parameter_from_yaml_default(self):
 393         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 394                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 395                 as mock_yaml:
 396             mock_yaml.return_value = self.file_yaml
 397             self.assertEqual(functest_utils.
 398                              get_parameter_from_yaml(self.parameter,
 399                                                      self.test_file),
 400                              'test_image_name')
 401
 402     @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
 403     def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
 404         with mock.patch.dict(os.environ,
 405                              {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
 406             functest_utils.get_functest_config(self.parameter)
 407             mock_get_parameter_from_yaml. \
 408                 assert_called_once_with(self.parameter,
 409                                         self.config_yaml)
 410
 411     def test_get_functest_yaml(self):
 412         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 413                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 414                 as mock_yaml:
 415             mock_yaml.return_value = self.file_yaml
 416             resp = functest_utils.get_functest_yaml()
 417             self.assertEqual(resp, self.file_yaml)
 418
 419     @mock.patch('functest.utils.functest_utils.logger.info')
 420     def test_print_separator(self, mock_logger_info):
 421         functest_utils.print_separator()
 422         mock_logger_info.assert_called_once_with("======================="
 423                                                  "=======================")
 424
 425
 426 if __name__ == "__main__":
 427     logging.disable(logging.CRITICAL)
 428     unittest.main(verbosity=2)