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.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
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.db_url_env = 'http://foo/testdb'
61 self.file_yaml = {'general': {'openstack': {'image_name':
64 @mock.patch('urllib2.urlopen',
65 side_effect=urllib2.URLError('no host given'))
66 def test_check_internet_connectivity_failed(self, mock_method):
67 self.assertFalse(functest_utils.check_internet_connectivity())
68 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
70 @mock.patch('urllib2.urlopen')
71 def test_check_internet_connectivity_default(self, mock_method):
72 self.assertTrue(functest_utils.check_internet_connectivity())
73 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
75 @mock.patch('urllib2.urlopen')
76 def test_check_internet_connectivity_debian(self, mock_method):
77 self.url = "https://www.debian.org/"
78 self.assertTrue(functest_utils.check_internet_connectivity(self.url))
79 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
81 @mock.patch('urllib2.urlopen',
82 side_effect=urllib2.URLError('no host given'))
83 def test_download_url_failed(self, mock_url):
84 self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
86 @mock.patch('urllib2.urlopen')
87 def test_download_url_default(self, mock_url):
88 with mock.patch("__builtin__.open", mock.mock_open()) as m, \
89 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
91 name = self.url.rsplit('/')[-1]
92 dest = self.dest_path + "/" + name
93 self.assertTrue(functest_utils.download_url(self.url,
95 m.assert_called_once_with(dest, 'wb')
96 self.assertTrue(mock_sh.called)
98 def test_get_git_branch(self):
99 with mock.patch('functest.utils.functest_utils.Repo') as mock_repo:
100 mock_obj2 = mock.Mock()
101 attrs = {'name': 'test_branch'}
102 mock_obj2.configure_mock(**attrs)
104 mock_obj = mock.Mock()
105 attrs = {'active_branch': mock_obj2}
106 mock_obj.configure_mock(**attrs)
108 mock_repo.return_value = mock_obj
109 self.assertEqual(functest_utils.get_git_branch(self.repo_path),
112 @mock.patch('functest.utils.functest_utils.Repo',
113 side_effect=NoSuchPathError)
114 def test_get_git_branch_failed(self, mock_repo):
115 self.assertRaises(NoSuchPathError,
116 lambda: functest_utils.get_git_branch(self.repo_path
119 @mock.patch('functest.utils.functest_utils.logger.error')
120 def test_get_installer_type_failed(self, mock_logger_error):
121 with mock.patch.dict(os.environ,
124 self.assertEqual(functest_utils.get_installer_type(),
126 mock_logger_error.assert_called_once_with("Impossible to retrieve"
127 " the installer type")
129 def test_get_installer_type_default(self):
130 with mock.patch.dict(os.environ,
131 {'INSTALLER_TYPE': 'test_installer'},
133 self.assertEqual(functest_utils.get_installer_type(),
136 @mock.patch('functest.utils.functest_utils.logger.info')
137 def test_get_scenario_failed(self, mock_logger_info):
138 with mock.patch.dict(os.environ,
141 self.assertEqual(functest_utils.get_scenario(),
142 "os-nosdn-nofeature-noha")
143 mock_logger_info.assert_called_once_with("Impossible to retrieve "
146 "os-nosdn-nofeature-noha")
148 def test_get_scenario_default(self):
149 with mock.patch.dict(os.environ,
150 {'DEPLOY_SCENARIO': 'test_scenario'},
152 self.assertEqual(functest_utils.get_scenario(),
155 @mock.patch('functest.utils.functest_utils.get_build_tag')
156 def test_get_version_daily_job(self, mock_get_build_tag):
157 mock_get_build_tag.return_value = self.build_tag
158 self.assertEqual(functest_utils.get_version(), self.version)
160 @mock.patch('functest.utils.functest_utils.get_build_tag')
161 def test_get_version_weekly_job(self, mock_get_build_tag):
162 mock_get_build_tag.return_value = self.build_tag_week
163 self.assertEqual(functest_utils.get_version(), self.version)
165 @mock.patch('functest.utils.functest_utils.get_build_tag')
166 def test_get_version_with_dummy_build_tag(self, mock_get_build_tag):
167 mock_get_build_tag.return_value = 'whatever'
168 self.assertEqual(functest_utils.get_version(), 'unknown')
170 @mock.patch('functest.utils.functest_utils.get_build_tag')
171 def test_get_version_unknown(self, mock_get_build_tag):
172 mock_get_build_tag.return_value = "unknown_build_tag"
173 self.assertEqual(functest_utils.get_version(), "unknown")
175 @mock.patch('functest.utils.functest_utils.logger.info')
176 def test_get_pod_name_failed(self, mock_logger_info):
177 with mock.patch.dict(os.environ,
180 self.assertEqual(functest_utils.get_pod_name(),
182 mock_logger_info.assert_called_once_with("Unable to retrieve "
184 "environment. Using "
185 "pod name 'unknown-pod'")
187 def test_get_pod_name_default(self):
188 with mock.patch.dict(os.environ,
189 {'NODE_NAME': 'test_node_name'},
191 self.assertEqual(functest_utils.get_pod_name(),
194 @mock.patch('functest.utils.functest_utils.logger.info')
195 def test_get_build_tag_failed(self, mock_logger_info):
196 with mock.patch.dict(os.environ,
199 self.assertEqual(functest_utils.get_build_tag(),
201 mock_logger_info.assert_called_once_with("Impossible to retrieve"
204 def test_get_build_tag_default(self):
205 with mock.patch.dict(os.environ,
206 {'BUILD_TAG': self.build_tag},
208 self.assertEqual(functest_utils.get_build_tag(),
211 def test_get_db_url_env_var(self):
212 with mock.patch.dict(os.environ,
213 {'TEST_DB_URL': self.db_url_env,
214 'CONFIG_FUNCTEST_YAML':
215 "./functest/ci/config_functest.yaml"},
217 self.assertEqual(functest_utils.get_db_url(),
220 @mock.patch('functest.utils.functest_utils.get_functest_config')
221 def test_get_db_url_default(self, mock_get_functest_config):
222 mock_get_functest_config.return_value = self.db_url
223 self.assertEqual(functest_utils.get_db_url(), self.db_url)
224 mock_get_functest_config.assert_called_once_with('results.test_db_url')
226 @mock.patch('functest.utils.functest_utils.logger.info')
227 def test_logger_test_results(self, mock_logger_info):
228 with mock.patch('functest.utils.functest_utils.get_pod_name',
229 return_value=self.node_name), \
230 mock.patch('functest.utils.functest_utils.get_scenario',
231 return_value=self.scenario), \
232 mock.patch('functest.utils.functest_utils.get_version',
233 return_value=self.version), \
234 mock.patch('functest.utils.functest_utils.get_build_tag',
235 return_value=self.build_tag), \
236 mock.patch('functest.utils.functest_utils.get_db_url',
237 return_value=self.db_url):
238 functest_utils.logger_test_results(self.project, self.case_name,
239 self.status, self.details)
240 mock_logger_info.assert_called_once_with(
242 "****************************************\n"
243 "\t %(p)s/%(n)s results \n\n"
244 "****************************************\n"
250 "build tag:\t%(b)s\n"
252 % {'p': self.project,
255 'pod': self.node_name,
262 def _get_env_dict(self, var):
263 dic = {'INSTALLER_TYPE': self.installer,
264 'DEPLOY_SCENARIO': self.scenario,
265 'NODE_NAME': self.node_name,
266 'BUILD_TAG': self.build_tag}
270 def _test_push_results_to_db_missing_env(self, env_var):
271 dic = self._get_env_dict(env_var)
272 with mock.patch('functest.utils.functest_utils.get_db_url',
273 return_value=self.db_url), \
274 mock.patch.dict(os.environ,
277 mock.patch('functest.utils.functest_utils.logger.error') \
278 as mock_logger_error:
279 functest_utils.push_results_to_db(self.project, self.case_name,
280 self.start_date, self.stop_date,
281 self.criteria, self.details)
282 mock_logger_error.assert_called_once_with("Please set env var: " +
286 def test_push_results_to_db_missing_installer(self):
287 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
289 def test_push_results_to_db_missing_scenario(self):
290 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
292 def test_push_results_to_db_missing_nodename(self):
293 self._test_push_results_to_db_missing_env('NODE_NAME')
295 def test_push_results_to_db_missing_buildtag(self):
296 self._test_push_results_to_db_missing_env('BUILD_TAG')
298 def test_push_results_to_db_incorrect_buildtag(self):
299 dic = self._get_env_dict(None)
300 dic['BUILD_TAG'] = 'incorrect_build_tag'
301 with mock.patch('functest.utils.functest_utils.get_db_url',
302 return_value=self.db_url), \
303 mock.patch.dict(os.environ,
306 mock.patch('functest.utils.functest_utils.logger.error') \
307 as mock_logger_error:
308 self.assertFalse(functest_utils.
309 push_results_to_db(self.project, self.case_name,
312 self.criteria, self.details))
313 mock_logger_error.assert_called_once_with("Please fix BUILD_TAG"
314 " env var: incorrect_"
317 def test_push_results_to_db_request_post_failed(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=requests.RequestException):
328 self.assertFalse(functest_utils.
329 push_results_to_db(self.project, self.case_name,
332 self.criteria, self.details))
333 mock_logger_error.assert_called_once_with(test_utils.
334 RegexMatch("Pushing "
340 def test_push_results_to_db_request_post_exception(self):
341 dic = self._get_env_dict(None)
342 with mock.patch('functest.utils.functest_utils.get_db_url',
343 return_value=self.db_url), \
344 mock.patch.dict(os.environ,
347 mock.patch('functest.utils.functest_utils.logger.error') \
348 as mock_logger_error, \
349 mock.patch('functest.utils.functest_utils.requests.post',
350 side_effect=Exception):
351 self.assertFalse(functest_utils.
352 push_results_to_db(self.project, self.case_name,
355 self.criteria, self.details))
356 self.assertTrue(mock_logger_error.called)
358 def test_push_results_to_db_default(self):
359 dic = self._get_env_dict(None)
360 with mock.patch('functest.utils.functest_utils.get_db_url',
361 return_value=self.db_url), \
362 mock.patch.dict(os.environ,
365 mock.patch('functest.utils.functest_utils.requests.post'):
366 self.assertTrue(functest_utils.
367 push_results_to_db(self.project, self.case_name,
370 self.criteria, self.details))
372 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
376 if FunctestUtilsTesting.readline == \
377 len(FunctestUtilsTesting.test_ip) - 1:
379 FunctestUtilsTesting.readline += 1
380 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
382 # TODO: get_resolvconf_ns
383 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
384 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
385 attrs = {'query.return_value': ["test"]}
386 mock_dns_resolve.configure_mock(**attrs)
389 attrs = {'readline.side_effect': self.readline_side}
390 m.configure_mock(**attrs)
392 with mock.patch("__builtin__.open") as mo:
394 self.assertEqual(functest_utils.get_resolvconf_ns(),
397 def _get_environ(self, var):
398 if var == 'INSTALLER_TYPE':
399 return self.installer
400 elif var == 'DEPLOY_SCENARIO':
404 def test_get_ci_envvars_default(self):
405 with mock.patch('os.environ.get',
406 side_effect=self._get_environ):
407 dic = {"installer": self.installer,
408 "scenario": self.scenario}
409 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
411 def cmd_readline(self):
412 return 'test_value\n'
414 @mock.patch('functest.utils.functest_utils.logger.error')
415 @mock.patch('functest.utils.functest_utils.logger.info')
416 def test_execute_command_args_present_with_error(self, mock_logger_info,
418 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
419 as mock_subproc_open, \
420 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
422 FunctestUtilsTesting.readline = 0
424 mock_obj = mock.Mock()
425 attrs = {'readline.side_effect': self.cmd_readline()}
426 mock_obj.configure_mock(**attrs)
428 mock_obj2 = mock.Mock()
429 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
430 mock_obj2.configure_mock(**attrs)
432 mock_subproc_open.return_value = mock_obj2
434 resp = functest_utils.execute_command(self.cmd, info=True,
435 error_msg=self.error_msg,
437 output_file=self.output_file)
438 self.assertEqual(resp, 1)
439 msg_exec = ("Executing command: '%s'" % self.cmd)
440 mock_logger_info.assert_called_once_with(msg_exec)
441 mopen.assert_called_once_with(self.output_file, "w")
442 mock_logger_error.assert_called_once_with(self.error_msg)
444 @mock.patch('functest.utils.functest_utils.logger.info')
445 def test_execute_command_args_present_with_success(self, mock_logger_info,
447 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
448 as mock_subproc_open, \
449 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
451 FunctestUtilsTesting.readline = 0
453 mock_obj = mock.Mock()
454 attrs = {'readline.side_effect': self.cmd_readline()}
455 mock_obj.configure_mock(**attrs)
457 mock_obj2 = mock.Mock()
458 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
459 mock_obj2.configure_mock(**attrs)
461 mock_subproc_open.return_value = mock_obj2
463 resp = functest_utils.execute_command(self.cmd, info=True,
464 error_msg=self.error_msg,
466 output_file=self.output_file)
467 self.assertEqual(resp, 0)
468 msg_exec = ("Executing command: '%s'" % self.cmd)
469 mock_logger_info.assert_called_once_with(msg_exec)
470 mopen.assert_called_once_with(self.output_file, "w")
472 @mock.patch('functest.utils.functest_utils.logger.info')
473 def test_execute_command_args_missing_with_success(self, mock_logger_info,
475 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
476 as mock_subproc_open:
478 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': 0}
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, 0)
496 @mock.patch('functest.utils.functest_utils.logger.error')
497 def test_execute_command_args_missing_with_error(self, mock_logger_error,
499 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
500 as mock_subproc_open:
502 FunctestUtilsTesting.readline = 2
503 mock_obj = mock.Mock()
504 attrs = {'readline.side_effect': self.cmd_readline()}
505 mock_obj.configure_mock(**attrs)
507 mock_obj2 = mock.Mock()
508 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
509 mock_obj2.configure_mock(**attrs)
511 mock_subproc_open.return_value = mock_obj2
513 resp = functest_utils.execute_command(self.cmd, info=False,
517 self.assertEqual(resp, 1)
519 def _get_functest_config(self, var):
522 @mock.patch('functest.utils.functest_utils.logger.error')
523 def test_get_dict_by_test(self, mock_logger_error):
524 with mock.patch('__builtin__.open', mock.mock_open()), \
525 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
527 mock.patch('functest.utils.functest_utils.get_testcases_'
529 mock_obj = mock.Mock()
530 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
531 mock_obj.configure_mock(**attrs)
533 mock_yaml.return_value = mock_obj
535 self.assertDictEqual(functest_utils.
536 get_dict_by_test(self.testname),
539 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
540 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
541 mock_get_dict_by_test.return_value = self.testcase_dict
542 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
545 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
546 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
547 mock_get_dict_by_test.return_value = None
548 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
550 def test_get_parameter_from_yaml_failed(self):
551 self.file_yaml['general'] = None
552 with mock.patch('__builtin__.open', mock.mock_open()), \
553 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
555 self.assertRaises(ValueError) as excep:
556 mock_yaml.return_value = self.file_yaml
557 functest_utils.get_parameter_from_yaml(self.parameter,
559 self.assertTrue(("The parameter %s is not"
560 " defined in config_functest.yaml" %
561 self.parameter) in excep.exception)
563 def test_get_parameter_from_yaml_default(self):
564 with mock.patch('__builtin__.open', mock.mock_open()), \
565 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
567 mock_yaml.return_value = self.file_yaml
568 self.assertEqual(functest_utils.
569 get_parameter_from_yaml(self.parameter,
573 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
574 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
575 with mock.patch.dict(os.environ,
576 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
577 functest_utils.get_functest_config(self.parameter)
578 mock_get_parameter_from_yaml. \
579 assert_called_once_with(self.parameter,
582 def test_check_success_rate_default(self):
583 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
585 mock_criteria.return_value = self.criteria
586 resp = functest_utils.check_success_rate(self.case_name,
588 self.assertEqual(resp, 'PASS')
590 def test_check_success_rate_failed(self):
591 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
593 mock_criteria.return_value = self.criteria
594 resp = functest_utils.check_success_rate(self.case_name,
596 self.assertEqual(resp, 'FAIL')
600 def test_get_testcases_file_dir(self):
601 resp = functest_utils.get_testcases_file_dir()
602 self.assertEqual(resp,
603 "/home/opnfv/repos/functest/"
604 "functest/ci/testcases.yaml")
606 def test_get_functest_yaml(self):
607 with mock.patch('__builtin__.open', mock.mock_open()), \
608 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
610 mock_yaml.return_value = self.file_yaml
611 resp = functest_utils.get_functest_yaml()
612 self.assertEqual(resp, self.file_yaml)
614 @mock.patch('functest.utils.functest_utils.logger.info')
615 def test_print_separator(self, mock_logger_info):
616 functest_utils.print_separator()
617 mock_logger_info.assert_called_once_with("======================="
618 "=======================")
621 if __name__ == "__main__":
622 unittest.main(verbosity=2)