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):
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'
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 = {'case_name': 'testname',
58 'criteria': self.criteria}
59 self.parameter = 'general.openstack.image_name'
60 self.config_yaml = 'test_config_yaml-'
61 self.db_url_env = 'http://foo/testdb'
62 self.file_yaml = {'general': {'openstack': {'image_name':
65 @mock.patch('six.moves.urllib.request.urlopen',
66 side_effect=urllib.error.URLError('no host given'))
67 def test_check_internet_connectivity_failed(self, mock_method):
68 self.assertFalse(functest_utils.check_internet_connectivity())
69 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
71 @mock.patch('six.moves.urllib.request.urlopen')
72 def test_check_internet_connectivity_default(self, mock_method):
73 self.assertTrue(functest_utils.check_internet_connectivity())
74 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
76 @mock.patch('six.moves.urllib.request.urlopen')
77 def test_check_internet_connectivity_debian(self, mock_method):
78 self.url = "https://www.debian.org/"
79 self.assertTrue(functest_utils.check_internet_connectivity(self.url))
80 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
82 @mock.patch('six.moves.urllib.request.urlopen',
83 side_effect=urllib.error.URLError('no host given'))
84 def test_download_url_failed(self, mock_url):
85 self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
87 @mock.patch('six.moves.urllib.request.urlopen')
88 def test_download_url_default(self, mock_url):
89 with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
90 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
92 name = self.url.rsplit('/')[-1]
93 dest = self.dest_path + "/" + name
94 self.assertTrue(functest_utils.download_url(self.url,
96 m.assert_called_once_with(dest, 'wb')
97 self.assertTrue(mock_sh.called)
99 def test_get_git_branch(self):
100 with mock.patch('functest.utils.functest_utils.Repo') as mock_repo:
101 mock_obj2 = mock.Mock()
102 attrs = {'name': 'test_branch'}
103 mock_obj2.configure_mock(**attrs)
105 mock_obj = mock.Mock()
106 attrs = {'active_branch': mock_obj2}
107 mock_obj.configure_mock(**attrs)
109 mock_repo.return_value = mock_obj
110 self.assertEqual(functest_utils.get_git_branch(self.repo_path),
113 @mock.patch('functest.utils.functest_utils.Repo',
114 side_effect=NoSuchPathError)
115 def test_get_git_branch_failed(self, mock_repo):
116 self.assertRaises(NoSuchPathError,
117 lambda: functest_utils.get_git_branch(self.repo_path
120 @mock.patch('functest.utils.functest_utils.logger.error')
121 def test_get_installer_type_failed(self, mock_logger_error):
122 with mock.patch.dict(os.environ,
125 self.assertEqual(functest_utils.get_installer_type(),
127 mock_logger_error.assert_called_once_with("Impossible to retrieve"
128 " the installer type")
130 def test_get_installer_type_default(self):
131 with mock.patch.dict(os.environ,
132 {'INSTALLER_TYPE': 'test_installer'},
134 self.assertEqual(functest_utils.get_installer_type(),
137 @mock.patch('functest.utils.functest_utils.logger.info')
138 def test_get_scenario_failed(self, mock_logger_info):
139 with mock.patch.dict(os.environ,
142 self.assertEqual(functest_utils.get_scenario(),
143 "os-nosdn-nofeature-noha")
144 mock_logger_info.assert_called_once_with("Impossible to retrieve "
147 "os-nosdn-nofeature-noha")
149 def test_get_scenario_default(self):
150 with mock.patch.dict(os.environ,
151 {'DEPLOY_SCENARIO': 'test_scenario'},
153 self.assertEqual(functest_utils.get_scenario(),
156 @mock.patch('functest.utils.functest_utils.get_build_tag')
157 def test_get_version_daily_job(self, mock_get_build_tag):
158 mock_get_build_tag.return_value = self.build_tag
159 self.assertEqual(functest_utils.get_version(), self.version)
161 @mock.patch('functest.utils.functest_utils.get_build_tag')
162 def test_get_version_weekly_job(self, mock_get_build_tag):
163 mock_get_build_tag.return_value = self.build_tag_week
164 self.assertEqual(functest_utils.get_version(), self.version)
166 @mock.patch('functest.utils.functest_utils.get_build_tag')
167 def test_get_version_with_dummy_build_tag(self, mock_get_build_tag):
168 mock_get_build_tag.return_value = 'whatever'
169 self.assertEqual(functest_utils.get_version(), 'unknown')
171 @mock.patch('functest.utils.functest_utils.get_build_tag')
172 def test_get_version_unknown(self, mock_get_build_tag):
173 mock_get_build_tag.return_value = "unknown_build_tag"
174 self.assertEqual(functest_utils.get_version(), "unknown")
176 @mock.patch('functest.utils.functest_utils.logger.info')
177 def test_get_pod_name_failed(self, mock_logger_info):
178 with mock.patch.dict(os.environ,
181 self.assertEqual(functest_utils.get_pod_name(),
183 mock_logger_info.assert_called_once_with("Unable to retrieve "
185 "environment. Using "
186 "pod name 'unknown-pod'")
188 def test_get_pod_name_default(self):
189 with mock.patch.dict(os.environ,
190 {'NODE_NAME': 'test_node_name'},
192 self.assertEqual(functest_utils.get_pod_name(),
195 @mock.patch('functest.utils.functest_utils.logger.info')
196 def test_get_build_tag_failed(self, mock_logger_info):
197 with mock.patch.dict(os.environ,
200 self.assertEqual(functest_utils.get_build_tag(),
202 mock_logger_info.assert_called_once_with("Impossible to retrieve"
205 def test_get_build_tag_default(self):
206 with mock.patch.dict(os.environ,
207 {'BUILD_TAG': self.build_tag},
209 self.assertEqual(functest_utils.get_build_tag(),
212 def test_get_db_url_env_var(self):
213 with mock.patch.dict(os.environ,
214 {'TEST_DB_URL': self.db_url_env,
215 'CONFIG_FUNCTEST_YAML':
216 "./functest/ci/config_functest.yaml"},
218 self.assertEqual(functest_utils.get_db_url(),
221 @mock.patch('functest.utils.functest_utils.get_functest_config')
222 def test_get_db_url_default(self, mock_get_functest_config):
223 mock_get_functest_config.return_value = self.db_url
224 self.assertEqual(functest_utils.get_db_url(), self.db_url)
225 mock_get_functest_config.assert_called_once_with('results.test_db_url')
227 @mock.patch('functest.utils.functest_utils.logger.info')
228 def test_logger_test_results(self, mock_logger_info):
229 with mock.patch('functest.utils.functest_utils.get_pod_name',
230 return_value=self.node_name), \
231 mock.patch('functest.utils.functest_utils.get_scenario',
232 return_value=self.scenario), \
233 mock.patch('functest.utils.functest_utils.get_version',
234 return_value=self.version), \
235 mock.patch('functest.utils.functest_utils.get_build_tag',
236 return_value=self.build_tag), \
237 mock.patch('functest.utils.functest_utils.get_db_url',
238 return_value=self.db_url):
239 functest_utils.logger_test_results(self.project, self.case_name,
240 self.status, self.details)
241 mock_logger_info.assert_called_once_with(
243 "****************************************\n"
244 "\t %(p)s/%(n)s results \n\n"
245 "****************************************\n"
251 "build tag:\t%(b)s\n"
253 % {'p': self.project,
256 'pod': self.node_name,
263 def _get_env_dict(self, var):
264 dic = {'INSTALLER_TYPE': self.installer,
265 'DEPLOY_SCENARIO': self.scenario,
266 'NODE_NAME': self.node_name,
267 'BUILD_TAG': self.build_tag}
271 def _test_push_results_to_db_missing_env(self, env_var):
272 dic = self._get_env_dict(env_var)
273 with mock.patch('functest.utils.functest_utils.get_db_url',
274 return_value=self.db_url), \
275 mock.patch.dict(os.environ,
278 mock.patch('functest.utils.functest_utils.logger.error') \
279 as mock_logger_error:
280 functest_utils.push_results_to_db(self.project, self.case_name,
281 self.start_date, self.stop_date,
282 self.result, self.details)
283 mock_logger_error.assert_called_once_with("Please set env var: " +
287 def test_push_results_to_db_missing_installer(self):
288 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
290 def test_push_results_to_db_missing_scenario(self):
291 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
293 def test_push_results_to_db_missing_nodename(self):
294 self._test_push_results_to_db_missing_env('NODE_NAME')
296 def test_push_results_to_db_missing_buildtag(self):
297 self._test_push_results_to_db_missing_env('BUILD_TAG')
299 def test_push_results_to_db_request_post_failed(self):
300 dic = self._get_env_dict(None)
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 mock.patch('functest.utils.functest_utils.requests.post',
309 side_effect=requests.RequestException):
310 self.assertFalse(functest_utils.
311 push_results_to_db(self.project, self.case_name,
314 self.result, self.details))
315 mock_logger_error.assert_called_once_with(test_utils.
316 RegexMatch("Pushing "
322 def test_push_results_to_db_request_post_exception(self):
323 dic = self._get_env_dict(None)
324 with mock.patch('functest.utils.functest_utils.get_db_url',
325 return_value=self.db_url), \
326 mock.patch.dict(os.environ,
329 mock.patch('functest.utils.functest_utils.logger.error') \
330 as mock_logger_error, \
331 mock.patch('functest.utils.functest_utils.requests.post',
332 side_effect=Exception):
333 self.assertFalse(functest_utils.
334 push_results_to_db(self.project, self.case_name,
337 self.result, self.details))
338 self.assertTrue(mock_logger_error.called)
340 def test_push_results_to_db_default(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.requests.post'):
348 self.assertTrue(functest_utils.
349 push_results_to_db(self.project, self.case_name,
352 self.result, self.details))
354 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
358 if FunctestUtilsTesting.readline == \
359 len(FunctestUtilsTesting.test_ip) - 1:
361 FunctestUtilsTesting.readline += 1
362 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
364 # TODO: get_resolvconf_ns
365 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
366 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
367 attrs = {'query.return_value': ["test"]}
368 mock_dns_resolve.configure_mock(**attrs)
371 attrs = {'readline.side_effect': self.readline_side}
372 m.configure_mock(**attrs)
374 with mock.patch("six.moves.builtins.open") as mo:
376 self.assertEqual(functest_utils.get_resolvconf_ns(),
379 def _get_environ(self, var):
380 if var == 'INSTALLER_TYPE':
381 return self.installer
382 elif var == 'DEPLOY_SCENARIO':
386 def test_get_ci_envvars_default(self):
387 with mock.patch('os.environ.get',
388 side_effect=self._get_environ):
389 dic = {"installer": self.installer,
390 "scenario": self.scenario}
391 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
393 def cmd_readline(self):
394 return 'test_value\n'
396 @mock.patch('functest.utils.functest_utils.logger.error')
397 @mock.patch('functest.utils.functest_utils.logger.info')
398 def test_execute_command_args_present_with_error(self, mock_logger_info,
400 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
401 as mock_subproc_open, \
402 mock.patch('six.moves.builtins.open',
403 mock.mock_open()) as mopen:
405 FunctestUtilsTesting.readline = 0
407 mock_obj = mock.Mock()
408 attrs = {'readline.side_effect': self.cmd_readline()}
409 mock_obj.configure_mock(**attrs)
411 mock_obj2 = mock.Mock()
412 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
413 mock_obj2.configure_mock(**attrs)
415 mock_subproc_open.return_value = mock_obj2
417 resp = functest_utils.execute_command(self.cmd, info=True,
418 error_msg=self.error_msg,
420 output_file=self.output_file)
421 self.assertEqual(resp, 1)
422 msg_exec = ("Executing command: '%s'" % self.cmd)
423 mock_logger_info.assert_called_once_with(msg_exec)
424 mopen.assert_called_once_with(self.output_file, "w")
425 mock_logger_error.assert_called_once_with(self.error_msg)
427 @mock.patch('functest.utils.functest_utils.logger.info')
428 def test_execute_command_args_present_with_success(self, mock_logger_info,
430 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
431 as mock_subproc_open, \
432 mock.patch('six.moves.builtins.open',
433 mock.mock_open()) as mopen:
435 FunctestUtilsTesting.readline = 0
437 mock_obj = mock.Mock()
438 attrs = {'readline.side_effect': self.cmd_readline()}
439 mock_obj.configure_mock(**attrs)
441 mock_obj2 = mock.Mock()
442 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
443 mock_obj2.configure_mock(**attrs)
445 mock_subproc_open.return_value = mock_obj2
447 resp = functest_utils.execute_command(self.cmd, info=True,
448 error_msg=self.error_msg,
450 output_file=self.output_file)
451 self.assertEqual(resp, 0)
452 msg_exec = ("Executing command: '%s'" % self.cmd)
453 mock_logger_info.assert_called_once_with(msg_exec)
454 mopen.assert_called_once_with(self.output_file, "w")
456 @mock.patch('functest.utils.functest_utils.logger.info')
457 def test_execute_command_args_missing_with_success(self, mock_logger_info,
459 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
460 as mock_subproc_open:
462 FunctestUtilsTesting.readline = 2
464 mock_obj = mock.Mock()
465 attrs = {'readline.side_effect': self.cmd_readline()}
466 mock_obj.configure_mock(**attrs)
468 mock_obj2 = mock.Mock()
469 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
470 mock_obj2.configure_mock(**attrs)
472 mock_subproc_open.return_value = mock_obj2
474 resp = functest_utils.execute_command(self.cmd, info=False,
478 self.assertEqual(resp, 0)
480 @mock.patch('functest.utils.functest_utils.logger.error')
481 def test_execute_command_args_missing_with_error(self, mock_logger_error,
483 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
484 as mock_subproc_open:
486 FunctestUtilsTesting.readline = 2
487 mock_obj = mock.Mock()
488 attrs = {'readline.side_effect': self.cmd_readline()}
489 mock_obj.configure_mock(**attrs)
491 mock_obj2 = mock.Mock()
492 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
493 mock_obj2.configure_mock(**attrs)
495 mock_subproc_open.return_value = mock_obj2
497 resp = functest_utils.execute_command(self.cmd, info=False,
501 self.assertEqual(resp, 1)
503 def _get_functest_config(self, var):
506 @mock.patch('functest.utils.functest_utils.logger.error')
507 def test_get_dict_by_test(self, mock_logger_error):
508 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
509 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
511 mock.patch('functest.utils.functest_utils.get_testcases_'
513 mock_obj = mock.Mock()
514 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
515 mock_obj.configure_mock(**attrs)
517 mock_yaml.return_value = mock_obj
519 self.assertDictEqual(functest_utils.
520 get_dict_by_test(self.testname),
523 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
524 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
525 mock_get_dict_by_test.return_value = self.testcase_dict
526 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
529 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
530 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
531 mock_get_dict_by_test.return_value = None
532 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
534 def test_get_parameter_from_yaml_failed(self):
535 self.file_yaml['general'] = None
536 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
537 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
539 self.assertRaises(ValueError) as excep:
540 mock_yaml.return_value = self.file_yaml
541 functest_utils.get_parameter_from_yaml(self.parameter,
543 self.assertTrue(("The parameter %s is not"
544 " defined in config_functest.yaml" %
545 self.parameter) in excep.exception)
547 def test_get_parameter_from_yaml_default(self):
548 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
549 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
551 mock_yaml.return_value = self.file_yaml
552 self.assertEqual(functest_utils.
553 get_parameter_from_yaml(self.parameter,
557 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
558 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
559 with mock.patch.dict(os.environ,
560 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
561 functest_utils.get_functest_config(self.parameter)
562 mock_get_parameter_from_yaml. \
563 assert_called_once_with(self.parameter,
566 def test_check_success_rate_default(self):
567 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
569 mock_criteria.return_value = self.criteria
570 resp = functest_utils.check_success_rate(self.case_name,
572 self.assertEqual(resp, 'PASS')
574 def test_check_success_rate_failed(self):
575 with mock.patch('functest.utils.functest_utils.get_criteria_by_test') \
577 mock_criteria.return_value = self.criteria
578 resp = functest_utils.check_success_rate(self.case_name,
580 self.assertEqual(resp, 'FAIL')
584 def test_get_testcases_file_dir(self):
585 resp = functest_utils.get_testcases_file_dir()
586 self.assertEqual(resp,
587 "/home/opnfv/repos/functest/"
588 "functest/ci/testcases.yaml")
590 def test_get_functest_yaml(self):
591 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
592 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
594 mock_yaml.return_value = self.file_yaml
595 resp = functest_utils.get_functest_yaml()
596 self.assertEqual(resp, self.file_yaml)
598 @mock.patch('functest.utils.functest_utils.logger.info')
599 def test_print_separator(self, mock_logger_info):
600 functest_utils.print_separator()
601 mock_logger_info.assert_called_once_with("======================="
602 "=======================")
605 if __name__ == "__main__":
606 unittest.main(verbosity=2)