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 mock.patch('logging.FileHandler').start() # noqa
22 from functest.utils import functest_utils
25 class FunctestUtilsTesting(unittest.TestCase):
27 logging.disable(logging.CRITICAL)
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.version = 'master'
38 self.node_name = 'test_node_name'
39 self.project = 'test_project'
40 self.case_name = 'test_case_name'
41 self.status = 'test_status'
42 self.details = 'test_details'
43 self.db_url = 'test_db_url'
44 self.success_rate = 2.0
45 self.criteria = 'test_criteria==2.0'
46 self.start_date = 1482624000
47 self.stop_date = 1482624000
48 self.start_time = time.time()
49 self.stop_time = time.time()
51 self.test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
52 self.test_file = 'test_file'
53 self.error_msg = 'test_error_msg'
55 self.output_file = 'test_output_file'
56 self.testname = 'testname'
57 self.testcase_dict = {'name': 'testname', 'criteria': self.criteria}
58 self.parameter = 'general.openstack.image_name'
59 self.config_yaml = 'test_config_yaml-'
60 self.file_yaml = {'general': {'openstack': {'image_name':
63 @mock.patch('urllib2.urlopen',
64 side_effect=urllib2.URLError('no host given'))
65 def test_check_internet_connectivity_failed(self, mock_method):
66 self.assertFalse(functest_utils.check_internet_connectivity())
67 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
69 @mock.patch('urllib2.urlopen')
70 def test_check_internet_connectivity_default(self, mock_method):
71 self.assertTrue(functest_utils.check_internet_connectivity())
72 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
74 @mock.patch('urllib2.urlopen')
75 def test_check_internet_connectivity_debian(self, mock_method):
76 self.url = "https://www.debian.org/"
77 self.assertTrue(functest_utils.check_internet_connectivity(self.url))
78 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
80 @mock.patch('urllib2.urlopen',
81 side_effect=urllib2.URLError('no host given'))
82 def test_download_url_failed(self, mock_url):
83 self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
85 @mock.patch('urllib2.urlopen')
86 def test_download_url_default(self, mock_url):
87 with mock.patch("__builtin__.open", mock.mock_open()) as m, \
88 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
90 name = self.url.rsplit('/')[-1]
91 dest = self.dest_path + "/" + name
92 self.assertTrue(functest_utils.download_url(self.url,
94 m.assert_called_once_with(dest, 'wb')
95 self.assertTrue(mock_sh.called)
97 def test_get_git_branch(self):
98 with mock.patch('functest.utils.functest_utils.Repo') as mock_repo:
99 mock_obj2 = mock.Mock()
100 attrs = {'name': 'test_branch'}
101 mock_obj2.configure_mock(**attrs)
103 mock_obj = mock.Mock()
104 attrs = {'active_branch': mock_obj2}
105 mock_obj.configure_mock(**attrs)
107 mock_repo.return_value = mock_obj
108 self.assertEqual(functest_utils.get_git_branch(self.repo_path),
111 @mock.patch('functest.utils.functest_utils.Repo',
112 side_effect=NoSuchPathError)
113 def test_get_git_branch_failed(self, mock_repo):
114 self.assertRaises(NoSuchPathError,
115 lambda: functest_utils.get_git_branch(self.repo_path
118 @mock.patch('functest.utils.functest_utils.logger.error')
119 def test_get_installer_type_failed(self, mock_logger_error):
120 with mock.patch.dict(os.environ,
123 self.assertEqual(functest_utils.get_installer_type(),
125 mock_logger_error.assert_called_once_with("Impossible to retrieve"
126 " the installer type")
128 def test_get_installer_type_default(self):
129 with mock.patch.dict(os.environ,
130 {'INSTALLER_TYPE': 'test_installer'},
132 self.assertEqual(functest_utils.get_installer_type(),
135 @mock.patch('functest.utils.functest_utils.logger.error')
136 def test_get_scenario_failed(self, mock_logger_error):
137 with mock.patch.dict(os.environ,
140 self.assertEqual(functest_utils.get_scenario(),
142 mock_logger_error.assert_called_once_with("Impossible to retrieve"
145 def test_get_scenario_default(self):
146 with mock.patch.dict(os.environ,
147 {'DEPLOY_SCENARIO': 'test_scenario'},
149 self.assertEqual(functest_utils.get_scenario(),
152 @mock.patch('functest.utils.functest_utils.get_build_tag')
153 def test_get_version_default(self, mock_get_build_tag):
154 mock_get_build_tag.return_value = self.build_tag
155 self.assertEqual(functest_utils.get_version(), self.version)
157 @mock.patch('functest.utils.functest_utils.get_build_tag')
158 def test_get_version_unknown(self, mock_get_build_tag):
159 mock_get_build_tag.return_value = "unknown_build_tag"
160 self.assertEqual(functest_utils.get_version(), "unknown")
162 @mock.patch('functest.utils.functest_utils.logger.error')
163 def test_get_pod_name_failed(self, mock_logger_error):
164 with mock.patch.dict(os.environ,
167 self.assertEqual(functest_utils.get_pod_name(),
169 mock_logger_error.assert_called_once_with("Unable to retrieve "
171 "environment. Using "
172 "pod name 'unknown-pod'")
174 def test_get_pod_name_default(self):
175 with mock.patch.dict(os.environ,
176 {'NODE_NAME': 'test_node_name'},
178 self.assertEqual(functest_utils.get_pod_name(),
181 @mock.patch('functest.utils.functest_utils.logger.error')
182 def test_get_build_tag_failed(self, mock_logger_error):
183 with mock.patch.dict(os.environ,
186 self.assertEqual(functest_utils.get_build_tag(),
188 mock_logger_error.assert_called_once_with("Impossible to retrieve"
191 def test_get_build_tag_default(self):
192 with mock.patch.dict(os.environ,
193 {'BUILD_TAG': self.build_tag},
195 self.assertEqual(functest_utils.get_build_tag(),
198 @mock.patch('functest.utils.functest_utils.get_functest_config')
199 def test_get_db_url(self, mock_get_functest_config):
200 mock_get_functest_config.return_value = self.db_url
201 self.assertEqual(functest_utils.get_db_url(), self.db_url)
202 mock_get_functest_config.assert_called_once_with('results.test_db_url')
204 @mock.patch('functest.utils.functest_utils.logger.info')
205 def test_logger_test_results(self, mock_logger_info):
206 with mock.patch('functest.utils.functest_utils.get_pod_name',
207 return_value=self.node_name), \
208 mock.patch('functest.utils.functest_utils.get_scenario',
209 return_value=self.scenario), \
210 mock.patch('functest.utils.functest_utils.get_version',
211 return_value=self.version), \
212 mock.patch('functest.utils.functest_utils.get_build_tag',
213 return_value=self.build_tag), \
214 mock.patch('functest.utils.functest_utils.get_db_url',
215 return_value=self.db_url):
216 functest_utils.logger_test_results(self.project, self.case_name,
217 self.status, self.details)
218 mock_logger_info.assert_called_once_with(
220 "****************************************\n"
221 "\t %(p)s/%(n)s results \n\n"
222 "****************************************\n"
228 "build tag:\t%(b)s\n"
230 % {'p': self.project,
233 'pod': self.node_name,
240 def _get_env_dict(self, var):
241 dic = {'INSTALLER_TYPE': self.installer,
242 'DEPLOY_SCENARIO': self.scenario,
243 'NODE_NAME': self.node_name,
244 'BUILD_TAG': self.build_tag}
248 def _test_push_results_to_db_missing_env(self, env_var):
249 dic = self._get_env_dict(env_var)
250 with mock.patch('functest.utils.functest_utils.get_db_url',
251 return_value=self.db_url), \
252 mock.patch.dict(os.environ,
255 mock.patch('functest.utils.functest_utils.logger.error') \
256 as mock_logger_error:
257 functest_utils.push_results_to_db(self.project, self.case_name,
258 self.start_date, self.stop_date,
259 self.criteria, self.details)
260 mock_logger_error.assert_called_once_with("Please set env var: " +
264 def test_push_results_to_db_missing_installer(self):
265 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
267 def test_push_results_to_db_missing_scenario(self):
268 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
270 def test_push_results_to_db_missing_nodename(self):
271 self._test_push_results_to_db_missing_env('NODE_NAME')
273 def test_push_results_to_db_missing_buildtag(self):
274 self._test_push_results_to_db_missing_env('BUILD_TAG')
276 def test_push_results_to_db_incorrect_buildtag(self):
277 dic = self._get_env_dict(None)
278 dic['BUILD_TAG'] = 'incorrect_build_tag'
279 with mock.patch('functest.utils.functest_utils.get_db_url',
280 return_value=self.db_url), \
281 mock.patch.dict(os.environ,
284 mock.patch('functest.utils.functest_utils.logger.error') \
285 as mock_logger_error:
286 self.assertFalse(functest_utils.
287 push_results_to_db(self.project, self.case_name,
290 self.criteria, self.details))
291 mock_logger_error.assert_called_once_with("Please fix BUILD_TAG"
292 " env var: incorrect_"
295 def test_push_results_to_db_request_post_failed(self):
296 dic = self._get_env_dict(None)
297 with mock.patch('functest.utils.functest_utils.get_db_url',
298 return_value=self.db_url), \
299 mock.patch.dict(os.environ,
302 mock.patch('functest.utils.functest_utils.logger.error') \
303 as mock_logger_error, \
304 mock.patch('functest.utils.functest_utils.requests.post',
305 side_effect=requests.RequestException):
306 self.assertFalse(functest_utils.
307 push_results_to_db(self.project, self.case_name,
310 self.criteria, self.details))
311 mock_logger_error.assert_called_once_with(test_utils.
312 RegexMatch("Pushing "
318 def test_push_results_to_db_request_post_exception(self):
319 dic = self._get_env_dict(None)
320 with mock.patch('functest.utils.functest_utils.get_db_url',
321 return_value=self.db_url), \
322 mock.patch.dict(os.environ,
325 mock.patch('functest.utils.functest_utils.logger.error') \
326 as mock_logger_error, \
327 mock.patch('functest.utils.functest_utils.requests.post',
328 side_effect=Exception):
329 self.assertFalse(functest_utils.
330 push_results_to_db(self.project, self.case_name,
333 self.criteria, self.details))
334 self.assertTrue(mock_logger_error.called)
336 def test_push_results_to_db_default(self):
337 dic = self._get_env_dict(None)
338 with mock.patch('functest.utils.functest_utils.get_db_url',
339 return_value=self.db_url), \
340 mock.patch.dict(os.environ,
343 mock.patch('functest.utils.functest_utils.requests.post'):
344 self.assertTrue(functest_utils.
345 push_results_to_db(self.project, self.case_name,
348 self.criteria, self.details))
350 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
354 if FunctestUtilsTesting.readline == \
355 len(FunctestUtilsTesting.test_ip) - 1:
357 FunctestUtilsTesting.readline += 1
358 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
360 # TODO: get_resolvconf_ns
361 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
362 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
363 attrs = {'query.return_value': ["test"]}
364 mock_dns_resolve.configure_mock(**attrs)
367 attrs = {'readline.side_effect': self.readline_side}
368 m.configure_mock(**attrs)
370 with mock.patch("__builtin__.open") as mo:
372 self.assertEqual(functest_utils.get_resolvconf_ns(),
375 def _get_environ(self, var):
376 if var == 'INSTALLER_TYPE':
377 return self.installer
378 elif var == 'DEPLOY_SCENARIO':
382 def test_get_ci_envvars_default(self):
383 with mock.patch('os.environ.get',
384 side_effect=self._get_environ):
385 dic = {"installer": self.installer,
386 "scenario": self.scenario}
387 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
389 def cmd_readline(self):
390 return 'test_value\n'
392 @mock.patch('functest.utils.functest_utils.logger.error')
393 @mock.patch('functest.utils.functest_utils.logger.info')
394 def test_execute_command_args_present_with_error(self, mock_logger_info,
396 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
397 as mock_subproc_open, \
398 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
400 FunctestUtilsTesting.readline = 0
402 mock_obj = mock.Mock()
403 attrs = {'readline.side_effect': self.cmd_readline()}
404 mock_obj.configure_mock(**attrs)
406 mock_obj2 = mock.Mock()
407 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
408 mock_obj2.configure_mock(**attrs)
410 mock_subproc_open.return_value = mock_obj2
412 resp = functest_utils.execute_command(self.cmd, info=True,
413 error_msg=self.error_msg,
415 output_file=self.output_file)
416 self.assertEqual(resp, 1)
417 msg_exec = ("Executing command: '%s'" % self.cmd)
418 mock_logger_info.assert_called_once_with(msg_exec)
419 mopen.assert_called_once_with(self.output_file, "w")
420 mock_logger_error.assert_called_once_with(self.error_msg)
422 @mock.patch('functest.utils.functest_utils.logger.info')
423 def test_execute_command_args_present_with_success(self, mock_logger_info,
425 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
426 as mock_subproc_open, \
427 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
429 FunctestUtilsTesting.readline = 0
431 mock_obj = mock.Mock()
432 attrs = {'readline.side_effect': self.cmd_readline()}
433 mock_obj.configure_mock(**attrs)
435 mock_obj2 = mock.Mock()
436 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
437 mock_obj2.configure_mock(**attrs)
439 mock_subproc_open.return_value = mock_obj2
441 resp = functest_utils.execute_command(self.cmd, info=True,
442 error_msg=self.error_msg,
444 output_file=self.output_file)
445 self.assertEqual(resp, 0)
446 msg_exec = ("Executing command: '%s'" % self.cmd)
447 mock_logger_info.assert_called_once_with(msg_exec)
448 mopen.assert_called_once_with(self.output_file, "w")
450 @mock.patch('functest.utils.functest_utils.logger.info')
451 def test_execute_command_args_missing_with_success(self, mock_logger_info,
453 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
454 as mock_subproc_open:
456 FunctestUtilsTesting.readline = 2
458 mock_obj = mock.Mock()
459 attrs = {'readline.side_effect': self.cmd_readline()}
460 mock_obj.configure_mock(**attrs)
462 mock_obj2 = mock.Mock()
463 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
464 mock_obj2.configure_mock(**attrs)
466 mock_subproc_open.return_value = mock_obj2
468 resp = functest_utils.execute_command(self.cmd, info=False,
472 self.assertEqual(resp, 0)
474 @mock.patch('functest.utils.functest_utils.logger.error')
475 def test_execute_command_args_missing_with_error(self, mock_logger_error,
477 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
478 as mock_subproc_open:
480 FunctestUtilsTesting.readline = 2
481 mock_obj = mock.Mock()
482 attrs = {'readline.side_effect': self.cmd_readline()}
483 mock_obj.configure_mock(**attrs)
485 mock_obj2 = mock.Mock()
486 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
487 mock_obj2.configure_mock(**attrs)
489 mock_subproc_open.return_value = mock_obj2
491 resp = functest_utils.execute_command(self.cmd, info=False,
495 self.assertEqual(resp, 1)
497 def _get_functest_config(self, var):
500 @mock.patch('functest.utils.functest_utils.logger.error')
501 def test_get_dict_by_test(self, mock_logger_error):
502 with mock.patch('__builtin__.open', mock.mock_open()), \
503 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
505 mock.patch('functest.utils.functest_utils.get_testcases_'
507 mock_obj = mock.Mock()
508 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
509 mock_obj.configure_mock(**attrs)
511 mock_yaml.return_value = mock_obj
513 self.assertDictEqual(functest_utils.
514 get_dict_by_test(self.testname),
517 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
518 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
519 mock_get_dict_by_test.return_value = self.testcase_dict
520 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
523 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
524 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
525 mock_get_dict_by_test.return_value = None
526 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
528 def test_get_parameter_from_yaml_failed(self):
529 self.file_yaml['general'] = None
530 with mock.patch('__builtin__.open', mock.mock_open()), \
531 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
533 self.assertRaises(ValueError) as excep:
534 mock_yaml.return_value = self.file_yaml
535 functest_utils.get_parameter_from_yaml(self.parameter,
537 self.assertTrue(("The parameter %s is not"
538 " defined in config_functest.yaml" %
539 self.parameter) in excep.exception)
541 def test_get_parameter_from_yaml_default(self):
542 with mock.patch('__builtin__.open', mock.mock_open()), \
543 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
545 mock_yaml.return_value = self.file_yaml
546 self.assertEqual(functest_utils.
547 get_parameter_from_yaml(self.parameter,
551 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
552 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
553 with mock.patch.dict(os.environ,
554 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
555 functest_utils.get_functest_config(self.parameter)
556 mock_get_parameter_from_yaml. \
557 assert_called_once_with(self.parameter,
560 def test_check_success_rate_default(self):
561 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
563 mock_criteria.return_value = self.criteria
564 resp = functest_utils.check_success_rate(self.case_name,
566 self.assertEqual(resp, 'PASS')
568 def test_check_success_rate_failed(self):
569 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
571 mock_criteria.return_value = self.criteria
572 resp = functest_utils.check_success_rate(self.case_name,
574 self.assertEqual(resp, 'FAIL')
578 def test_get_testcases_file_dir(self):
579 resp = functest_utils.get_testcases_file_dir()
580 self.assertEqual(resp,
581 "/home/opnfv/repos/functest/"
582 "functest/ci/testcases.yaml")
584 def test_get_functest_yaml(self):
585 with mock.patch('__builtin__.open', mock.mock_open()), \
586 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
588 mock_yaml.return_value = self.file_yaml
589 resp = functest_utils.get_functest_yaml()
590 self.assertEqual(resp, self.file_yaml)
592 @mock.patch('functest.utils.functest_utils.logger.info')
593 def test_print_separator(self, mock_logger_info):
594 functest_utils.print_separator()
595 mock_logger_info.assert_called_once_with("======================="
596 "=======================")
599 if __name__ == "__main__":
600 unittest.main(verbosity=2)