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     @mock.patch('functest.utils.functest_utils.logger.error')
 101     def test_get_installer_type_failed(self, mock_logger_error):
 102         with mock.patch.dict(os.environ,
 103                              {},
 104                              clear=True):
 105             self.assertEqual(functest_utils.get_installer_type(),
 106                              "Unknown_installer")
 107             mock_logger_error.assert_called_once_with("Impossible to retrieve"
 108                                                       " the installer type")
 109
 110     def test_get_installer_type_default(self):
 111         with mock.patch.dict(os.environ,
 112                              {'INSTALLER_TYPE': 'test_installer'},
 113                              clear=True):
 114             self.assertEqual(functest_utils.get_installer_type(),
 115                              self.installer)
 116
 117     @mock.patch('functest.utils.functest_utils.logger.info')
 118     def test_get_scenario_failed(self, mock_logger_info):
 119         with mock.patch.dict(os.environ,
 120                              {},
 121                              clear=True):
 122             self.assertEqual(functest_utils.get_scenario(),
 123                              "os-nosdn-nofeature-noha")
 124             mock_logger_info.assert_called_once_with("Impossible to retrieve "
 125                                                      "the scenario.Use "
 126                                                      "default "
 127                                                      "os-nosdn-nofeature-noha")
 128
 129     def test_get_scenario_default(self):
 130         with mock.patch.dict(os.environ,
 131                              {'DEPLOY_SCENARIO': 'test_scenario'},
 132                              clear=True):
 133             self.assertEqual(functest_utils.get_scenario(),
 134                              self.scenario)
 135
 136     @mock.patch('functest.utils.functest_utils.get_build_tag')
 137     def test_get_version_daily_job(self, mock_get_build_tag):
 138         mock_get_build_tag.return_value = self.build_tag
 139         self.assertEqual(functest_utils.get_version(), self.version)
 140
 141     @mock.patch('functest.utils.functest_utils.get_build_tag')
 142     def test_get_version_weekly_job(self, mock_get_build_tag):
 143         mock_get_build_tag.return_value = self.build_tag_week
 144         self.assertEqual(functest_utils.get_version(), self.version)
 145
 146     @mock.patch('functest.utils.functest_utils.get_build_tag')
 147     def test_get_version_with_dummy_build_tag(self, mock_get_build_tag):
 148         mock_get_build_tag.return_value = 'whatever'
 149         self.assertEqual(functest_utils.get_version(), 'unknown')
 150
 151     @mock.patch('functest.utils.functest_utils.get_build_tag')
 152     def test_get_version_unknown(self, mock_get_build_tag):
 153         mock_get_build_tag.return_value = "unknown_build_tag"
 154         self.assertEqual(functest_utils.get_version(), "unknown")
 155
 156     @mock.patch('functest.utils.functest_utils.logger.info')
 157     def test_get_pod_name_failed(self, mock_logger_info):
 158         with mock.patch.dict(os.environ,
 159                              {},
 160                              clear=True):
 161             self.assertEqual(functest_utils.get_pod_name(),
 162                              "unknown-pod")
 163             mock_logger_info.assert_called_once_with("Unable to retrieve "
 164                                                      "the POD name from "
 165                                                      "environment. Using "
 166                                                      "pod name 'unknown-pod'")
 167
 168     def test_get_pod_name_default(self):
 169         with mock.patch.dict(os.environ,
 170                              {'NODE_NAME': 'test_node_name'},
 171                              clear=True):
 172             self.assertEqual(functest_utils.get_pod_name(),
 173                              self.node_name)
 174
 175     @mock.patch('functest.utils.functest_utils.logger.info')
 176     def test_get_build_tag_failed(self, mock_logger_info):
 177         with mock.patch.dict(os.environ,
 178                              {},
 179                              clear=True):
 180             self.assertEqual(functest_utils.get_build_tag(),
 181                              "none")
 182             mock_logger_info.assert_called_once_with("Impossible to retrieve"
 183                                                      " the build tag")
 184
 185     def test_get_build_tag_default(self):
 186         with mock.patch.dict(os.environ,
 187                              {'BUILD_TAG': self.build_tag},
 188                              clear=True):
 189             self.assertEqual(functest_utils.get_build_tag(),
 190                              self.build_tag)
 191
 192     @mock.patch('functest.utils.functest_utils.logger.info')
 193     def test_logger_test_results(self, mock_logger_info):
 194         CONST.__setattr__('results_test_db_url', self.db_url)
 195         with mock.patch('functest.utils.functest_utils.get_pod_name',
 196                         return_value=self.node_name), \
 197                 mock.patch('functest.utils.functest_utils.get_scenario',
 198                            return_value=self.scenario), \
 199                 mock.patch('functest.utils.functest_utils.get_version',
 200                            return_value=self.version), \
 201                 mock.patch('functest.utils.functest_utils.get_build_tag',
 202                            return_value=self.build_tag):
 203             functest_utils.logger_test_results(self.project, self.case_name,
 204                                                self.status, self.details)
 205             mock_logger_info.assert_called_once_with(
 206                 "\n"
 207                 "****************************************\n"
 208                 "\t %(p)s/%(n)s results \n\n"
 209                 "****************************************\n"
 210                 "DB:\t%(db)s\n"
 211                 "pod:\t%(pod)s\n"
 212                 "version:\t%(v)s\n"
 213                 "scenario:\t%(s)s\n"
 214                 "status:\t%(c)s\n"
 215                 "build tag:\t%(b)s\n"
 216                 "details:\t%(d)s\n"
 217                 % {'p': self.project,
 218                     'n': self.case_name,
 219                     'db': CONST.__getattribute__('results_test_db_url'),
 220                     'pod': self.node_name,
 221                     'v': self.version,
 222                     's': self.scenario,
 223                     'c': self.status,
 224                     'b': self.build_tag,
 225                     'd': self.details})
 226
 227     def _get_env_dict(self, var):
 228         dic = {'INSTALLER_TYPE': self.installer,
 229                'DEPLOY_SCENARIO': self.scenario,
 230                'NODE_NAME': self.node_name,
 231                'BUILD_TAG': self.build_tag}
 232         dic.pop(var, None)
 233         return dic
 234
 235     def _test_push_results_to_db_missing_env(self, env_var):
 236         dic = self._get_env_dict(env_var)
 237         CONST.__setattr__('results_test_db_url', self.db_url)
 238         with mock.patch.dict(os.environ,
 239                              dic,
 240                              clear=True), \
 241                 mock.patch('functest.utils.functest_utils.logger.error') \
 242                 as mock_logger_error:
 243             functest_utils.push_results_to_db(self.project, self.case_name,
 244                                               self.start_date, self.stop_date,
 245                                               self.result, self.details)
 246             mock_logger_error.assert_called_once_with("Please set env var: " +
 247                                                       str("\'" + env_var +
 248                                                           "\'"))
 249
 250     def test_push_results_to_db_missing_installer(self):
 251         self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
 252
 253     def test_push_results_to_db_missing_scenario(self):
 254         self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
 255
 256     def test_push_results_to_db_missing_nodename(self):
 257         self._test_push_results_to_db_missing_env('NODE_NAME')
 258
 259     def test_push_results_to_db_missing_buildtag(self):
 260         self._test_push_results_to_db_missing_env('BUILD_TAG')
 261
 262     def test_push_results_to_db_request_post_failed(self):
 263         dic = self._get_env_dict(None)
 264         CONST.__setattr__('results_test_db_url', self.db_url)
 265         with mock.patch.dict(os.environ,
 266                              dic,
 267                              clear=True), \
 268                 mock.patch('functest.utils.functest_utils.logger.error') \
 269                 as mock_logger_error, \
 270                 mock.patch('functest.utils.functest_utils.requests.post',
 271                            side_effect=requests.RequestException):
 272             self.assertFalse(functest_utils.
 273                              push_results_to_db(self.project, self.case_name,
 274                                                 self.start_date,
 275                                                 self.stop_date,
 276                                                 self.result, self.details))
 277             mock_logger_error.assert_called_once_with(test_utils.
 278                                                       RegexMatch("Pushing "
 279                                                                  "Result to"
 280                                                                  " DB"
 281                                                                  "(\S+\s*) "
 282                                                                  "failed:"))
 283
 284     def test_push_results_to_db_request_post_exception(self):
 285         dic = self._get_env_dict(None)
 286         CONST.__setattr__('results_test_db_url', self.db_url)
 287         with mock.patch.dict(os.environ,
 288                              dic,
 289                              clear=True), \
 290                 mock.patch('functest.utils.functest_utils.logger.error') \
 291                 as mock_logger_error, \
 292                 mock.patch('functest.utils.functest_utils.requests.post',
 293                            side_effect=Exception):
 294             self.assertFalse(functest_utils.
 295                              push_results_to_db(self.project, self.case_name,
 296                                                 self.start_date,
 297                                                 self.stop_date,
 298                                                 self.result, self.details))
 299             self.assertTrue(mock_logger_error.called)
 300
 301     def test_push_results_to_db_default(self):
 302         dic = self._get_env_dict(None)
 303         CONST.__setattr__('results_test_db_url', self.db_url)
 304         with mock.patch.dict(os.environ,
 305                              dic,
 306                              clear=True), \
 307                 mock.patch('functest.utils.functest_utils.requests.post'):
 308             self.assertTrue(functest_utils.
 309                             push_results_to_db(self.project, self.case_name,
 310                                                self.start_date,
 311                                                self.stop_date,
 312                                                self.result, self.details))
 313     readline = 0
 314     test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
 315
 316     @staticmethod
 317     def readline_side():
 318         if FunctestUtilsTesting.readline == \
 319                 len(FunctestUtilsTesting.test_ip) - 1:
 320             return False
 321         FunctestUtilsTesting.readline += 1
 322         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
 323
 324     # TODO: get_resolvconf_ns
 325     @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
 326     def test_get_resolvconf_ns_default(self, mock_dns_resolve):
 327         attrs = {'query.return_value': ["test"]}
 328         mock_dns_resolve.configure_mock(**attrs)
 329
 330         m = mock.Mock()
 331         attrs = {'readline.side_effect': self.readline_side}
 332         m.configure_mock(**attrs)
 333
 334         with mock.patch("six.moves.builtins.open") as mo:
 335             mo.return_value = m
 336             self.assertEqual(functest_utils.get_resolvconf_ns(),
 337                              self.test_ip[1:])
 338
 339     def _get_environ(self, var):
 340         if var == 'INSTALLER_TYPE':
 341             return self.installer
 342         elif var == 'DEPLOY_SCENARIO':
 343             return self.scenario
 344         return var
 345
 346     def test_get_ci_envvars_default(self):
 347         with mock.patch('os.environ.get',
 348                         side_effect=self._get_environ):
 349             dic = {"installer": self.installer,
 350                    "scenario": self.scenario}
 351             self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
 352
 353     def cmd_readline(self):
 354         return 'test_value\n'
 355
 356     @mock.patch('functest.utils.functest_utils.logger.error')
 357     @mock.patch('functest.utils.functest_utils.logger.info')
 358     def test_execute_command_args_present_with_error(self, mock_logger_info,
 359                                                      mock_logger_error):
 360         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 361                 as mock_subproc_open, \
 362                 mock.patch('six.moves.builtins.open',
 363                            mock.mock_open()) as mopen:
 364
 365             FunctestUtilsTesting.readline = 0
 366
 367             mock_obj = mock.Mock()
 368             attrs = {'readline.side_effect': self.cmd_readline()}
 369             mock_obj.configure_mock(**attrs)
 370
 371             mock_obj2 = mock.Mock()
 372             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 373             mock_obj2.configure_mock(**attrs)
 374
 375             mock_subproc_open.return_value = mock_obj2
 376
 377             resp = functest_utils.execute_command(self.cmd, info=True,
 378                                                   error_msg=self.error_msg,
 379                                                   verbose=True,
 380                                                   output_file=self.output_file)
 381             self.assertEqual(resp, 1)
 382             msg_exec = ("Executing command: '%s'" % self.cmd)
 383             mock_logger_info.assert_called_once_with(msg_exec)
 384             mopen.assert_called_once_with(self.output_file, "w")
 385             mock_logger_error.assert_called_once_with(self.error_msg)
 386
 387     @mock.patch('functest.utils.functest_utils.logger.info')
 388     def test_execute_command_args_present_with_success(self, mock_logger_info,
 389                                                        ):
 390         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 391                 as mock_subproc_open, \
 392                 mock.patch('six.moves.builtins.open',
 393                            mock.mock_open()) as mopen:
 394
 395             FunctestUtilsTesting.readline = 0
 396
 397             mock_obj = mock.Mock()
 398             attrs = {'readline.side_effect': self.cmd_readline()}
 399             mock_obj.configure_mock(**attrs)
 400
 401             mock_obj2 = mock.Mock()
 402             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 403             mock_obj2.configure_mock(**attrs)
 404
 405             mock_subproc_open.return_value = mock_obj2
 406
 407             resp = functest_utils.execute_command(self.cmd, info=True,
 408                                                   error_msg=self.error_msg,
 409                                                   verbose=True,
 410                                                   output_file=self.output_file)
 411             self.assertEqual(resp, 0)
 412             msg_exec = ("Executing command: '%s'" % self.cmd)
 413             mock_logger_info.assert_called_once_with(msg_exec)
 414             mopen.assert_called_once_with(self.output_file, "w")
 415
 416     @mock.patch('sys.stdout')
 417     def test_execute_command_args_missing_with_success(self, stdout=None):
 418         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 419                 as mock_subproc_open:
 420
 421             FunctestUtilsTesting.readline = 2
 422
 423             mock_obj = mock.Mock()
 424             attrs = {'readline.side_effect': self.cmd_readline()}
 425             mock_obj.configure_mock(**attrs)
 426
 427             mock_obj2 = mock.Mock()
 428             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 429             mock_obj2.configure_mock(**attrs)
 430
 431             mock_subproc_open.return_value = mock_obj2
 432
 433             resp = functest_utils.execute_command(self.cmd, info=False,
 434                                                   error_msg="",
 435                                                   verbose=False,
 436                                                   output_file=None)
 437             self.assertEqual(resp, 0)
 438
 439     @mock.patch('sys.stdout')
 440     def test_execute_command_args_missing_with_error(self, stdout=None):
 441         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 442                 as mock_subproc_open:
 443
 444             FunctestUtilsTesting.readline = 2
 445             mock_obj = mock.Mock()
 446             attrs = {'readline.side_effect': self.cmd_readline()}
 447             mock_obj.configure_mock(**attrs)
 448
 449             mock_obj2 = mock.Mock()
 450             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 451             mock_obj2.configure_mock(**attrs)
 452
 453             mock_subproc_open.return_value = mock_obj2
 454
 455             resp = functest_utils.execute_command(self.cmd, info=False,
 456                                                   error_msg="",
 457                                                   verbose=False,
 458                                                   output_file=None)
 459             self.assertEqual(resp, 1)
 460
 461     def _get_functest_config(self, var):
 462         return var
 463
 464     @mock.patch('functest.utils.functest_utils.logger.error')
 465     def test_get_dict_by_test(self, mock_logger_error):
 466         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 467                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 468                 as mock_yaml:
 469             mock_obj = mock.Mock()
 470             attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
 471             mock_obj.configure_mock(**attrs)
 472
 473             mock_yaml.return_value = mock_obj
 474
 475             self.assertDictEqual(functest_utils.
 476                                  get_dict_by_test(self.testname),
 477                                  self.testcase_dict)
 478
 479     @mock.patch('functest.utils.functest_utils.get_dict_by_test')
 480     def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
 481         mock_get_dict_by_test.return_value = self.testcase_dict
 482         self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
 483                          self.criteria)
 484
 485     @mock.patch('functest.utils.functest_utils.get_dict_by_test')
 486     def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
 487         mock_get_dict_by_test.return_value = None
 488         self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
 489
 490     def test_get_parameter_from_yaml_failed(self):
 491         self.file_yaml['general'] = None
 492         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 493                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 494                 as mock_yaml, \
 495                 self.assertRaises(ValueError) as excep:
 496             mock_yaml.return_value = self.file_yaml
 497             functest_utils.get_parameter_from_yaml(self.parameter,
 498                                                    self.test_file)
 499             self.assertTrue(("The parameter %s is not"
 500                              " defined in config_functest.yaml" %
 501                              self.parameter) in excep.exception)
 502
 503     def test_get_parameter_from_yaml_default(self):
 504         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 505                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 506                 as mock_yaml:
 507             mock_yaml.return_value = self.file_yaml
 508             self.assertEqual(functest_utils.
 509                              get_parameter_from_yaml(self.parameter,
 510                                                      self.test_file),
 511                              'test_image_name')
 512
 513     @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
 514     def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
 515         with mock.patch.dict(os.environ,
 516                              {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
 517             functest_utils.get_functest_config(self.parameter)
 518             mock_get_parameter_from_yaml. \
 519                 assert_called_once_with(self.parameter,
 520                                         self.config_yaml)
 521
 522     def test_get_functest_yaml(self):
 523         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 524                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 525                 as mock_yaml:
 526             mock_yaml.return_value = self.file_yaml
 527             resp = functest_utils.get_functest_yaml()
 528             self.assertEqual(resp, self.file_yaml)
 529
 530     @mock.patch('functest.utils.functest_utils.logger.info')
 531     def test_print_separator(self, mock_logger_info):
 532         functest_utils.print_separator()
 533         mock_logger_info.assert_called_once_with("======================="
 534                                                  "=======================")
 535
 536
 537 if __name__ == "__main__":
 538     logging.disable(logging.CRITICAL)
 539     unittest.main(verbosity=2)