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
18 from six.moves import urllib
20 from functest.tests.unit import test_utils
21 from functest.utils import functest_utils
22 from functest.utils.constants import CONST
25 class FunctestUtilsTesting(unittest.TestCase):
28 self.url = 'http://www.opnfv.org/'
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'
45 self.start_date = 1482624000
46 self.stop_date = 1482624000
47 self.start_time = time.time()
48 self.stop_time = time.time()
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'
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':
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)
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)
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)
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))
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')\
93 name = self.url.rsplit('/')[-1]
94 dest = self.dest_path + "/" + name
95 self.assertTrue(functest_utils.download_url(self.url,
97 m.assert_called_once_with(dest, 'wb')
98 self.assertTrue(mock_sh.called)
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,
105 self.assertEqual(functest_utils.get_installer_type(),
107 mock_logger_error.assert_called_once_with("Impossible to retrieve"
108 " the installer type")
110 def test_get_installer_type_default(self):
111 with mock.patch.dict(os.environ,
112 {'INSTALLER_TYPE': 'test_installer'},
114 self.assertEqual(functest_utils.get_installer_type(),
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,
122 self.assertEqual(functest_utils.get_scenario(),
123 "os-nosdn-nofeature-noha")
124 mock_logger_info.assert_called_once_with("Impossible to retrieve "
127 "os-nosdn-nofeature-noha")
129 def test_get_scenario_default(self):
130 with mock.patch.dict(os.environ,
131 {'DEPLOY_SCENARIO': 'test_scenario'},
133 self.assertEqual(functest_utils.get_scenario(),
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)
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)
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')
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")
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,
161 self.assertEqual(functest_utils.get_pod_name(),
163 mock_logger_info.assert_called_once_with("Unable to retrieve "
165 "environment. Using "
166 "pod name 'unknown-pod'")
168 def test_get_pod_name_default(self):
169 with mock.patch.dict(os.environ,
170 {'NODE_NAME': 'test_node_name'},
172 self.assertEqual(functest_utils.get_pod_name(),
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,
180 self.assertEqual(functest_utils.get_build_tag(),
182 mock_logger_info.assert_called_once_with("Impossible to retrieve"
185 def test_get_build_tag_default(self):
186 with mock.patch.dict(os.environ,
187 {'BUILD_TAG': self.build_tag},
189 self.assertEqual(functest_utils.get_build_tag(),
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(
207 "****************************************\n"
208 "\t %(p)s/%(n)s results \n\n"
209 "****************************************\n"
215 "build tag:\t%(b)s\n"
217 % {'p': self.project,
219 'db': CONST.__getattribute__('results_test_db_url'),
220 'pod': self.node_name,
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}
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,
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: " +
250 def test_push_results_to_db_missing_installer(self):
251 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
253 def test_push_results_to_db_missing_scenario(self):
254 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
256 def test_push_results_to_db_missing_nodename(self):
257 self._test_push_results_to_db_missing_env('NODE_NAME')
259 def test_push_results_to_db_missing_buildtag(self):
260 self._test_push_results_to_db_missing_env('BUILD_TAG')
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,
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,
276 self.result, self.details))
277 mock_logger_error.assert_called_once_with(test_utils.
278 RegexMatch("Pushing "
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,
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,
298 self.result, self.details))
299 self.assertTrue(mock_logger_error.called)
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,
307 mock.patch('functest.utils.functest_utils.requests.post'):
308 self.assertTrue(functest_utils.
309 push_results_to_db(self.project, self.case_name,
312 self.result, self.details))
314 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
318 if FunctestUtilsTesting.readline == \
319 len(FunctestUtilsTesting.test_ip) - 1:
321 FunctestUtilsTesting.readline += 1
322 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
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)
331 attrs = {'readline.side_effect': self.readline_side}
332 m.configure_mock(**attrs)
334 with mock.patch("six.moves.builtins.open") as mo:
336 self.assertEqual(functest_utils.get_resolvconf_ns(),
339 def _get_environ(self, var):
340 if var == 'INSTALLER_TYPE':
341 return self.installer
342 elif var == 'DEPLOY_SCENARIO':
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)
353 def cmd_readline(self):
354 return 'test_value\n'
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,
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:
365 FunctestUtilsTesting.readline = 0
367 mock_obj = mock.Mock()
368 attrs = {'readline.side_effect': self.cmd_readline()}
369 mock_obj.configure_mock(**attrs)
371 mock_obj2 = mock.Mock()
372 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
373 mock_obj2.configure_mock(**attrs)
375 mock_subproc_open.return_value = mock_obj2
377 resp = functest_utils.execute_command(self.cmd, info=True,
378 error_msg=self.error_msg,
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)
387 @mock.patch('functest.utils.functest_utils.logger.info')
388 def test_execute_command_args_present_with_success(self, mock_logger_info,
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:
395 FunctestUtilsTesting.readline = 0
397 mock_obj = mock.Mock()
398 attrs = {'readline.side_effect': self.cmd_readline()}
399 mock_obj.configure_mock(**attrs)
401 mock_obj2 = mock.Mock()
402 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
403 mock_obj2.configure_mock(**attrs)
405 mock_subproc_open.return_value = mock_obj2
407 resp = functest_utils.execute_command(self.cmd, info=True,
408 error_msg=self.error_msg,
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")
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:
421 FunctestUtilsTesting.readline = 2
423 mock_obj = mock.Mock()
424 attrs = {'readline.side_effect': self.cmd_readline()}
425 mock_obj.configure_mock(**attrs)
427 mock_obj2 = mock.Mock()
428 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
429 mock_obj2.configure_mock(**attrs)
431 mock_subproc_open.return_value = mock_obj2
433 resp = functest_utils.execute_command(self.cmd, info=False,
437 self.assertEqual(resp, 0)
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:
444 FunctestUtilsTesting.readline = 2
445 mock_obj = mock.Mock()
446 attrs = {'readline.side_effect': self.cmd_readline()}
447 mock_obj.configure_mock(**attrs)
449 mock_obj2 = mock.Mock()
450 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
451 mock_obj2.configure_mock(**attrs)
453 mock_subproc_open.return_value = mock_obj2
455 resp = functest_utils.execute_command(self.cmd, info=False,
459 self.assertEqual(resp, 1)
461 def _get_functest_config(self, var):
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') \
469 mock_obj = mock.Mock()
470 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
471 mock_obj.configure_mock(**attrs)
473 mock_yaml.return_value = mock_obj
475 self.assertDictEqual(functest_utils.
476 get_dict_by_test(self.testname),
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),
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))
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') \
495 self.assertRaises(ValueError) as excep:
496 mock_yaml.return_value = self.file_yaml
497 functest_utils.get_parameter_from_yaml(self.parameter,
499 self.assertTrue(("The parameter %s is not"
500 " defined in config_functest.yaml" %
501 self.parameter) in excep.exception)
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') \
507 mock_yaml.return_value = self.file_yaml
508 self.assertEqual(functest_utils.
509 get_parameter_from_yaml(self.parameter,
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,
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') \
526 mock_yaml.return_value = self.file_yaml
527 resp = functest_utils.get_functest_yaml()
528 self.assertEqual(resp, self.file_yaml)
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 "=======================")
537 if __name__ == "__main__":
538 logging.disable(logging.CRITICAL)
539 unittest.main(verbosity=2)