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
16 from git.exc import NoSuchPathError
20 from functest.tests.unit import test_utils
21 from functest.utils import functest_utils
24 class FunctestUtilsTesting(unittest.TestCase):
26 logging.disable(logging.CRITICAL)
29 self.url = 'http://www.opnfv.org/'
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.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'
43 self.success_rate = 2.0
44 self.criteria = 'test_criteria==2.0'
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 = {'name': 'testname', 'criteria': self.criteria}
57 self.parameter = 'general.openstack.image_name'
58 self.config_yaml = 'test_config_yaml-'
59 self.file_yaml = {'general': {'openstack': {'image_name':
62 @mock.patch('urllib2.urlopen',
63 side_effect=urllib2.URLError('no host given'))
64 def test_check_internet_connectivity_failed(self, mock_method):
65 self.assertFalse(functest_utils.check_internet_connectivity())
66 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
68 @mock.patch('urllib2.urlopen')
69 def test_check_internet_connectivity_default(self, mock_method):
70 self.assertTrue(functest_utils.check_internet_connectivity())
71 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
73 @mock.patch('urllib2.urlopen')
74 def test_check_internet_connectivity_debian(self, mock_method):
75 self.url = "https://www.debian.org/"
76 self.assertTrue(functest_utils.check_internet_connectivity(self.url))
77 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
79 @mock.patch('urllib2.urlopen',
80 side_effect=urllib2.URLError('no host given'))
81 def test_download_url_failed(self, mock_url):
82 self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
84 @mock.patch('urllib2.urlopen')
85 def test_download_url_default(self, mock_url):
86 with mock.patch("__builtin__.open", mock.mock_open()) as m, \
87 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
89 name = self.url.rsplit('/')[-1]
90 dest = self.dest_path + "/" + name
91 self.assertTrue(functest_utils.download_url(self.url,
93 m.assert_called_once_with(dest, 'wb')
94 self.assertTrue(mock_sh.called)
96 def test_get_git_branch(self):
97 with mock.patch('functest.utils.functest_utils.Repo') as mock_repo:
98 mock_obj2 = mock.Mock()
99 attrs = {'name': 'test_branch'}
100 mock_obj2.configure_mock(**attrs)
102 mock_obj = mock.Mock()
103 attrs = {'active_branch': mock_obj2}
104 mock_obj.configure_mock(**attrs)
106 mock_repo.return_value = mock_obj
107 self.assertEqual(functest_utils.get_git_branch(self.repo_path),
110 @mock.patch('functest.utils.functest_utils.Repo',
111 side_effect=NoSuchPathError)
112 def test_get_git_branch_failed(self, mock_repo):
113 self.assertRaises(NoSuchPathError,
114 lambda: functest_utils.get_git_branch(self.repo_path
117 @mock.patch('functest.utils.functest_utils.logger.error')
118 def test_get_installer_type_failed(self, mock_logger_error):
119 with mock.patch.dict(os.environ,
122 self.assertEqual(functest_utils.get_installer_type(),
124 mock_logger_error.assert_called_once_with("Impossible to retrieve"
125 " the installer type")
127 def test_get_installer_type_default(self):
128 with mock.patch.dict(os.environ,
129 {'INSTALLER_TYPE': 'test_installer'},
131 self.assertEqual(functest_utils.get_installer_type(),
134 @mock.patch('functest.utils.functest_utils.logger.error')
135 def test_get_scenario_failed(self, mock_logger_error):
136 with mock.patch.dict(os.environ,
139 self.assertEqual(functest_utils.get_scenario(),
141 mock_logger_error.assert_called_once_with("Impossible to retrieve"
144 def test_get_scenario_default(self):
145 with mock.patch.dict(os.environ,
146 {'DEPLOY_SCENARIO': 'test_scenario'},
148 self.assertEqual(functest_utils.get_scenario(),
151 @mock.patch('functest.utils.functest_utils.get_build_tag')
152 def test_get_version_default(self, mock_get_build_tag):
153 mock_get_build_tag.return_value = self.build_tag
154 self.assertEqual(functest_utils.get_version(), self.version)
156 @mock.patch('functest.utils.functest_utils.get_build_tag')
157 def test_get_version_unknown(self, mock_get_build_tag):
158 mock_get_build_tag.return_value = "unknown_build_tag"
159 self.assertEqual(functest_utils.get_version(), "unknown")
161 @mock.patch('functest.utils.functest_utils.logger.error')
162 def test_get_pod_name_failed(self, mock_logger_error):
163 with mock.patch.dict(os.environ,
166 self.assertEqual(functest_utils.get_pod_name(),
168 mock_logger_error.assert_called_once_with("Unable to retrieve "
170 "environment. Using "
171 "pod name 'unknown-pod'")
173 def test_get_pod_name_default(self):
174 with mock.patch.dict(os.environ,
175 {'NODE_NAME': 'test_node_name'},
177 self.assertEqual(functest_utils.get_pod_name(),
180 @mock.patch('functest.utils.functest_utils.logger.error')
181 def test_get_build_tag_failed(self, mock_logger_error):
182 with mock.patch.dict(os.environ,
185 self.assertEqual(functest_utils.get_build_tag(),
187 mock_logger_error.assert_called_once_with("Impossible to retrieve"
190 def test_get_build_tag_default(self):
191 with mock.patch.dict(os.environ,
192 {'BUILD_TAG': self.build_tag},
194 self.assertEqual(functest_utils.get_build_tag(),
197 @mock.patch('functest.utils.functest_utils.get_functest_config')
198 def test_get_db_url(self, mock_get_functest_config):
199 mock_get_functest_config.return_value = self.db_url
200 self.assertEqual(functest_utils.get_db_url(), self.db_url)
201 mock_get_functest_config.assert_called_once_with('results.test_db_url')
203 @mock.patch('functest.utils.functest_utils.logger.info')
204 def test_logger_test_results(self, mock_logger_info):
205 with mock.patch('functest.utils.functest_utils.get_pod_name',
206 return_value=self.node_name), \
207 mock.patch('functest.utils.functest_utils.get_scenario',
208 return_value=self.scenario), \
209 mock.patch('functest.utils.functest_utils.get_version',
210 return_value=self.version), \
211 mock.patch('functest.utils.functest_utils.get_build_tag',
212 return_value=self.build_tag), \
213 mock.patch('functest.utils.functest_utils.get_db_url',
214 return_value=self.db_url):
215 functest_utils.logger_test_results(self.project, self.case_name,
216 self.status, self.details)
217 mock_logger_info.assert_called_once_with(
219 "****************************************\n"
220 "\t %(p)s/%(n)s results \n\n"
221 "****************************************\n"
227 "build tag:\t%(b)s\n"
229 % {'p': self.project,
232 'pod': self.node_name,
239 def _get_env_dict(self, var):
240 dic = {'INSTALLER_TYPE': self.installer,
241 'DEPLOY_SCENARIO': self.scenario,
242 'NODE_NAME': self.node_name,
243 'BUILD_TAG': self.build_tag}
247 def _test_push_results_to_db_missing_env(self, env_var):
248 dic = self._get_env_dict(env_var)
249 with mock.patch('functest.utils.functest_utils.get_db_url',
250 return_value=self.db_url), \
251 mock.patch.dict(os.environ,
254 mock.patch('functest.utils.functest_utils.logger.error') \
255 as mock_logger_error:
256 functest_utils.push_results_to_db(self.project, self.case_name,
257 self.start_date, self.stop_date,
258 self.criteria, self.details)
259 mock_logger_error.assert_called_once_with("Please set env var: " +
263 def test_push_results_to_db_missing_installer(self):
264 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
266 def test_push_results_to_db_missing_scenario(self):
267 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
269 def test_push_results_to_db_missing_nodename(self):
270 self._test_push_results_to_db_missing_env('NODE_NAME')
272 def test_push_results_to_db_missing_buildtag(self):
273 self._test_push_results_to_db_missing_env('BUILD_TAG')
275 def test_push_results_to_db_incorrect_buildtag(self):
276 dic = self._get_env_dict(None)
277 dic['BUILD_TAG'] = 'incorrect_build_tag'
278 with mock.patch('functest.utils.functest_utils.get_db_url',
279 return_value=self.db_url), \
280 mock.patch.dict(os.environ,
283 mock.patch('functest.utils.functest_utils.logger.error') \
284 as mock_logger_error:
285 self.assertFalse(functest_utils.
286 push_results_to_db(self.project, self.case_name,
289 self.criteria, self.details))
290 mock_logger_error.assert_called_once_with("Please fix BUILD_TAG"
291 " env var: incorrect_"
294 def test_push_results_to_db_request_post_failed(self):
295 dic = self._get_env_dict(None)
296 with mock.patch('functest.utils.functest_utils.get_db_url',
297 return_value=self.db_url), \
298 mock.patch.dict(os.environ,
301 mock.patch('functest.utils.functest_utils.logger.error') \
302 as mock_logger_error, \
303 mock.patch('functest.utils.functest_utils.requests.post',
304 side_effect=requests.RequestException):
305 self.assertFalse(functest_utils.
306 push_results_to_db(self.project, self.case_name,
309 self.criteria, self.details))
310 mock_logger_error.assert_called_once_with(test_utils.
311 RegexMatch("Pushing "
317 def test_push_results_to_db_request_post_exception(self):
318 dic = self._get_env_dict(None)
319 with mock.patch('functest.utils.functest_utils.get_db_url',
320 return_value=self.db_url), \
321 mock.patch.dict(os.environ,
324 mock.patch('functest.utils.functest_utils.logger.error') \
325 as mock_logger_error, \
326 mock.patch('functest.utils.functest_utils.requests.post',
327 side_effect=Exception):
328 self.assertFalse(functest_utils.
329 push_results_to_db(self.project, self.case_name,
332 self.criteria, self.details))
333 self.assertTrue(mock_logger_error.called)
335 def test_push_results_to_db_default(self):
336 dic = self._get_env_dict(None)
337 with mock.patch('functest.utils.functest_utils.get_db_url',
338 return_value=self.db_url), \
339 mock.patch.dict(os.environ,
342 mock.patch('functest.utils.functest_utils.requests.post'):
343 self.assertTrue(functest_utils.
344 push_results_to_db(self.project, self.case_name,
347 self.criteria, self.details))
349 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
353 if FunctestUtilsTesting.readline == \
354 len(FunctestUtilsTesting.test_ip) - 1:
356 FunctestUtilsTesting.readline += 1
357 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
359 # TODO: get_resolvconf_ns
360 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
361 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
362 attrs = {'query.return_value': ["test"]}
363 mock_dns_resolve.configure_mock(**attrs)
366 attrs = {'readline.side_effect': self.readline_side}
367 m.configure_mock(**attrs)
369 with mock.patch("__builtin__.open") as mo:
371 self.assertEqual(functest_utils.get_resolvconf_ns(),
374 def _get_environ(self, var):
375 if var == 'INSTALLER_TYPE':
376 return self.installer
377 elif var == 'DEPLOY_SCENARIO':
381 def test_get_ci_envvars_default(self):
382 with mock.patch('os.environ.get',
383 side_effect=self._get_environ):
384 dic = {"installer": self.installer,
385 "scenario": self.scenario}
386 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
388 def cmd_readline(self):
389 return 'test_value\n'
391 @mock.patch('functest.utils.functest_utils.logger.error')
392 @mock.patch('functest.utils.functest_utils.logger.info')
393 def test_execute_command_args_present_with_error(self, mock_logger_info,
395 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
396 as mock_subproc_open, \
397 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
399 FunctestUtilsTesting.readline = 0
401 mock_obj = mock.Mock()
402 attrs = {'readline.side_effect': self.cmd_readline()}
403 mock_obj.configure_mock(**attrs)
405 mock_obj2 = mock.Mock()
406 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
407 mock_obj2.configure_mock(**attrs)
409 mock_subproc_open.return_value = mock_obj2
411 resp = functest_utils.execute_command(self.cmd, info=True,
412 error_msg=self.error_msg,
414 output_file=self.output_file)
415 self.assertEqual(resp, 1)
416 msg_exec = ("Executing command: '%s'" % self.cmd)
417 mock_logger_info.assert_called_once_with(msg_exec)
418 mopen.assert_called_once_with(self.output_file, "w")
419 mock_logger_error.assert_called_once_with(self.error_msg)
421 @mock.patch('functest.utils.functest_utils.logger.info')
422 def test_execute_command_args_present_with_success(self, mock_logger_info,
424 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
425 as mock_subproc_open, \
426 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
428 FunctestUtilsTesting.readline = 0
430 mock_obj = mock.Mock()
431 attrs = {'readline.side_effect': self.cmd_readline()}
432 mock_obj.configure_mock(**attrs)
434 mock_obj2 = mock.Mock()
435 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
436 mock_obj2.configure_mock(**attrs)
438 mock_subproc_open.return_value = mock_obj2
440 resp = functest_utils.execute_command(self.cmd, info=True,
441 error_msg=self.error_msg,
443 output_file=self.output_file)
444 self.assertEqual(resp, 0)
445 msg_exec = ("Executing command: '%s'" % self.cmd)
446 mock_logger_info.assert_called_once_with(msg_exec)
447 mopen.assert_called_once_with(self.output_file, "w")
449 @mock.patch('functest.utils.functest_utils.logger.info')
450 def test_execute_command_args_missing_with_success(self, mock_logger_info,
452 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
453 as mock_subproc_open:
455 FunctestUtilsTesting.readline = 2
457 mock_obj = mock.Mock()
458 attrs = {'readline.side_effect': self.cmd_readline()}
459 mock_obj.configure_mock(**attrs)
461 mock_obj2 = mock.Mock()
462 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
463 mock_obj2.configure_mock(**attrs)
465 mock_subproc_open.return_value = mock_obj2
467 resp = functest_utils.execute_command(self.cmd, info=False,
471 self.assertEqual(resp, 0)
473 @mock.patch('functest.utils.functest_utils.logger.error')
474 def test_execute_command_args_missing_with_error(self, mock_logger_error,
476 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
477 as mock_subproc_open:
479 FunctestUtilsTesting.readline = 2
480 mock_obj = mock.Mock()
481 attrs = {'readline.side_effect': self.cmd_readline()}
482 mock_obj.configure_mock(**attrs)
484 mock_obj2 = mock.Mock()
485 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
486 mock_obj2.configure_mock(**attrs)
488 mock_subproc_open.return_value = mock_obj2
490 resp = functest_utils.execute_command(self.cmd, info=False,
494 self.assertEqual(resp, 1)
496 def _get_functest_config(self, var):
499 @mock.patch('functest.utils.functest_utils.logger.error')
500 def test_get_dict_by_test(self, mock_logger_error):
501 with mock.patch('__builtin__.open', mock.mock_open()), \
502 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
504 mock.patch('functest.utils.functest_utils.get_testcases_'
506 mock_obj = mock.Mock()
507 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
508 mock_obj.configure_mock(**attrs)
510 mock_yaml.return_value = mock_obj
512 self.assertDictEqual(functest_utils.
513 get_dict_by_test(self.testname),
516 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
517 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
518 mock_get_dict_by_test.return_value = self.testcase_dict
519 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
522 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
523 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
524 mock_get_dict_by_test.return_value = None
525 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
527 def test_get_parameter_from_yaml_failed(self):
528 self.file_yaml['general'] = None
529 with mock.patch('__builtin__.open', mock.mock_open()), \
530 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
532 self.assertRaises(ValueError) as excep:
533 mock_yaml.return_value = self.file_yaml
534 functest_utils.get_parameter_from_yaml(self.parameter,
536 self.assertTrue(("The parameter %s is not"
537 " defined in config_functest.yaml" %
538 self.parameter) in excep.exception)
540 def test_get_parameter_from_yaml_default(self):
541 with mock.patch('__builtin__.open', mock.mock_open()), \
542 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
544 mock_yaml.return_value = self.file_yaml
545 self.assertEqual(functest_utils.
546 get_parameter_from_yaml(self.parameter,
550 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
551 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
552 with mock.patch.dict(os.environ,
553 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
554 functest_utils.get_functest_config(self.parameter)
555 mock_get_parameter_from_yaml. \
556 assert_called_once_with(self.parameter,
559 def test_check_success_rate_default(self):
560 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
562 mock_criteria.return_value = self.criteria
563 resp = functest_utils.check_success_rate(self.case_name,
565 self.assertEqual(resp, 'PASS')
567 def test_check_success_rate_failed(self):
568 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
570 mock_criteria.return_value = self.criteria
571 resp = functest_utils.check_success_rate(self.case_name,
573 self.assertEqual(resp, 'FAIL')
577 def test_get_testcases_file_dir(self):
578 resp = functest_utils.get_testcases_file_dir()
579 self.assertEqual(resp,
580 "/home/opnfv/repos/functest/"
581 "functest/ci/testcases.yaml")
583 def test_get_functest_yaml(self):
584 with mock.patch('__builtin__.open', mock.mock_open()), \
585 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
587 mock_yaml.return_value = self.file_yaml
588 resp = functest_utils.get_functest_yaml()
589 self.assertEqual(resp, self.file_yaml)
591 @mock.patch('functest.utils.functest_utils.logger.info')
592 def test_print_separator(self, mock_logger_info):
593 functest_utils.print_separator()
594 mock_logger_info.assert_called_once_with("======================="
595 "=======================")
598 if __name__ == "__main__":
599 unittest.main(verbosity=2)