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 os
  15 import unittest
  16
  17 import mock
  18 import munch
  19 import pkg_resources
  20
  21 from functest.utils import functest_utils
  22
  23
  24 class FunctestUtilsTesting(unittest.TestCase):
  25     # pylint: disable=too-many-instance-attributes
  26
  27     readline = 0
  28     test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
  29
  30     def setUp(self):
  31         self.url = 'http://www.opnfv.org/'
  32         self.timeout = 5
  33         self.dest_path = 'test_path'
  34         self.repo_path = 'test_repo_path'
  35         self.installer = 'test_installer'
  36         self.scenario = 'test_scenario'
  37         self.build_tag = 'jenkins-functest-fuel-opnfv-jump-2-daily-master-190'
  38         self.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
  39         self.version = 'master'
  40         self.node_name = 'test_node_name'
  41         self.project = 'test_project'
  42         self.case_name = 'test_case_name'
  43         self.status = 'test_status'
  44         self.details = 'test_details'
  45         self.db_url = 'test_db_url'
  46         self.criteria = 50
  47         self.result = 75
  48         self.start_date = 1482624000
  49         self.stop_date = 1482624000
  50         self.start_time = time.time()
  51         self.stop_time = time.time()
  52         self.readline = -1
  53         self.test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
  54         self.test_file = 'test_file'
  55         self.error_msg = 'test_error_msg'
  56         self.cmd = 'test_cmd'
  57         self.output_file = 'test_output_file'
  58         self.testname = 'testname'
  59         self.parameter = 'general.openstack.image_name'
  60         self.config_yaml = pkg_resources.resource_filename(
  61             'functest', 'ci/config_functest.yaml')
  62         self.db_url_env = 'http://foo/testdb'
  63         self.testcases_yaml = "test_testcases_yaml"
  64         self.file_yaml = {'general': {'openstack': {'image_name':
  65                                                     'test_image_name'}}}
  66
  67     def _get_env_dict(self, var):
  68         dic = {'INSTALLER_TYPE': self.installer,
  69                'DEPLOY_SCENARIO': self.scenario,
  70                'NODE_NAME': self.node_name,
  71                'BUILD_TAG': self.build_tag}
  72         dic.pop(var, None)
  73         return dic
  74
  75     @staticmethod
  76     def readline_side():
  77         if FunctestUtilsTesting.readline == \
  78                 len(FunctestUtilsTesting.test_ip) - 1:
  79             return False
  80         FunctestUtilsTesting.readline += 1
  81         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
  82
  83     def _get_environ(self, var, *args):  # pylint: disable=unused-argument
  84         if var == 'INSTALLER_TYPE':
  85             return self.installer
  86         elif var == 'DEPLOY_SCENARIO':
  87             return self.scenario
  88         return var
  89
  90     @staticmethod
  91     def cmd_readline():
  92         return 'test_value\n'
  93
  94     @mock.patch('functest.utils.functest_utils.LOGGER.error')
  95     @mock.patch('functest.utils.functest_utils.LOGGER.info')
  96     def test_exec_cmd_args_present_ko(self, mock_logger_info,
  97                                       mock_logger_error):
  98         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
  99                 as mock_subproc_open, \
 100                 mock.patch('six.moves.builtins.open',
 101                            mock.mock_open()) as mopen:
 102
 103             FunctestUtilsTesting.readline = 0
 104
 105             mock_obj = mock.Mock()
 106             attrs = {'readline.side_effect': self.cmd_readline()}
 107             mock_obj.configure_mock(**attrs)
 108
 109             mock_obj2 = mock.Mock()
 110             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 111             mock_obj2.configure_mock(**attrs)
 112
 113             mock_subproc_open.return_value = mock_obj2
 114
 115             resp = functest_utils.execute_command(self.cmd, info=True,
 116                                                   error_msg=self.error_msg,
 117                                                   verbose=True,
 118                                                   output_file=self.output_file)
 119             self.assertEqual(resp, 1)
 120             msg_exec = ("Executing command: '%s'" % self.cmd)
 121             mock_logger_info.assert_called_once_with(msg_exec)
 122             mopen.assert_called_once_with(self.output_file, "w")
 123             mock_logger_error.assert_called_once_with(self.error_msg)
 124
 125     @mock.patch('functest.utils.functest_utils.LOGGER.info')
 126     def test_exec_cmd_args_present_ok(self, mock_logger_info):
 127         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 128                 as mock_subproc_open, \
 129                 mock.patch('six.moves.builtins.open',
 130                            mock.mock_open()) as mopen:
 131
 132             FunctestUtilsTesting.readline = 0
 133
 134             mock_obj = mock.Mock()
 135             attrs = {'readline.side_effect': self.cmd_readline()}
 136             mock_obj.configure_mock(**attrs)
 137
 138             mock_obj2 = mock.Mock()
 139             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
 140             mock_obj2.configure_mock(**attrs)
 141
 142             mock_subproc_open.return_value = mock_obj2
 143
 144             resp = functest_utils.execute_command(self.cmd, info=True,
 145                                                   error_msg=self.error_msg,
 146                                                   verbose=True,
 147                                                   output_file=self.output_file)
 148             self.assertEqual(resp, 0)
 149             msg_exec = ("Executing command: '%s'" % self.cmd)
 150             mock_logger_info.assert_called_once_with(msg_exec)
 151             mopen.assert_called_once_with(self.output_file, "w")
 152
 153     @mock.patch('sys.stdout')
 154     def test_exec_cmd_args_missing_ok(self, stdout=None):
 155         # pylint: disable=unused-argument
 156         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 157                 as mock_subproc_open:
 158
 159             FunctestUtilsTesting.readline = 2
 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': 0}
 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=False,
 172                                                   error_msg="",
 173                                                   verbose=False,
 174                                                   output_file=None)
 175             self.assertEqual(resp, 0)
 176
 177     @mock.patch('sys.stdout')
 178     def test_exec_cmd_args_missing_ko(self, stdout=None):
 179         # pylint: disable=unused-argument
 180         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
 181                 as mock_subproc_open:
 182
 183             FunctestUtilsTesting.readline = 2
 184             mock_obj = mock.Mock()
 185             attrs = {'readline.side_effect': self.cmd_readline()}
 186             mock_obj.configure_mock(**attrs)
 187
 188             mock_obj2 = mock.Mock()
 189             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
 190             mock_obj2.configure_mock(**attrs)
 191
 192             mock_subproc_open.return_value = mock_obj2
 193
 194             resp = functest_utils.execute_command(self.cmd, info=False,
 195                                                   error_msg="",
 196                                                   verbose=False,
 197                                                   output_file=None)
 198             self.assertEqual(resp, 1)
 199
 200     def test_get_param_from_yaml_failed(self):
 201         self.file_yaml['general'] = None
 202         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 203                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 204                 as mock_yaml, \
 205                 self.assertRaises(ValueError) as excep:
 206             mock_yaml.return_value = self.file_yaml
 207             functest_utils.get_parameter_from_yaml(self.parameter,
 208                                                    self.test_file)
 209             self.assertTrue(("The parameter %s is not"
 210                              " defined in config_functest.yaml" %
 211                              self.parameter) in excep.exception)
 212
 213     def test_get_param_from_yaml_def(self):
 214         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
 215                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
 216                 as mock_yaml:
 217             mock_yaml.return_value = self.file_yaml
 218             self.assertEqual(functest_utils.
 219                              get_parameter_from_yaml(self.parameter,
 220                                                      self.test_file),
 221                              'test_image_name')
 222
 223     def test_get_extnetwork_nocloud(self):
 224         with self.assertRaises(AssertionError):
 225             functest_utils.get_external_network(None)
 226
 227     def test_get_extnetwork_env_ok1(self):
 228         cloud = mock.Mock()
 229         cloud.get_network.return_value = munch.Munch(name="dummy")
 230         os.environ["EXTERNAL_NETWORK"] = 'dummy'
 231         self.assertEqual(
 232             functest_utils.get_external_network(cloud),
 233             cloud.get_network.return_value)
 234         cloud.get_network.assert_called_once_with(
 235             'dummy', {'router:external': True})
 236         cloud.list_networks.assert_not_called()
 237
 238     def test_get_extnetwork_env_ok2(self):
 239         cloud = mock.Mock()
 240         cloud.get_network.return_value = None
 241         cloud.list_networks.return_value = None
 242         os.environ["EXTERNAL_NETWORK"] = 'dummy'
 243         self.assertEqual(functest_utils.get_external_network(cloud), None)
 244         cloud.get_network.assert_called_once_with(
 245             'dummy', {'router:external': True})
 246         cloud.list_networks.assert_called_once_with(
 247             {'router:external': True})
 248
 249     def test_get_extnetwork_env_ko(self):
 250         cloud = mock.Mock()
 251         cloud.get_network.return_value = None
 252         cloud.list_networks.return_value = [munch.Munch(name="dummy")]
 253         os.environ["EXTERNAL_NETWORK"] = 'dummy'
 254         self.assertEqual(
 255             functest_utils.get_external_network(cloud),
 256             cloud.list_networks.return_value[0])
 257         cloud.get_network.assert_called_once_with(
 258             'dummy', {'router:external': True})
 259         cloud.list_networks.assert_called_once_with(
 260             {'router:external': True})
 261
 262     def test_get_extnetwork_noenv_ko(self):
 263         try:
 264             del os.environ["EXTERNAL_NETWORK"]
 265         except Exception:  # pylint: disable=broad-except
 266             pass
 267         cloud = mock.Mock()
 268         cloud.list_networks.return_value = None
 269         self.assertEqual(functest_utils.get_external_network(cloud), None)
 270         cloud.get_network.assert_not_called()
 271         cloud.list_networks.assert_called_once_with(
 272             {'router:external': True})
 273
 274     def test_get_extnetwork_noenv_ok(self):
 275         try:
 276             del os.environ["EXTERNAL_NETWORK"]
 277         except Exception:  # pylint: disable=broad-except
 278             pass
 279         cloud = mock.Mock()
 280         cloud.list_networks.return_value = [munch.Munch(name="dummy")]
 281         self.assertEqual(
 282             functest_utils.get_external_network(cloud),
 283             cloud.list_networks.return_value[0])
 284         cloud.get_network.assert_not_called()
 285         cloud.list_networks.assert_called_once_with(
 286             {'router:external': True})
 287
 288 if __name__ == "__main__":
 289     logging.disable(logging.CRITICAL)
 290     unittest.main(verbosity=2)