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 import six
  19
  20 from functest.utils import functest_utils
  21
  22
  23 class FunctestUtilsTesting(unittest.TestCase):
  24     # pylint: disable=too-many-instance-attributes,too-many-public-methods
  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.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     def _get_env_dict(self, var):
  67         dic = {'INSTALLER_TYPE': self.installer,
  68                'DEPLOY_SCENARIO': self.scenario,
  69                'NODE_NAME': self.node_name,
  70                'BUILD_TAG': self.build_tag}
  71         dic.pop(var, None)
  72         return dic
  73
  74     @staticmethod
  75     def readline_side():
  76         if FunctestUtilsTesting.readline == \
  77                 len(FunctestUtilsTesting.test_ip) - 1:
  78             return False
  79         FunctestUtilsTesting.readline += 1
  80         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
  81
  82     def _get_environ(self, var, *args):  # pylint: disable=unused-argument
  83         if var == 'INSTALLER_TYPE':
  84             return self.installer
  85         if var == 'DEPLOY_SCENARIO':
  86             return self.scenario
  87         return var
  88
  89     @staticmethod
  90     def cmd_readline():
  91         return 'test_value\n'
  92
  93     @mock.patch('functest.utils.functest_utils.LOGGER.error')
  94     @mock.patch('functest.utils.functest_utils.LOGGER.info')
  95     def test_exec_cmd_args_present_ko(self, mock_logger_info,
  96                                       mock_logger_error):
  97         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
  98                 as mock_subproc_open, \
  99                 mock.patch('six.moves.builtins.open',
 100                            mock.mock_open()) as mopen:
 101             stream = six.BytesIO()
 102             stream.write(self.cmd_readline().encode("utf-8"))
 103             attrs = {
 104                 'return_value.__enter__.return_value.stdout': stream,
 105                 'return_value.__enter__.return_value.wait.return_value': 1}
 106             mock_subproc_open.configure_mock(**attrs)
 107             resp = functest_utils.execute_command(
 108                 self.cmd, info=True, error_msg=self.error_msg, verbose=True,
 109                 output_file=self.output_file)
 110             self.assertEqual(resp, 1)
 111             msg_exec = f"Executing command: '{self.cmd}'"
 112             mock_logger_info.assert_called_once_with(msg_exec)
 113             mopen.assert_called_once_with(
 114                 self.output_file, "w", encoding='utf-8')
 115             mock_logger_error.assert_called_once_with(self.error_msg)
 116
 117     @mock.patch('functest.utils.functest_utils.LOGGER.info')
 118     def test_exec_cmd_args_present_ok(self, mock_logger_info):
 119         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 120                 as mock_subproc_open, \
 121                 mock.patch('six.moves.builtins.open',
 122                            mock.mock_open()) as mopen:
 123             stream = six.BytesIO()
 124             stream.write(self.cmd_readline().encode("utf-8"))
 125             attrs = {
 126                 'return_value.__enter__.return_value.stdout': stream,
 127                 'return_value.__enter__.return_value.wait.return_value': 0}
 128             mock_subproc_open.configure_mock(**attrs)
 129             resp = functest_utils.execute_command(
 130                 self.cmd, info=True, error_msg=self.error_msg, verbose=True,
 131                 output_file=self.output_file)
 132             self.assertEqual(resp, 0)
 133             msg_exec = (f"Executing command: '{self.cmd}'")
 134             mock_logger_info.assert_called_once_with(msg_exec)
 135             mopen.assert_called_once_with(
 136                 self.output_file, "w", encoding='utf-8')
 137
 138     @mock.patch('sys.stdout')
 139     def test_exec_cmd_args_missing_ok(self, stdout=None):
 140         # pylint: disable=unused-argument
 141         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 142                 as mock_subproc_open:
 143             stream = six.BytesIO()
 144             stream.write(self.cmd_readline().encode("utf-8"))
 145             attrs = {
 146                 'return_value.__enter__.return_value.stdout': stream,
 147                 'return_value.__enter__.return_value.wait.return_value': 0}
 148             mock_subproc_open.configure_mock(**attrs)
 149             resp = functest_utils.execute_command(
 150                 self.cmd, info=False, error_msg="", verbose=False,
 151                 output_file=None)
 152             self.assertEqual(resp, 0)
 153
 154     @mock.patch('sys.stdout')
 155     def test_exec_cmd_args_missing_ko(self, stdout=None):
 156         # pylint: disable=unused-argument
 157         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 158                 as mock_subproc_open:
 159             attrs = {}
 160             stream = six.BytesIO()
 161             stream.write(self.cmd_readline().encode("utf-8"))
 162             attrs = {
 163                 'return_value.__enter__.return_value.stdout': stream,
 164                 'return_value.__enter__.return_value.wait.return_value': 1}
 165             mock_subproc_open.configure_mock(**attrs)
 166             resp = functest_utils.execute_command(
 167                 self.cmd, info=False, error_msg="", verbose=False,
 168                 output_file=None)
 169             self.assertEqual(resp, 1)
 170
 171     def test_get_param_from_yaml_failed(self):
 172         self.file_yaml['general'] = None
 173         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 174                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 175                 as mock_yaml, \
 176                 self.assertRaises(ValueError) as excep:
 177             mock_yaml.return_value = self.file_yaml
 178             functest_utils.get_parameter_from_yaml(self.parameter,
 179                                                    self.test_file)
 180             self.assertTrue((f"The parameter {self.parameter} is not"
 181                              " defined in config_functest.yaml"
 182                              ) in excep.exception)
 183
 184     def test_get_param_from_yaml_def(self):
 185         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 186                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 187                 as mock_yaml:
 188             mock_yaml.return_value = self.file_yaml
 189             self.assertEqual(functest_utils.
 190                              get_parameter_from_yaml(self.parameter,
 191                                                      self.test_file),
 192                              'test_image_name')
 193
 194     def test_nova_version_exc1(self):
 195         # pylint: disable=protected-access
 196         cloud = mock.Mock()
 197         cloud._compute_client.request.return_value = None
 198         self.assertEqual(functest_utils.get_nova_version(cloud), None)
 199         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 200
 201     def test_nova_version_exc2(self):
 202         # pylint: disable=protected-access
 203         cloud = mock.Mock()
 204         cloud._compute_client.request.return_value = {"version": None}
 205         self.assertEqual(functest_utils.get_nova_version(cloud), None)
 206         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 207
 208     def test_nova_version_exc3(self):
 209         # pylint: disable=protected-access
 210         cloud = mock.Mock()
 211         cloud._compute_client.request.return_value = {
 212             "version": {"version": None}}
 213         self.assertEqual(functest_utils.get_nova_version(cloud), None)
 214         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 215
 216     def test_nova_version_exc4(self):
 217         # pylint: disable=protected-access
 218         cloud = mock.Mock()
 219         cloud._compute_client.request.return_value = {
 220             "version": {"version": "a.b"}}
 221         self.assertEqual(functest_utils.get_nova_version(cloud), None)
 222         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 223
 224     def test_nova_version(self):
 225         # pylint: disable=protected-access
 226         cloud = mock.Mock()
 227         cloud._compute_client.request.return_value = {
 228             "version": {"version": "2.1"}}
 229         self.assertEqual(functest_utils.get_nova_version(cloud), (2, 1))
 230         cloud._compute_client.request.assert_called_once_with('/', 'GET')
 231
 232     @mock.patch('functest.utils.functest_utils.get_nova_version',
 233                 return_value=(2, 61))
 234     def test_openstack_version1(self, *args):
 235         cloud = mock.Mock()
 236         self.assertEqual(functest_utils.get_openstack_version(
 237             cloud), "Rocky")
 238         args[0].assert_called_once_with(cloud)
 239
 240     @mock.patch('functest.utils.functest_utils.get_nova_version',
 241                 return_value=(2, 60))
 242     def test_openstack_version2(self, *args):
 243         cloud = mock.Mock()
 244         self.assertEqual(functest_utils.get_openstack_version(cloud), "Queens")
 245         args[0].assert_called_once_with(cloud)
 246
 247     @mock.patch('functest.utils.functest_utils.get_nova_version',
 248                 return_value=(2, 43))
 249     def test_openstack_version3(self, *args):
 250         cloud = mock.Mock()
 251         self.assertEqual(functest_utils.get_openstack_version(cloud), "Pike")
 252         args[0].assert_called_once_with(cloud)
 253
 254     @mock.patch('functest.utils.functest_utils.get_nova_version',
 255                 return_value=(2, 39))
 256     def test_openstack_version4(self, *args):
 257         cloud = mock.Mock()
 258         self.assertEqual(functest_utils.get_openstack_version(cloud), "Ocata")
 259         args[0].assert_called_once_with(cloud)
 260
 261     @mock.patch('functest.utils.functest_utils.get_nova_version',
 262                 return_value=(2, 26))
 263     def test_openstack_version5(self, *args):
 264         cloud = mock.Mock()
 265         self.assertEqual(functest_utils.get_openstack_version(cloud), "Newton")
 266         args[0].assert_called_once_with(cloud)
 267
 268     @mock.patch('functest.utils.functest_utils.get_nova_version',
 269                 return_value=(2, 13))
 270     def test_openstack_version6(self, *args):
 271         cloud = mock.Mock()
 272         self.assertEqual(functest_utils.get_openstack_version(cloud), "Mitaka")
 273         args[0].assert_called_once_with(cloud)
 274
 275     @mock.patch('functest.utils.functest_utils.get_nova_version',
 276                 return_value=(2, 4))
 277     def test_openstack_version7(self, *args):
 278         cloud = mock.Mock()
 279         self.assertEqual(
 280             functest_utils.get_openstack_version(cloud), "Liberty")
 281         args[0].assert_called_once_with(cloud)
 282
 283     @mock.patch('functest.utils.functest_utils.get_nova_version',
 284                 return_value=(2, 1))
 285     def test_openstack_version8(self, *args):
 286         cloud = mock.Mock()
 287         self.assertEqual(functest_utils.get_openstack_version(cloud), "Kilo")
 288         args[0].assert_called_once_with(cloud)
 289
 290     @mock.patch('functest.utils.functest_utils.get_nova_version',
 291                 return_value=(1, 9))
 292     def test_openstack_version9(self, *args):
 293         cloud = mock.Mock()
 294         self.assertEqual(
 295             functest_utils.get_openstack_version(cloud), "Unknown")
 296         args[0].assert_called_once_with(cloud)
 297
 298     @mock.patch('functest.utils.functest_utils.get_nova_version',
 299                 return_value=(3, 1))
 300     def test_openstack_version10(self, *args):
 301         cloud = mock.Mock()
 302         self.assertEqual(
 303             functest_utils.get_openstack_version(cloud), "Master")
 304         args[0].assert_called_once_with(cloud)
 305
 306     @mock.patch('functest.utils.functest_utils.get_nova_version',
 307                 return_value=(2, 66))
 308     def test_openstack_version11(self, *args):
 309         cloud = mock.Mock()
 310         self.assertEqual(functest_utils.get_openstack_version(
 311             cloud), "Stein")
 312         args[0].assert_called_once_with(cloud)
 313
 314     @mock.patch('functest.utils.functest_utils.get_nova_version',
 315                 return_value=(2, 78))
 316     def test_openstack_version12(self, *args):
 317         cloud = mock.Mock()
 318         self.assertEqual(functest_utils.get_openstack_version(
 319             cloud), "Train")
 320         args[0].assert_called_once_with(cloud)
 321
 322     @mock.patch('functest.utils.functest_utils.get_nova_version',
 323                 return_value=(2, 87))
 324     def test_openstack_version13(self, *args):
 325         cloud = mock.Mock()
 326         self.assertEqual(functest_utils.get_openstack_version(
 327             cloud), "Ussuri")
 328         args[0].assert_called_once_with(cloud)
 329
 330     @mock.patch('functest.utils.functest_utils.get_nova_version',
 331                 return_value=(2, 88))
 332     def test_openstack_version14(self, *args):
 333         cloud = mock.Mock()
 334         self.assertEqual(functest_utils.get_openstack_version(
 335             cloud), "Wallaby")
 336         args[0].assert_called_once_with(cloud)
 337
 338     @mock.patch('functest.utils.functest_utils.get_nova_version',
 339                 return_value=(2, 92))
 340     def test_openstack_version14(self, *args):
 341         cloud = mock.Mock()
 342         self.assertEqual(functest_utils.get_openstack_version(
 343             cloud), "Zed")
 344         args[0].assert_called_once_with(cloud)
 345
 346     @mock.patch('functest.utils.functest_utils.get_nova_version',
 347                 return_value=(2, 89))
 348     def test_openstack_version14(self, *args):
 349         cloud = mock.Mock()
 350         self.assertEqual(functest_utils.get_openstack_version(
 351             cloud), "Xena")
 352         args[0].assert_called_once_with(cloud)
 353
 354     @mock.patch('functest.utils.functest_utils.get_nova_version',
 355                 return_value=None)
 356     def test_openstack_version_exc(self, *args):
 357         cloud = mock.Mock()
 358         self.assertEqual(
 359             functest_utils.get_openstack_version(cloud), "Unknown")
 360         args[0].assert_called_once_with(cloud)
 361
 362     def test_convert_dict_to_ini(self):
 363         self.assertEqual(
 364             functest_utils.convert_dict_to_ini({}), "")
 365         self.assertEqual(
 366             functest_utils.convert_dict_to_ini({"a": "b"}), "a:b")
 367         value = functest_utils.convert_dict_to_ini({"a": "b", "c": "d"})
 368         self.assertTrue(value in ('a:b,c:d', 'c:d,a:b'))
 369         with self.assertRaises(AssertionError):
 370             functest_utils.convert_list_to_ini("")
 371
 372     def test_convert_list_to_ini(self):
 373         self.assertEqual(
 374             functest_utils.convert_list_to_ini([]), "")
 375         self.assertEqual(
 376             functest_utils.convert_list_to_ini(["a"]), "a")
 377         self.assertEqual(
 378             functest_utils.convert_list_to_ini(["a", "b"]), "a,b")
 379         with self.assertRaises(AssertionError):
 380             functest_utils.convert_list_to_ini("")
 381
 382     def test_convert_ini_to_dict(self):
 383         self.assertEqual(
 384             functest_utils.convert_ini_to_dict(""), {})
 385         self.assertEqual(
 386             functest_utils.convert_ini_to_dict("a:b"), {"a": "b"})
 387         self.assertEqual(
 388             functest_utils.convert_ini_to_dict(
 389                 "a:b,c:d"), {"a": "b", "c": "d"})
 390         self.assertEqual(
 391             functest_utils.convert_ini_to_dict(
 392                 "a:b:c,d:e:f"), {"a:b": "c", "d:e": "f"})
 393         with self.assertRaises(AssertionError):
 394             functest_utils.convert_list_to_ini({})
 395
 396     def test_convert_ini_to_list(self):
 397         self.assertEqual(
 398             functest_utils.convert_ini_to_list(""), [])
 399         self.assertEqual(
 400             functest_utils.convert_ini_to_list("a"), ["a"])
 401         self.assertEqual(
 402             functest_utils.convert_ini_to_list("a,b"), ["a", "b"])
 403         with self.assertRaises(AssertionError):
 404             functest_utils.convert_ini_to_list([])
 405
 406
 407 if __name__ == "__main__":
 408     logging.disable(logging.CRITICAL)
 409     unittest.main(verbosity=2)