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.info')
135 def test_get_scenario_failed(self, mock_logger_info):
136 with mock.patch.dict(os.environ,
139 self.assertEqual(functest_utils.get_scenario(),
140 "os-nosdn-nofeature-noha")
141 mock_logger_info.assert_called_once_with("Impossible to retrieve "
144 "os-nosdn-nofeature-noha")
146 def test_get_scenario_default(self):
147 with mock.patch.dict(os.environ,
148 {'DEPLOY_SCENARIO': 'test_scenario'},
150 self.assertEqual(functest_utils.get_scenario(),
153 @mock.patch('functest.utils.functest_utils.get_build_tag')
154 def test_get_version_default(self, mock_get_build_tag):
155 mock_get_build_tag.return_value = self.build_tag
156 self.assertEqual(functest_utils.get_version(), self.version)
158 @mock.patch('functest.utils.functest_utils.get_build_tag')
159 def test_get_version_unknown(self, mock_get_build_tag):
160 mock_get_build_tag.return_value = "unknown_build_tag"
161 self.assertEqual(functest_utils.get_version(), "unknown")
163 @mock.patch('functest.utils.functest_utils.logger.info')
164 def test_get_pod_name_failed(self, mock_logger_info):
165 with mock.patch.dict(os.environ,
168 self.assertEqual(functest_utils.get_pod_name(),
170 mock_logger_info.assert_called_once_with("Unable to retrieve "
172 "environment. Using "
173 "pod name 'unknown-pod'")
175 def test_get_pod_name_default(self):
176 with mock.patch.dict(os.environ,
177 {'NODE_NAME': 'test_node_name'},
179 self.assertEqual(functest_utils.get_pod_name(),
182 @mock.patch('functest.utils.functest_utils.logger.info')
183 def test_get_build_tag_failed(self, mock_logger_info):
184 with mock.patch.dict(os.environ,
187 self.assertEqual(functest_utils.get_build_tag(),
189 mock_logger_info.assert_called_once_with("Impossible to retrieve"
192 def test_get_build_tag_default(self):
193 with mock.patch.dict(os.environ,
194 {'BUILD_TAG': self.build_tag},
196 self.assertEqual(functest_utils.get_build_tag(),
199 @mock.patch('functest.utils.functest_utils.get_functest_config')
200 def test_get_db_url(self, mock_get_functest_config):
201 mock_get_functest_config.return_value = self.db_url
202 self.assertEqual(functest_utils.get_db_url(), self.db_url)
203 mock_get_functest_config.assert_called_once_with('results.test_db_url')
205 @mock.patch('functest.utils.functest_utils.logger.info')
206 def test_logger_test_results(self, mock_logger_info):
207 with mock.patch('functest.utils.functest_utils.get_pod_name',
208 return_value=self.node_name), \
209 mock.patch('functest.utils.functest_utils.get_scenario',
210 return_value=self.scenario), \
211 mock.patch('functest.utils.functest_utils.get_version',
212 return_value=self.version), \
213 mock.patch('functest.utils.functest_utils.get_build_tag',
214 return_value=self.build_tag), \
215 mock.patch('functest.utils.functest_utils.get_db_url',
216 return_value=self.db_url):
217 functest_utils.logger_test_results(self.project, self.case_name,
218 self.status, self.details)
219 mock_logger_info.assert_called_once_with(
221 "****************************************\n"
222 "\t %(p)s/%(n)s results \n\n"
223 "****************************************\n"
229 "build tag:\t%(b)s\n"
231 % {'p': self.project,
234 'pod': self.node_name,
241 def _get_env_dict(self, var):
242 dic = {'INSTALLER_TYPE': self.installer,
243 'DEPLOY_SCENARIO': self.scenario,
244 'NODE_NAME': self.node_name,
245 'BUILD_TAG': self.build_tag}
249 def _test_push_results_to_db_missing_env(self, env_var):
250 dic = self._get_env_dict(env_var)
251 with mock.patch('functest.utils.functest_utils.get_db_url',
252 return_value=self.db_url), \
253 mock.patch.dict(os.environ,
256 mock.patch('functest.utils.functest_utils.logger.error') \
257 as mock_logger_error:
258 functest_utils.push_results_to_db(self.project, self.case_name,
259 self.start_date, self.stop_date,
260 self.criteria, self.details)
261 mock_logger_error.assert_called_once_with("Please set env var: " +
265 def test_push_results_to_db_missing_installer(self):
266 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
268 def test_push_results_to_db_missing_scenario(self):
269 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
271 def test_push_results_to_db_missing_nodename(self):
272 self._test_push_results_to_db_missing_env('NODE_NAME')
274 def test_push_results_to_db_missing_buildtag(self):
275 self._test_push_results_to_db_missing_env('BUILD_TAG')
277 def test_push_results_to_db_incorrect_buildtag(self):
278 dic = self._get_env_dict(None)
279 dic['BUILD_TAG'] = 'incorrect_build_tag'
280 with mock.patch('functest.utils.functest_utils.get_db_url',
281 return_value=self.db_url), \
282 mock.patch.dict(os.environ,
285 mock.patch('functest.utils.functest_utils.logger.error') \
286 as mock_logger_error:
287 self.assertFalse(functest_utils.
288 push_results_to_db(self.project, self.case_name,
291 self.criteria, self.details))
292 mock_logger_error.assert_called_once_with("Please fix BUILD_TAG"
293 " env var: incorrect_"
296 def test_push_results_to_db_request_post_failed(self):
297 dic = self._get_env_dict(None)
298 with mock.patch('functest.utils.functest_utils.get_db_url',
299 return_value=self.db_url), \
300 mock.patch.dict(os.environ,
303 mock.patch('functest.utils.functest_utils.logger.error') \
304 as mock_logger_error, \
305 mock.patch('functest.utils.functest_utils.requests.post',
306 side_effect=requests.RequestException):
307 self.assertFalse(functest_utils.
308 push_results_to_db(self.project, self.case_name,
311 self.criteria, self.details))
312 mock_logger_error.assert_called_once_with(test_utils.
313 RegexMatch("Pushing "
319 def test_push_results_to_db_request_post_exception(self):
320 dic = self._get_env_dict(None)
321 with mock.patch('functest.utils.functest_utils.get_db_url',
322 return_value=self.db_url), \
323 mock.patch.dict(os.environ,
326 mock.patch('functest.utils.functest_utils.logger.error') \
327 as mock_logger_error, \
328 mock.patch('functest.utils.functest_utils.requests.post',
329 side_effect=Exception):
330 self.assertFalse(functest_utils.
331 push_results_to_db(self.project, self.case_name,
334 self.criteria, self.details))
335 self.assertTrue(mock_logger_error.called)
337 def test_push_results_to_db_default(self):
338 dic = self._get_env_dict(None)
339 with mock.patch('functest.utils.functest_utils.get_db_url',
340 return_value=self.db_url), \
341 mock.patch.dict(os.environ,
344 mock.patch('functest.utils.functest_utils.requests.post'):
345 self.assertTrue(functest_utils.
346 push_results_to_db(self.project, self.case_name,
349 self.criteria, self.details))
351 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
355 if FunctestUtilsTesting.readline == \
356 len(FunctestUtilsTesting.test_ip) - 1:
358 FunctestUtilsTesting.readline += 1
359 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
361 # TODO: get_resolvconf_ns
362 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
363 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
364 attrs = {'query.return_value': ["test"]}
365 mock_dns_resolve.configure_mock(**attrs)
368 attrs = {'readline.side_effect': self.readline_side}
369 m.configure_mock(**attrs)
371 with mock.patch("__builtin__.open") as mo:
373 self.assertEqual(functest_utils.get_resolvconf_ns(),
376 def _get_environ(self, var):
377 if var == 'INSTALLER_TYPE':
378 return self.installer
379 elif var == 'DEPLOY_SCENARIO':
383 def test_get_ci_envvars_default(self):
384 with mock.patch('os.environ.get',
385 side_effect=self._get_environ):
386 dic = {"installer": self.installer,
387 "scenario": self.scenario}
388 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
390 def cmd_readline(self):
391 return 'test_value\n'
393 @mock.patch('functest.utils.functest_utils.logger.error')
394 @mock.patch('functest.utils.functest_utils.logger.info')
395 def test_execute_command_args_present_with_error(self, mock_logger_info,
397 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
398 as mock_subproc_open, \
399 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
401 FunctestUtilsTesting.readline = 0
403 mock_obj = mock.Mock()
404 attrs = {'readline.side_effect': self.cmd_readline()}
405 mock_obj.configure_mock(**attrs)
407 mock_obj2 = mock.Mock()
408 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
409 mock_obj2.configure_mock(**attrs)
411 mock_subproc_open.return_value = mock_obj2
413 resp = functest_utils.execute_command(self.cmd, info=True,
414 error_msg=self.error_msg,
416 output_file=self.output_file)
417 self.assertEqual(resp, 1)
418 msg_exec = ("Executing command: '%s'" % self.cmd)
419 mock_logger_info.assert_called_once_with(msg_exec)
420 mopen.assert_called_once_with(self.output_file, "w")
421 mock_logger_error.assert_called_once_with(self.error_msg)
423 @mock.patch('functest.utils.functest_utils.logger.info')
424 def test_execute_command_args_present_with_success(self, mock_logger_info,
426 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
427 as mock_subproc_open, \
428 mock.patch('__builtin__.open', mock.mock_open()) as mopen:
430 FunctestUtilsTesting.readline = 0
432 mock_obj = mock.Mock()
433 attrs = {'readline.side_effect': self.cmd_readline()}
434 mock_obj.configure_mock(**attrs)
436 mock_obj2 = mock.Mock()
437 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
438 mock_obj2.configure_mock(**attrs)
440 mock_subproc_open.return_value = mock_obj2
442 resp = functest_utils.execute_command(self.cmd, info=True,
443 error_msg=self.error_msg,
445 output_file=self.output_file)
446 self.assertEqual(resp, 0)
447 msg_exec = ("Executing command: '%s'" % self.cmd)
448 mock_logger_info.assert_called_once_with(msg_exec)
449 mopen.assert_called_once_with(self.output_file, "w")
451 @mock.patch('functest.utils.functest_utils.logger.info')
452 def test_execute_command_args_missing_with_success(self, mock_logger_info,
454 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
455 as mock_subproc_open:
457 FunctestUtilsTesting.readline = 2
459 mock_obj = mock.Mock()
460 attrs = {'readline.side_effect': self.cmd_readline()}
461 mock_obj.configure_mock(**attrs)
463 mock_obj2 = mock.Mock()
464 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
465 mock_obj2.configure_mock(**attrs)
467 mock_subproc_open.return_value = mock_obj2
469 resp = functest_utils.execute_command(self.cmd, info=False,
473 self.assertEqual(resp, 0)
475 @mock.patch('functest.utils.functest_utils.logger.error')
476 def test_execute_command_args_missing_with_error(self, mock_logger_error,
478 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
479 as mock_subproc_open:
481 FunctestUtilsTesting.readline = 2
482 mock_obj = mock.Mock()
483 attrs = {'readline.side_effect': self.cmd_readline()}
484 mock_obj.configure_mock(**attrs)
486 mock_obj2 = mock.Mock()
487 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
488 mock_obj2.configure_mock(**attrs)
490 mock_subproc_open.return_value = mock_obj2
492 resp = functest_utils.execute_command(self.cmd, info=False,
496 self.assertEqual(resp, 1)
498 def _get_functest_config(self, var):
501 @mock.patch('functest.utils.functest_utils.logger.error')
502 def test_get_dict_by_test(self, mock_logger_error):
503 with mock.patch('__builtin__.open', mock.mock_open()), \
504 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
506 mock.patch('functest.utils.functest_utils.get_testcases_'
508 mock_obj = mock.Mock()
509 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
510 mock_obj.configure_mock(**attrs)
512 mock_yaml.return_value = mock_obj
514 self.assertDictEqual(functest_utils.
515 get_dict_by_test(self.testname),
518 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
519 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
520 mock_get_dict_by_test.return_value = self.testcase_dict
521 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
524 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
525 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
526 mock_get_dict_by_test.return_value = None
527 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
529 def test_get_parameter_from_yaml_failed(self):
530 self.file_yaml['general'] = None
531 with mock.patch('__builtin__.open', mock.mock_open()), \
532 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
534 self.assertRaises(ValueError) as excep:
535 mock_yaml.return_value = self.file_yaml
536 functest_utils.get_parameter_from_yaml(self.parameter,
538 self.assertTrue(("The parameter %s is not"
539 " defined in config_functest.yaml" %
540 self.parameter) in excep.exception)
542 def test_get_parameter_from_yaml_default(self):
543 with mock.patch('__builtin__.open', mock.mock_open()), \
544 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
546 mock_yaml.return_value = self.file_yaml
547 self.assertEqual(functest_utils.
548 get_parameter_from_yaml(self.parameter,
552 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
553 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
554 with mock.patch.dict(os.environ,
555 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
556 functest_utils.get_functest_config(self.parameter)
557 mock_get_parameter_from_yaml. \
558 assert_called_once_with(self.parameter,
561 def test_check_success_rate_default(self):
562 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
564 mock_criteria.return_value = self.criteria
565 resp = functest_utils.check_success_rate(self.case_name,
567 self.assertEqual(resp, 'PASS')
569 def test_check_success_rate_failed(self):
570 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
572 mock_criteria.return_value = self.criteria
573 resp = functest_utils.check_success_rate(self.case_name,
575 self.assertEqual(resp, 'FAIL')
579 def test_get_testcases_file_dir(self):
580 resp = functest_utils.get_testcases_file_dir()
581 self.assertEqual(resp,
582 "/home/opnfv/repos/functest/"
583 "functest/ci/testcases.yaml")
585 def test_get_functest_yaml(self):
586 with mock.patch('__builtin__.open', mock.mock_open()), \
587 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
589 mock_yaml.return_value = self.file_yaml
590 resp = functest_utils.get_functest_yaml()
591 self.assertEqual(resp, self.file_yaml)
593 @mock.patch('functest.utils.functest_utils.logger.info')
594 def test_print_separator(self, mock_logger_info):
595 functest_utils.print_separator()
596 mock_logger_info.assert_called_once_with("======================="
597 "=======================")
600 if __name__ == "__main__":
601 unittest.main(verbosity=2)