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
15 from git.exc import NoSuchPathError
18 from six.moves import urllib
20 from functest.tests.unit import test_utils
21 from functest.utils import functest_utils
24 class FunctestUtilsTesting(unittest.TestCase):
27 self.url = 'http://www.opnfv.org/'
29 self.dest_path = 'test_path'
30 self.repo_path = 'test_repo_path'
31 self.installer = 'test_installer'
32 self.scenario = 'test_scenario'
33 self.build_tag = 'jenkins-functest-fuel-opnfv-jump-2-daily-master-190'
34 self.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
35 self.version = 'master'
36 self.node_name = 'test_node_name'
37 self.project = 'test_project'
38 self.case_name = 'test_case_name'
39 self.status = 'test_status'
40 self.details = 'test_details'
41 self.db_url = 'test_db_url'
44 self.start_date = 1482624000
45 self.stop_date = 1482624000
46 self.start_time = time.time()
47 self.stop_time = time.time()
49 self.test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
50 self.test_file = 'test_file'
51 self.error_msg = 'test_error_msg'
53 self.output_file = 'test_output_file'
54 self.testname = 'testname'
55 self.testcase_dict = {'case_name': 'testname',
56 '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('six.moves.urllib.request.urlopen',
64 side_effect=urllib.error.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('six.moves.urllib.request.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('six.moves.urllib.request.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('six.moves.urllib.request.urlopen',
81 side_effect=urllib.error.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('six.moves.urllib.request.urlopen')
86 def test_download_url_default(self, mock_url):
87 with mock.patch("six.moves.builtins.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_daily_job(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_weekly_job(self, mock_get_build_tag):
161 mock_get_build_tag.return_value = self.build_tag_week
162 self.assertEqual(functest_utils.get_version(), self.version)
164 @mock.patch('functest.utils.functest_utils.get_build_tag')
165 def test_get_version_with_dummy_build_tag(self, mock_get_build_tag):
166 mock_get_build_tag.return_value = 'whatever'
167 self.assertEqual(functest_utils.get_version(), 'unknown')
169 @mock.patch('functest.utils.functest_utils.get_build_tag')
170 def test_get_version_unknown(self, mock_get_build_tag):
171 mock_get_build_tag.return_value = "unknown_build_tag"
172 self.assertEqual(functest_utils.get_version(), "unknown")
174 @mock.patch('functest.utils.functest_utils.logger.info')
175 def test_get_pod_name_failed(self, mock_logger_info):
176 with mock.patch.dict(os.environ,
179 self.assertEqual(functest_utils.get_pod_name(),
181 mock_logger_info.assert_called_once_with("Unable to retrieve "
183 "environment. Using "
184 "pod name 'unknown-pod'")
186 def test_get_pod_name_default(self):
187 with mock.patch.dict(os.environ,
188 {'NODE_NAME': 'test_node_name'},
190 self.assertEqual(functest_utils.get_pod_name(),
193 @mock.patch('functest.utils.functest_utils.logger.info')
194 def test_get_build_tag_failed(self, mock_logger_info):
195 with mock.patch.dict(os.environ,
198 self.assertEqual(functest_utils.get_build_tag(),
200 mock_logger_info.assert_called_once_with("Impossible to retrieve"
203 def test_get_build_tag_default(self):
204 with mock.patch.dict(os.environ,
205 {'BUILD_TAG': self.build_tag},
207 self.assertEqual(functest_utils.get_build_tag(),
210 def test_get_db_url_env_var(self):
211 with mock.patch.dict(os.environ,
212 {'TEST_DB_URL': self.db_url_env,
213 'CONFIG_FUNCTEST_YAML':
214 "./functest/ci/config_functest.yaml"},
216 self.assertEqual(functest_utils.get_db_url(),
219 @mock.patch('functest.utils.functest_utils.get_functest_config')
220 def test_get_db_url_default(self, mock_get_functest_config):
221 mock_get_functest_config.return_value = self.db_url
222 self.assertEqual(functest_utils.get_db_url(), self.db_url)
223 mock_get_functest_config.assert_called_once_with('results.test_db_url')
225 @mock.patch('functest.utils.functest_utils.logger.info')
226 def test_logger_test_results(self, mock_logger_info):
227 with mock.patch('functest.utils.functest_utils.get_pod_name',
228 return_value=self.node_name), \
229 mock.patch('functest.utils.functest_utils.get_scenario',
230 return_value=self.scenario), \
231 mock.patch('functest.utils.functest_utils.get_version',
232 return_value=self.version), \
233 mock.patch('functest.utils.functest_utils.get_build_tag',
234 return_value=self.build_tag), \
235 mock.patch('functest.utils.functest_utils.get_db_url',
236 return_value=self.db_url):
237 functest_utils.logger_test_results(self.project, self.case_name,
238 self.status, self.details)
239 mock_logger_info.assert_called_once_with(
241 "****************************************\n"
242 "\t %(p)s/%(n)s results \n\n"
243 "****************************************\n"
249 "build tag:\t%(b)s\n"
251 % {'p': self.project,
254 'pod': self.node_name,
261 def _get_env_dict(self, var):
262 dic = {'INSTALLER_TYPE': self.installer,
263 'DEPLOY_SCENARIO': self.scenario,
264 'NODE_NAME': self.node_name,
265 'BUILD_TAG': self.build_tag}
269 def _test_push_results_to_db_missing_env(self, env_var):
270 dic = self._get_env_dict(env_var)
271 with mock.patch('functest.utils.functest_utils.get_db_url',
272 return_value=self.db_url), \
273 mock.patch.dict(os.environ,
276 mock.patch('functest.utils.functest_utils.logger.error') \
277 as mock_logger_error:
278 functest_utils.push_results_to_db(self.project, self.case_name,
279 self.start_date, self.stop_date,
280 self.result, self.details)
281 mock_logger_error.assert_called_once_with("Please set env var: " +
285 def test_push_results_to_db_missing_installer(self):
286 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
288 def test_push_results_to_db_missing_scenario(self):
289 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
291 def test_push_results_to_db_missing_nodename(self):
292 self._test_push_results_to_db_missing_env('NODE_NAME')
294 def test_push_results_to_db_missing_buildtag(self):
295 self._test_push_results_to_db_missing_env('BUILD_TAG')
297 def test_push_results_to_db_request_post_failed(self):
298 dic = self._get_env_dict(None)
299 with mock.patch('functest.utils.functest_utils.get_db_url',
300 return_value=self.db_url), \
301 mock.patch.dict(os.environ,
304 mock.patch('functest.utils.functest_utils.logger.error') \
305 as mock_logger_error, \
306 mock.patch('functest.utils.functest_utils.requests.post',
307 side_effect=requests.RequestException):
308 self.assertFalse(functest_utils.
309 push_results_to_db(self.project, self.case_name,
312 self.result, self.details))
313 mock_logger_error.assert_called_once_with(test_utils.
314 RegexMatch("Pushing "
320 def test_push_results_to_db_request_post_exception(self):
321 dic = self._get_env_dict(None)
322 with mock.patch('functest.utils.functest_utils.get_db_url',
323 return_value=self.db_url), \
324 mock.patch.dict(os.environ,
327 mock.patch('functest.utils.functest_utils.logger.error') \
328 as mock_logger_error, \
329 mock.patch('functest.utils.functest_utils.requests.post',
330 side_effect=Exception):
331 self.assertFalse(functest_utils.
332 push_results_to_db(self.project, self.case_name,
335 self.result, self.details))
336 self.assertTrue(mock_logger_error.called)
338 def test_push_results_to_db_default(self):
339 dic = self._get_env_dict(None)
340 with mock.patch('functest.utils.functest_utils.get_db_url',
341 return_value=self.db_url), \
342 mock.patch.dict(os.environ,
345 mock.patch('functest.utils.functest_utils.requests.post'):
346 self.assertTrue(functest_utils.
347 push_results_to_db(self.project, self.case_name,
350 self.result, self.details))
352 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
356 if FunctestUtilsTesting.readline == \
357 len(FunctestUtilsTesting.test_ip) - 1:
359 FunctestUtilsTesting.readline += 1
360 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
362 # TODO: get_resolvconf_ns
363 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
364 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
365 attrs = {'query.return_value': ["test"]}
366 mock_dns_resolve.configure_mock(**attrs)
369 attrs = {'readline.side_effect': self.readline_side}
370 m.configure_mock(**attrs)
372 with mock.patch("six.moves.builtins.open") as mo:
374 self.assertEqual(functest_utils.get_resolvconf_ns(),
377 def _get_environ(self, var):
378 if var == 'INSTALLER_TYPE':
379 return self.installer
380 elif var == 'DEPLOY_SCENARIO':
384 def test_get_ci_envvars_default(self):
385 with mock.patch('os.environ.get',
386 side_effect=self._get_environ):
387 dic = {"installer": self.installer,
388 "scenario": self.scenario}
389 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
391 def cmd_readline(self):
392 return 'test_value\n'
394 @mock.patch('functest.utils.functest_utils.logger.error')
395 @mock.patch('functest.utils.functest_utils.logger.info')
396 def test_execute_command_args_present_with_error(self, mock_logger_info,
398 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
399 as mock_subproc_open, \
400 mock.patch('six.moves.builtins.open',
401 mock.mock_open()) as mopen:
403 FunctestUtilsTesting.readline = 0
405 mock_obj = mock.Mock()
406 attrs = {'readline.side_effect': self.cmd_readline()}
407 mock_obj.configure_mock(**attrs)
409 mock_obj2 = mock.Mock()
410 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
411 mock_obj2.configure_mock(**attrs)
413 mock_subproc_open.return_value = mock_obj2
415 resp = functest_utils.execute_command(self.cmd, info=True,
416 error_msg=self.error_msg,
418 output_file=self.output_file)
419 self.assertEqual(resp, 1)
420 msg_exec = ("Executing command: '%s'" % self.cmd)
421 mock_logger_info.assert_called_once_with(msg_exec)
422 mopen.assert_called_once_with(self.output_file, "w")
423 mock_logger_error.assert_called_once_with(self.error_msg)
425 @mock.patch('functest.utils.functest_utils.logger.info')
426 def test_execute_command_args_present_with_success(self, mock_logger_info,
428 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
429 as mock_subproc_open, \
430 mock.patch('six.moves.builtins.open',
431 mock.mock_open()) as mopen:
433 FunctestUtilsTesting.readline = 0
435 mock_obj = mock.Mock()
436 attrs = {'readline.side_effect': self.cmd_readline()}
437 mock_obj.configure_mock(**attrs)
439 mock_obj2 = mock.Mock()
440 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
441 mock_obj2.configure_mock(**attrs)
443 mock_subproc_open.return_value = mock_obj2
445 resp = functest_utils.execute_command(self.cmd, info=True,
446 error_msg=self.error_msg,
448 output_file=self.output_file)
449 self.assertEqual(resp, 0)
450 msg_exec = ("Executing command: '%s'" % self.cmd)
451 mock_logger_info.assert_called_once_with(msg_exec)
452 mopen.assert_called_once_with(self.output_file, "w")
454 @mock.patch('functest.utils.functest_utils.logger.info')
455 def test_execute_command_args_missing_with_success(self, mock_logger_info,
457 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
458 as mock_subproc_open:
460 FunctestUtilsTesting.readline = 2
462 mock_obj = mock.Mock()
463 attrs = {'readline.side_effect': self.cmd_readline()}
464 mock_obj.configure_mock(**attrs)
466 mock_obj2 = mock.Mock()
467 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
468 mock_obj2.configure_mock(**attrs)
470 mock_subproc_open.return_value = mock_obj2
472 resp = functest_utils.execute_command(self.cmd, info=False,
476 self.assertEqual(resp, 0)
478 @mock.patch('functest.utils.functest_utils.logger.error')
479 def test_execute_command_args_missing_with_error(self, mock_logger_error,
481 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
482 as mock_subproc_open:
484 FunctestUtilsTesting.readline = 2
485 mock_obj = mock.Mock()
486 attrs = {'readline.side_effect': self.cmd_readline()}
487 mock_obj.configure_mock(**attrs)
489 mock_obj2 = mock.Mock()
490 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
491 mock_obj2.configure_mock(**attrs)
493 mock_subproc_open.return_value = mock_obj2
495 resp = functest_utils.execute_command(self.cmd, info=False,
499 self.assertEqual(resp, 1)
501 def _get_functest_config(self, var):
504 @mock.patch('functest.utils.functest_utils.logger.error')
505 def test_get_dict_by_test(self, mock_logger_error):
506 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
507 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
509 mock.patch('functest.utils.functest_utils.get_testcases_'
511 mock_obj = mock.Mock()
512 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
513 mock_obj.configure_mock(**attrs)
515 mock_yaml.return_value = mock_obj
517 self.assertDictEqual(functest_utils.
518 get_dict_by_test(self.testname),
521 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
522 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
523 mock_get_dict_by_test.return_value = self.testcase_dict
524 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
527 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
528 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
529 mock_get_dict_by_test.return_value = None
530 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
532 def test_get_parameter_from_yaml_failed(self):
533 self.file_yaml['general'] = None
534 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
535 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
537 self.assertRaises(ValueError) as excep:
538 mock_yaml.return_value = self.file_yaml
539 functest_utils.get_parameter_from_yaml(self.parameter,
541 self.assertTrue(("The parameter %s is not"
542 " defined in config_functest.yaml" %
543 self.parameter) in excep.exception)
545 def test_get_parameter_from_yaml_default(self):
546 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
547 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
549 mock_yaml.return_value = self.file_yaml
550 self.assertEqual(functest_utils.
551 get_parameter_from_yaml(self.parameter,
555 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
556 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
557 with mock.patch.dict(os.environ,
558 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
559 functest_utils.get_functest_config(self.parameter)
560 mock_get_parameter_from_yaml. \
561 assert_called_once_with(self.parameter,
566 def test_get_testcases_file_dir(self):
567 resp = functest_utils.get_testcases_file_dir()
568 self.assertEqual(resp,
569 "/home/opnfv/repos/functest/"
570 "functest/ci/testcases.yaml")
572 def test_get_functest_yaml(self):
573 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
574 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
576 mock_yaml.return_value = self.file_yaml
577 resp = functest_utils.get_functest_yaml()
578 self.assertEqual(resp, self.file_yaml)
580 @mock.patch('functest.utils.functest_utils.logger.info')
581 def test_print_separator(self, mock_logger_info):
582 functest_utils.print_separator()
583 mock_logger_info.assert_called_once_with("======================="
584 "=======================")
587 if __name__ == "__main__":
588 logging.disable(logging.CRITICAL)
589 unittest.main(verbosity=2)