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.db_url_env = 'http://foo/testdb'
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.info')
136 def test_get_scenario_failed(self, mock_logger_info):
137 with mock.patch.dict(os.environ,
140 self.assertEqual(functest_utils.get_scenario(),
141 "os-nosdn-nofeature-noha")
142 mock_logger_info.assert_called_once_with("Impossible to retrieve "
145 "os-nosdn-nofeature-noha")
147 def test_get_scenario_default(self):
148 with mock.patch.dict(os.environ,
149 {'DEPLOY_SCENARIO': 'test_scenario'},
151 self.assertEqual(functest_utils.get_scenario(),
154 @mock.patch('functest.utils.functest_utils.get_build_tag')
155 def test_get_version_default(self, mock_get_build_tag):
156 mock_get_build_tag.return_value = self.build_tag
157 self.assertEqual(functest_utils.get_version(), self.version)
159 @mock.patch('functest.utils.functest_utils.get_build_tag')
160 def test_get_version_unknown(self, mock_get_build_tag):
161 mock_get_build_tag.return_value = "unknown_build_tag"
162 self.assertEqual(functest_utils.get_version(), "unknown")
164 @mock.patch('functest.utils.functest_utils.logger.info')
165 def test_get_pod_name_failed(self, mock_logger_info):
166 with mock.patch.dict(os.environ,
169 self.assertEqual(functest_utils.get_pod_name(),
171 mock_logger_info.assert_called_once_with("Unable to retrieve "
173 "environment. Using "
174 "pod name 'unknown-pod'")
176 def test_get_pod_name_default(self):
177 with mock.patch.dict(os.environ,
178 {'NODE_NAME': 'test_node_name'},
180 self.assertEqual(functest_utils.get_pod_name(),
183 @mock.patch('functest.utils.functest_utils.logger.info')
184 def test_get_build_tag_failed(self, mock_logger_info):
185 with mock.patch.dict(os.environ,
188 self.assertEqual(functest_utils.get_build_tag(),
190 mock_logger_info.assert_called_once_with("Impossible to retrieve"
193 def test_get_build_tag_default(self):
194 with mock.patch.dict(os.environ,
195 {'BUILD_TAG': self.build_tag},
197 self.assertEqual(functest_utils.get_build_tag(),
200 def test_get_db_url_env_var(self):
201 with mock.patch.dict(os.environ,
202 {'TEST_DB_URL': self.db_url_env,
203 'CONFIG_FUNCTEST_YAML':
204 "./functest/ci/config_functest.yaml"},
206 self.assertEqual(functest_utils.get_db_url(),
209 @mock.patch('functest.utils.functest_utils.get_functest_config')
210 def test_get_db_url_default(self, mock_get_functest_config):
211 mock_get_functest_config.return_value = self.db_url
212 self.assertEqual(functest_utils.get_db_url(), self.db_url)
213 mock_get_functest_config.assert_called_once_with('results.test_db_url')
215 @mock.patch('functest.utils.functest_utils.logger.info')
216 def test_logger_test_results(self, mock_logger_info):
217 with mock.patch('functest.utils.functest_utils.get_pod_name',
218 return_value=self.node_name), \
219 mock.patch('functest.utils.functest_utils.get_scenario',
220 return_value=self.scenario), \
221 mock.patch('functest.utils.functest_utils.get_version',
222 return_value=self.version), \
223 mock.patch('functest.utils.functest_utils.get_build_tag',
224 return_value=self.build_tag), \
225 mock.patch('functest.utils.functest_utils.get_db_url',
226 return_value=self.db_url):
227 functest_utils.logger_test_results(self.project, self.case_name,
228 self.status, self.details)
229 mock_logger_info.assert_called_once_with(
231 "****************************************\n"
232 "\t %(p)s/%(n)s results \n\n"
233 "****************************************\n"
239 "build tag:\t%(b)s\n"
241 % {'p': self.project,
244 'pod': self.node_name,
251 def _get_env_dict(self, var):
252 dic = {'INSTALLER_TYPE': self.installer,
253 'DEPLOY_SCENARIO': self.scenario,
254 'NODE_NAME': self.node_name,
255 'BUILD_TAG': self.build_tag}
259 def _test_push_results_to_db_missing_env(self, env_var):
260 dic = self._get_env_dict(env_var)
261 with mock.patch('functest.utils.functest_utils.get_db_url',
262 return_value=self.db_url), \
263 mock.patch.dict(os.environ,
266 mock.patch('functest.utils.functest_utils.logger.error') \
267 as mock_logger_error:
268 functest_utils.push_results_to_db(self.project, self.case_name,
269 self.start_date, self.stop_date,
270 self.criteria, self.details)
271 mock_logger_error.assert_called_once_with("Please set env var: " +
275 def test_push_results_to_db_missing_installer(self):
276 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
278 def test_push_results_to_db_missing_scenario(self):
279 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
281 def test_push_results_to_db_missing_nodename(self):
282 self._test_push_results_to_db_missing_env('NODE_NAME')
284 def test_push_results_to_db_missing_buildtag(self):
285 self._test_push_results_to_db_missing_env('BUILD_TAG')
287 def test_push_results_to_db_incorrect_buildtag(self):
288 dic = self._get_env_dict(None)
289 dic['BUILD_TAG'] = 'incorrect_build_tag'
290 with mock.patch('functest.utils.functest_utils.get_db_url',
291 return_value=self.db_url), \
292 mock.patch.dict(os.environ,
295 mock.patch('functest.utils.functest_utils.logger.error') \
296 as mock_logger_error:
297 self.assertFalse(functest_utils.
298 push_results_to_db(self.project, self.case_name,
301 self.criteria, self.details))
302 mock_logger_error.assert_called_once_with("Please fix BUILD_TAG"
303 " env var: incorrect_"
306 def test_push_results_to_db_request_post_failed(self):
307 dic = self._get_env_dict(None)
308 with mock.patch('functest.utils.functest_utils.get_db_url',
309 return_value=self.db_url), \
310 mock.patch.dict(os.environ,
313 mock.patch('functest.utils.functest_utils.logger.error') \
314 as mock_logger_error, \
315 mock.patch('functest.utils.functest_utils.requests.post',
316 side_effect=requests.RequestException):
317 self.assertFalse(functest_utils.
318 push_results_to_db(self.project, self.case_name,
321 self.criteria, self.details))
322 mock_logger_error.assert_called_once_with(test_utils.
323 RegexMatch("Pushing "
329 def test_push_results_to_db_request_post_exception(self):
330 dic = self._get_env_dict(None)
331 with mock.patch('functest.utils.functest_utils.get_db_url',
332 return_value=self.db_url), \
333 mock.patch.dict(os.environ,
336 mock.patch('functest.utils.functest_utils.logger.error') \
337 as mock_logger_error, \
338 mock.patch('functest.utils.functest_utils.requests.post',
339 side_effect=Exception):
340 self.assertFalse(functest_utils.
341 push_results_to_db(self.project, self.case_name,
344 self.criteria, self.details))
345 self.assertTrue(mock_logger_error.called)
347 def test_push_results_to_db_default(self):
348 dic = self._get_env_dict(None)
349 with mock.patch('functest.utils.functest_utils.get_db_url',
350 return_value=self.db_url), \
351 mock.patch.dict(os.environ,
354 mock.patch('functest.utils.functest_utils.requests.post'):
355 self.assertTrue(functest_utils.
356 push_results_to_db(self.project, self.case_name,
359 self.criteria, self.details))
361 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
365 if FunctestUtilsTesting.readline == \
366 len(FunctestUtilsTesting.test_ip) - 1:
368 FunctestUtilsTesting.readline += 1
369 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
371 # TODO: get_resolvconf_ns
372 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
373 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
374 attrs = {'query.return_value': ["test"]}
375 mock_dns_resolve.configure_mock(**attrs)
378 attrs = {'readline.side_effect': self.readline_side}
379 m.configure_mock(**attrs)
381 with mock.patch("__builtin__.open") as mo:
383 self.assertEqual(functest_utils.get_resolvconf_ns(),
386 def _get_environ(self, var):
387 if var == 'INSTALLER_TYPE':
388 return self.installer
389 elif var == 'DEPLOY_SCENARIO':
393 def test_get_ci_envvars_default(self):
394 with mock.patch('os.environ.get',
395 side_effect=self._get_environ):
396 dic = {"installer": self.installer,
397 "scenario": self.scenario}
398 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
400 def cmd_readline(self):
401 return 'test_value\n'
403 @mock.patch('functest.utils.functest_utils.logger.error')
404 @mock.patch('functest.utils.functest_utils.logger.info')
405 def test_execute_command_args_present_with_error(self, mock_logger_info,
407 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
408 as mock_subproc_open, \
409 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
411 FunctestUtilsTesting.readline = 0
413 mock_obj = mock.Mock()
414 attrs = {'readline.side_effect': self.cmd_readline()}
415 mock_obj.configure_mock(**attrs)
417 mock_obj2 = mock.Mock()
418 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
419 mock_obj2.configure_mock(**attrs)
421 mock_subproc_open.return_value = mock_obj2
423 resp = functest_utils.execute_command(self.cmd, info=True,
424 error_msg=self.error_msg,
426 output_file=self.output_file)
427 self.assertEqual(resp, 1)
428 msg_exec = ("Executing command: '%s'" % self.cmd)
429 mock_logger_info.assert_called_once_with(msg_exec)
430 mopen.assert_called_once_with(self.output_file, "w")
431 mock_logger_error.assert_called_once_with(self.error_msg)
433 @mock.patch('functest.utils.functest_utils.logger.info')
434 def test_execute_command_args_present_with_success(self, mock_logger_info,
436 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
437 as mock_subproc_open, \
438 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
440 FunctestUtilsTesting.readline = 0
442 mock_obj = mock.Mock()
443 attrs = {'readline.side_effect': self.cmd_readline()}
444 mock_obj.configure_mock(**attrs)
446 mock_obj2 = mock.Mock()
447 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
448 mock_obj2.configure_mock(**attrs)
450 mock_subproc_open.return_value = mock_obj2
452 resp = functest_utils.execute_command(self.cmd, info=True,
453 error_msg=self.error_msg,
455 output_file=self.output_file)
456 self.assertEqual(resp, 0)
457 msg_exec = ("Executing command: '%s'" % self.cmd)
458 mock_logger_info.assert_called_once_with(msg_exec)
459 mopen.assert_called_once_with(self.output_file, "w")
461 @mock.patch('functest.utils.functest_utils.logger.info')
462 def test_execute_command_args_missing_with_success(self, mock_logger_info,
464 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
465 as mock_subproc_open:
467 FunctestUtilsTesting.readline = 2
469 mock_obj = mock.Mock()
470 attrs = {'readline.side_effect': self.cmd_readline()}
471 mock_obj.configure_mock(**attrs)
473 mock_obj2 = mock.Mock()
474 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
475 mock_obj2.configure_mock(**attrs)
477 mock_subproc_open.return_value = mock_obj2
479 resp = functest_utils.execute_command(self.cmd, info=False,
483 self.assertEqual(resp, 0)
485 @mock.patch('functest.utils.functest_utils.logger.error')
486 def test_execute_command_args_missing_with_error(self, mock_logger_error,
488 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
489 as mock_subproc_open:
491 FunctestUtilsTesting.readline = 2
492 mock_obj = mock.Mock()
493 attrs = {'readline.side_effect': self.cmd_readline()}
494 mock_obj.configure_mock(**attrs)
496 mock_obj2 = mock.Mock()
497 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
498 mock_obj2.configure_mock(**attrs)
500 mock_subproc_open.return_value = mock_obj2
502 resp = functest_utils.execute_command(self.cmd, info=False,
506 self.assertEqual(resp, 1)
508 def _get_functest_config(self, var):
511 @mock.patch('functest.utils.functest_utils.logger.error')
512 def test_get_dict_by_test(self, mock_logger_error):
513 with mock.patch('__builtin__.open', mock.mock_open()), \
514 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
516 mock.patch('functest.utils.functest_utils.get_testcases_'
518 mock_obj = mock.Mock()
519 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
520 mock_obj.configure_mock(**attrs)
522 mock_yaml.return_value = mock_obj
524 self.assertDictEqual(functest_utils.
525 get_dict_by_test(self.testname),
528 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
529 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
530 mock_get_dict_by_test.return_value = self.testcase_dict
531 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
534 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
535 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
536 mock_get_dict_by_test.return_value = None
537 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
539 def test_get_parameter_from_yaml_failed(self):
540 self.file_yaml['general'] = None
541 with mock.patch('__builtin__.open', mock.mock_open()), \
542 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
544 self.assertRaises(ValueError) as excep:
545 mock_yaml.return_value = self.file_yaml
546 functest_utils.get_parameter_from_yaml(self.parameter,
548 self.assertTrue(("The parameter %s is not"
549 " defined in config_functest.yaml" %
550 self.parameter) in excep.exception)
552 def test_get_parameter_from_yaml_default(self):
553 with mock.patch('__builtin__.open', mock.mock_open()), \
554 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
556 mock_yaml.return_value = self.file_yaml
557 self.assertEqual(functest_utils.
558 get_parameter_from_yaml(self.parameter,
562 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
563 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
564 with mock.patch.dict(os.environ,
565 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
566 functest_utils.get_functest_config(self.parameter)
567 mock_get_parameter_from_yaml. \
568 assert_called_once_with(self.parameter,
571 def test_check_success_rate_default(self):
572 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
574 mock_criteria.return_value = self.criteria
575 resp = functest_utils.check_success_rate(self.case_name,
577 self.assertEqual(resp, 'PASS')
579 def test_check_success_rate_failed(self):
580 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
582 mock_criteria.return_value = self.criteria
583 resp = functest_utils.check_success_rate(self.case_name,
585 self.assertEqual(resp, 'FAIL')
589 def test_get_testcases_file_dir(self):
590 resp = functest_utils.get_testcases_file_dir()
591 self.assertEqual(resp,
592 "/home/opnfv/repos/functest/"
593 "functest/ci/testcases.yaml")
595 def test_get_functest_yaml(self):
596 with mock.patch('__builtin__.open', mock.mock_open()), \
597 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
599 mock_yaml.return_value = self.file_yaml
600 resp = functest_utils.get_functest_yaml()
601 self.assertEqual(resp, self.file_yaml)
603 @mock.patch('functest.utils.functest_utils.logger.info')
604 def test_print_separator(self, mock_logger_info):
605 functest_utils.print_separator()
606 mock_logger_info.assert_called_once_with("======================="
607 "=======================")
610 if __name__ == "__main__":
611 unittest.main(verbosity=2)