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 time
  14 import unittest
  15
  16 import mock
  17 import pkg_resources
  18
  19 from functest.utils import functest_utils
  20
  21
  22 class FunctestUtilsTesting(unittest.TestCase):
  23     # pylint: disable=too-many-instance-attributes,too-many-public-methods
  24
  25     readline = 0
  26     test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
  27
  28     def setUp(self):
  29         self.url = 'http://www.opnfv.org/'
  30         self.timeout = 5
  31         self.dest_path = 'test_path'
  32         self.repo_path = 'test_repo_path'
  33         self.installer = 'test_installer'
  34         self.scenario = 'test_scenario'
  35         self.build_tag = 'jenkins-functest-fuel-opnfv-jump-2-daily-master-190'
  36         self.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
  37         self.version = 'master'
  38         self.node_name = 'test_node_name'
  39         self.project = 'test_project'
  40         self.case_name = 'test_case_name'
  41         self.status = 'test_status'
  42         self.details = 'test_details'
  43         self.db_url = 'test_db_url'
  44         self.criteria = 50
  45         self.result = 75
  46         self.start_date = 1482624000
  47         self.stop_date = 1482624000
  48         self.start_time = time.time()
  49         self.stop_time = time.time()
  50         self.readline = -1
  51         self.test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
  52         self.test_file = 'test_file'
  53         self.error_msg = 'test_error_msg'
  54         self.cmd = 'test_cmd'
  55         self.output_file = 'test_output_file'
  56         self.testname = 'testname'
  57         self.parameter = 'general.openstack.image_name'
  58         self.config_yaml = pkg_resources.resource_filename(
  59             'functest', 'ci/config_functest.yaml')
  60         self.db_url_env = 'http://foo/testdb'
  61         self.testcases_yaml = "test_testcases_yaml"
  62         self.file_yaml = {'general': {'openstack': {'image_name':
  63                                                     'test_image_name'}}}
  64
  65     def _get_env_dict(self, var):
  66         dic = {'INSTALLER_TYPE': self.installer,
  67                'DEPLOY_SCENARIO': self.scenario,
  68                'NODE_NAME': self.node_name,
  69                'BUILD_TAG': self.build_tag}
  70         dic.pop(var, None)
  71         return dic
  72
  73     @staticmethod
  74     def readline_side():
  75         if FunctestUtilsTesting.readline == \
  76                 len(FunctestUtilsTesting.test_ip) - 1:
  77             return False
  78         FunctestUtilsTesting.readline += 1
  79         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
  80
  81     def _get_environ(self, var, *args):  # pylint: disable=unused-argument
  82         if var == 'INSTALLER_TYPE':
  83             return self.installer
  84         elif var == 'DEPLOY_SCENARIO':
  85             return self.scenario
  86         return var
  87
  88     @staticmethod
  89     def cmd_readline():
  90         return 'test_value\n'
  91
  92     @mock.patch('functest.utils.functest_utils.LOGGER.error')
  93     @mock.patch('functest.utils.functest_utils.LOGGER.info')
  94     def test_exec_cmd_args_present_ko(self, mock_logger_info,
  95                                       mock_logger_error):
  96         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
  97                 as mock_subproc_open, \
  98                 mock.patch('six.moves.builtins.open',
  99                            mock.mock_open()) as mopen:
 100
 101             FunctestUtilsTesting.readline = 0
 102
 103             mock_obj = mock.Mock()
 104             attrs = {'readline.side_effect': self.cmd_readline()}
 105             mock_obj.configure_mock(**attrs)
 106
 107             mock_obj2 = mock.Mock()
 108             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 109             mock_obj2.configure_mock(**attrs)
 110
 111             mock_subproc_open.return_value = mock_obj2
 112
 113             resp = functest_utils.execute_command(self.cmd, info=True,
 114                                                   error_msg=self.error_msg,
 115                                                   verbose=True,
 116                                                   output_file=self.output_file)
 117             self.assertEqual(resp, 1)
 118             msg_exec = ("Executing command: '%s'" % self.cmd)
 119             mock_logger_info.assert_called_once_with(msg_exec)
 120             mopen.assert_called_once_with(self.output_file, "w")
 121             mock_logger_error.assert_called_once_with(self.error_msg)
 122
 123     @mock.patch('functest.utils.functest_utils.LOGGER.info')
 124     def test_exec_cmd_args_present_ok(self, mock_logger_info):
 125         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 126                 as mock_subproc_open, \
 127                 mock.patch('six.moves.builtins.open',
 128                            mock.mock_open()) as mopen:
 129
 130             FunctestUtilsTesting.readline = 0
 131
 132             mock_obj = mock.Mock()
 133             attrs = {'readline.side_effect': self.cmd_readline()}
 134             mock_obj.configure_mock(**attrs)
 135
 136             mock_obj2 = mock.Mock()
 137             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 138             mock_obj2.configure_mock(**attrs)
 139
 140             mock_subproc_open.return_value = mock_obj2
 141
 142             resp = functest_utils.execute_command(self.cmd, info=True,
 143                                                   error_msg=self.error_msg,
 144                                                   verbose=True,
 145                                                   output_file=self.output_file)
 146             self.assertEqual(resp, 0)
 147             msg_exec = ("Executing command: '%s'" % self.cmd)
 148             mock_logger_info.assert_called_once_with(msg_exec)
 149             mopen.assert_called_once_with(self.output_file, "w")
 150
 151     @mock.patch('sys.stdout')
 152     def test_exec_cmd_args_missing_ok(self, stdout=None):
 153         # pylint: disable=unused-argument
 154         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 155                 as mock_subproc_open:
 156
 157             FunctestUtilsTesting.readline = 2
 158
 159             mock_obj = mock.Mock()
 160             attrs = {'readline.side_effect': self.cmd_readline()}
 161             mock_obj.configure_mock(**attrs)
 162
 163             mock_obj2 = mock.Mock()
 164             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 165             mock_obj2.configure_mock(**attrs)
 166
 167             mock_subproc_open.return_value = mock_obj2
 168
 169             resp = functest_utils.execute_command(self.cmd, info=False,
 170                                                   error_msg="",
 171                                                   verbose=False,
 172                                                   output_file=None)
 173             self.assertEqual(resp, 0)
 174
 175     @mock.patch('sys.stdout')
 176     def test_exec_cmd_args_missing_ko(self, stdout=None):
 177         # pylint: disable=unused-argument
 178         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 179                 as mock_subproc_open:
 180
 181             FunctestUtilsTesting.readline = 2
 182             mock_obj = mock.Mock()
 183             attrs = {'readline.side_effect': self.cmd_readline()}
 184             mock_obj.configure_mock(**attrs)
 185
 186             mock_obj2 = mock.Mock()
 187             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 188             mock_obj2.configure_mock(**attrs)
 189
 190             mock_subproc_open.return_value = mock_obj2
 191
 192             resp = functest_utils.execute_command(self.cmd, info=False,
 193                                                   error_msg="",
 194                                                   verbose=False,
 195                                                   output_file=None)
 196             self.assertEqual(resp, 1)
 197
 198     def test_get_param_from_yaml_failed(self):
 199         self.file_yaml['general'] = None
 200         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 201                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 202                 as mock_yaml, \
 203                 self.assertRaises(ValueError) as excep:
 204             mock_yaml.return_value = self.file_yaml
 205             functest_utils.get_parameter_from_yaml(self.parameter,
 206                                                    self.test_file)
 207             self.assertTrue(("The parameter %s is not"
 208                              " defined in config_functest.yaml" %
 209                              self.parameter) in excep.exception)
 210
 211     def test_get_param_from_yaml_def(self):
 212         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 213                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 214                 as mock_yaml:
 215             mock_yaml.return_value = self.file_yaml
 216             self.assertEqual(functest_utils.
 217                              get_parameter_from_yaml(self.parameter,
 218                                                      self.test_file),
 219                              'test_image_name')
 220
 221     def test_nova_version_exc1(self):
 222         # pylint: disable=protected-access
 223         cloud = mock.Mock()
 224         cloud._compute_client.request.return_value = None
 225         self.assertEqual(functest_utils.get_nova_version(cloud), None)
 226         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 227
 228     def test_nova_version_exc2(self):
 229         # pylint: disable=protected-access
 230         cloud = mock.Mock()
 231         cloud._compute_client.request.return_value = {"version": None}
 232         self.assertEqual(functest_utils.get_nova_version(cloud), None)
 233         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 234
 235     def test_nova_version_exc3(self):
 236         # pylint: disable=protected-access
 237         cloud = mock.Mock()
 238         cloud._compute_client.request.return_value = {
 239             "version": {"version": None}}
 240         self.assertEqual(functest_utils.get_nova_version(cloud), None)
 241         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 242
 243     def test_nova_version_exc4(self):
 244         # pylint: disable=protected-access
 245         cloud = mock.Mock()
 246         cloud._compute_client.request.return_value = {
 247             "version": {"version": "a.b"}}
 248         self.assertEqual(functest_utils.get_nova_version(cloud), None)
 249         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 250
 251     def test_nova_version(self):
 252         # pylint: disable=protected-access
 253         cloud = mock.Mock()
 254         cloud._compute_client.request.return_value = {
 255             "version": {"version": "2.1"}}
 256         self.assertEqual(functest_utils.get_nova_version(cloud), (2, 1))
 257         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 258
 259     @mock.patch('functest.utils.functest_utils.get_nova_version',
 260                 return_value=(2, 61))
 261     def test_openstack_version1(self, *args):
 262         cloud = mock.Mock()
 263         self.assertEqual(functest_utils.get_openstack_version(
 264             cloud), "Rocky or newer")
 265         args[0].assert_called_once_with(cloud)
 266
 267     @mock.patch('functest.utils.functest_utils.get_nova_version',
 268                 return_value=(2, 60))
 269     def test_openstack_version2(self, *args):
 270         cloud = mock.Mock()
 271         self.assertEqual(functest_utils.get_openstack_version(cloud), "Queens")
 272         args[0].assert_called_once_with(cloud)
 273
 274     @mock.patch('functest.utils.functest_utils.get_nova_version',
 275                 return_value=(2, 43))
 276     def test_openstack_version3(self, *args):
 277         cloud = mock.Mock()
 278         self.assertEqual(functest_utils.get_openstack_version(cloud), "Pike")
 279         args[0].assert_called_once_with(cloud)
 280
 281     @mock.patch('functest.utils.functest_utils.get_nova_version',
 282                 return_value=(2, 39))
 283     def test_openstack_version4(self, *args):
 284         cloud = mock.Mock()
 285         self.assertEqual(functest_utils.get_openstack_version(cloud), "Ocata")
 286         args[0].assert_called_once_with(cloud)
 287
 288     @mock.patch('functest.utils.functest_utils.get_nova_version',
 289                 return_value=(2, 26))
 290     def test_openstack_version5(self, *args):
 291         cloud = mock.Mock()
 292         self.assertEqual(functest_utils.get_openstack_version(cloud), "Newton")
 293         args[0].assert_called_once_with(cloud)
 294
 295     @mock.patch('functest.utils.functest_utils.get_nova_version',
 296                 return_value=(2, 13))
 297     def test_openstack_version6(self, *args):
 298         cloud = mock.Mock()
 299         self.assertEqual(functest_utils.get_openstack_version(cloud), "Mitaka")
 300         args[0].assert_called_once_with(cloud)
 301
 302     @mock.patch('functest.utils.functest_utils.get_nova_version',
 303                 return_value=(2, 4))
 304     def test_openstack_version7(self, *args):
 305         cloud = mock.Mock()
 306         self.assertEqual(
 307             functest_utils.get_openstack_version(cloud), "Liberty")
 308         args[0].assert_called_once_with(cloud)
 309
 310     @mock.patch('functest.utils.functest_utils.get_nova_version',
 311                 return_value=(2, 1))
 312     def test_openstack_version8(self, *args):
 313         cloud = mock.Mock()
 314         self.assertEqual(functest_utils.get_openstack_version(cloud), "Kilo")
 315         args[0].assert_called_once_with(cloud)
 316
 317     @mock.patch('functest.utils.functest_utils.get_nova_version',
 318                 return_value=(1, 9))
 319     def test_openstack_version9(self, *args):
 320         cloud = mock.Mock()
 321         self.assertEqual(
 322             functest_utils.get_openstack_version(cloud), "Unknown")
 323         args[0].assert_called_once_with(cloud)
 324
 325     @mock.patch('functest.utils.functest_utils.get_nova_version',
 326                 return_value=(3, 1))
 327     def test_openstack_version10(self, *args):
 328         cloud = mock.Mock()
 329         self.assertEqual(
 330             functest_utils.get_openstack_version(cloud), "Rocky or newer")
 331         args[0].assert_called_once_with(cloud)
 332
 333     @mock.patch('functest.utils.functest_utils.get_nova_version',
 334                 return_value=None)
 335     def test_openstack_version_exc(self, *args):
 336         cloud = mock.Mock()
 337         self.assertEqual(
 338             functest_utils.get_openstack_version(cloud), "Unknown")
 339         args[0].assert_called_once_with(cloud)
 340
 341
 342 if __name__ == "__main__":
 343     logging.disable(logging.CRITICAL)
 344     unittest.main(verbosity=2)