3 # Copyright (c) 2016 Orange and others.
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
10 # pylint: disable=missing-docstring
19 from six.moves import urllib
21 from functest.utils import functest_utils
24 class FunctestUtilsTesting(unittest.TestCase):
27 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
30 self.url = 'http://www.opnfv.org/'
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'
47 self.start_date = 1482624000
48 self.stop_date = 1482624000
49 self.start_time = time.time()
50 self.stop_time = time.time()
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'
56 self.output_file = 'test_output_file'
57 self.testname = 'testname'
58 self.testcase_dict = {'case_name': 'testname',
59 'criteria': self.criteria}
60 self.parameter = 'general.openstack.image_name'
61 self.config_yaml = pkg_resources.resource_filename(
62 'functest', 'ci/config_functest.yaml')
63 self.db_url_env = 'http://foo/testdb'
64 self.testcases_yaml = "test_testcases_yaml"
65 self.file_yaml = {'general': {'openstack': {'image_name':
68 @mock.patch('six.moves.urllib.request.urlopen',
69 side_effect=urllib.error.URLError('no host given'))
70 def test_check_internet_connectivity_failed(self, mock_method):
71 self.assertFalse(functest_utils.check_internet_connectivity())
72 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
74 @mock.patch('six.moves.urllib.request.urlopen')
75 def test_check_internet_connectivity_default(self, mock_method):
76 self.assertTrue(functest_utils.check_internet_connectivity())
77 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
79 @mock.patch('six.moves.urllib.request.urlopen')
80 def test_check_internet_connectivity_debian(self, mock_method):
81 self.url = "https://www.debian.org/"
82 self.assertTrue(functest_utils.check_internet_connectivity(self.url))
83 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
85 @mock.patch('six.moves.urllib.request.urlopen',
86 side_effect=urllib.error.URLError('no host given'))
87 def test_download_url_failed(self, mock_url):
88 self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
90 @mock.patch('six.moves.urllib.request.urlopen')
91 def test_download_url_default(self, mock_url):
92 with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
93 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
95 name = self.url.rsplit('/')[-1]
96 dest = self.dest_path + "/" + name
97 self.assertTrue(functest_utils.download_url(self.url,
99 m.assert_called_once_with(dest, 'wb')
100 self.assertTrue(mock_sh.called)
102 def _get_env_dict(self, var):
103 dic = {'INSTALLER_TYPE': self.installer,
104 'DEPLOY_SCENARIO': self.scenario,
105 'NODE_NAME': self.node_name,
106 'BUILD_TAG': self.build_tag}
112 if FunctestUtilsTesting.readline == \
113 len(FunctestUtilsTesting.test_ip) - 1:
115 FunctestUtilsTesting.readline += 1
116 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
118 # TODO: get_resolvconf_ns
119 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
120 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
121 attrs = {'query.return_value': ["test"]}
122 mock_dns_resolve.configure_mock(**attrs)
125 attrs = {'readline.side_effect': self.readline_side}
126 m.configure_mock(**attrs)
128 with mock.patch("six.moves.builtins.open") as mo:
130 self.assertEqual(functest_utils.get_resolvconf_ns(),
133 def _get_environ(self, var):
134 if var == 'INSTALLER_TYPE':
135 return self.installer
136 elif var == 'DEPLOY_SCENARIO':
140 def test_get_ci_envvars_default(self):
141 with mock.patch('os.environ.get',
142 side_effect=self._get_environ):
143 dic = {"installer": self.installer,
144 "scenario": self.scenario}
145 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
147 def cmd_readline(self):
148 return 'test_value\n'
150 @mock.patch('functest.utils.functest_utils.LOGGER.error')
151 @mock.patch('functest.utils.functest_utils.LOGGER.info')
152 def test_execute_command_args_present_with_error(self, mock_logger_info,
154 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
155 as mock_subproc_open, \
156 mock.patch('six.moves.builtins.open',
157 mock.mock_open()) as mopen:
159 FunctestUtilsTesting.readline = 0
161 mock_obj = mock.Mock()
162 attrs = {'readline.side_effect': self.cmd_readline()}
163 mock_obj.configure_mock(**attrs)
165 mock_obj2 = mock.Mock()
166 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
167 mock_obj2.configure_mock(**attrs)
169 mock_subproc_open.return_value = mock_obj2
171 resp = functest_utils.execute_command(self.cmd, info=True,
172 error_msg=self.error_msg,
174 output_file=self.output_file)
175 self.assertEqual(resp, 1)
176 msg_exec = ("Executing command: '%s'" % self.cmd)
177 mock_logger_info.assert_called_once_with(msg_exec)
178 mopen.assert_called_once_with(self.output_file, "w")
179 mock_logger_error.assert_called_once_with(self.error_msg)
181 @mock.patch('functest.utils.functest_utils.LOGGER.info')
182 def test_execute_command_args_present_with_success(self, mock_logger_info,
184 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
185 as mock_subproc_open, \
186 mock.patch('six.moves.builtins.open',
187 mock.mock_open()) as mopen:
189 FunctestUtilsTesting.readline = 0
191 mock_obj = mock.Mock()
192 attrs = {'readline.side_effect': self.cmd_readline()}
193 mock_obj.configure_mock(**attrs)
195 mock_obj2 = mock.Mock()
196 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
197 mock_obj2.configure_mock(**attrs)
199 mock_subproc_open.return_value = mock_obj2
201 resp = functest_utils.execute_command(self.cmd, info=True,
202 error_msg=self.error_msg,
204 output_file=self.output_file)
205 self.assertEqual(resp, 0)
206 msg_exec = ("Executing command: '%s'" % self.cmd)
207 mock_logger_info.assert_called_once_with(msg_exec)
208 mopen.assert_called_once_with(self.output_file, "w")
210 @mock.patch('sys.stdout')
211 def test_execute_command_args_missing_with_success(self, stdout=None):
212 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
213 as mock_subproc_open:
215 FunctestUtilsTesting.readline = 2
217 mock_obj = mock.Mock()
218 attrs = {'readline.side_effect': self.cmd_readline()}
219 mock_obj.configure_mock(**attrs)
221 mock_obj2 = mock.Mock()
222 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
223 mock_obj2.configure_mock(**attrs)
225 mock_subproc_open.return_value = mock_obj2
227 resp = functest_utils.execute_command(self.cmd, info=False,
231 self.assertEqual(resp, 0)
233 @mock.patch('sys.stdout')
234 def test_execute_command_args_missing_with_error(self, stdout=None):
235 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
236 as mock_subproc_open:
238 FunctestUtilsTesting.readline = 2
239 mock_obj = mock.Mock()
240 attrs = {'readline.side_effect': self.cmd_readline()}
241 mock_obj.configure_mock(**attrs)
243 mock_obj2 = mock.Mock()
244 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
245 mock_obj2.configure_mock(**attrs)
247 mock_subproc_open.return_value = mock_obj2
249 resp = functest_utils.execute_command(self.cmd, info=False,
253 self.assertEqual(resp, 1)
255 def _get_functest_config(self, var):
258 @mock.patch('functest.utils.functest_utils.LOGGER.error')
259 def test_get_dict_by_test(self, mock_logger_error):
260 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
261 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
263 mock_obj = mock.Mock()
264 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
265 mock_obj.configure_mock(**attrs)
267 mock_yaml.return_value = mock_obj
269 self.assertDictEqual(functest_utils.
270 get_dict_by_test(self.testname),
273 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
274 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
275 mock_get_dict_by_test.return_value = self.testcase_dict
276 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
279 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
280 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
281 mock_get_dict_by_test.return_value = None
282 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
284 def test_get_parameter_from_yaml_failed(self):
285 self.file_yaml['general'] = None
286 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
287 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
289 self.assertRaises(ValueError) as excep:
290 mock_yaml.return_value = self.file_yaml
291 functest_utils.get_parameter_from_yaml(self.parameter,
293 self.assertTrue(("The parameter %s is not"
294 " defined in config_functest.yaml" %
295 self.parameter) in excep.exception)
297 def test_get_parameter_from_yaml_default(self):
298 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
299 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
301 mock_yaml.return_value = self.file_yaml
302 self.assertEqual(functest_utils.
303 get_parameter_from_yaml(self.parameter,
307 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
308 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
309 with mock.patch.dict(os.environ,
310 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
311 functest_utils.get_functest_config(self.parameter)
312 mock_get_parameter_from_yaml. \
313 assert_called_once_with(self.parameter,
316 def test_get_functest_yaml(self):
317 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
318 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
320 mock_yaml.return_value = self.file_yaml
321 resp = functest_utils.get_functest_yaml()
322 self.assertEqual(resp, self.file_yaml)
324 @mock.patch('functest.utils.functest_utils.LOGGER.info')
325 def test_print_separator(self, mock_logger_info):
326 functest_utils.print_separator()
327 mock_logger_info.assert_called_once_with("======================="
328 "=======================")
331 if __name__ == "__main__":
332 logging.disable(logging.CRITICAL)
333 unittest.main(verbosity=2)