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
22 from functest.utils.constants import CONST
25 class FunctestUtilsTesting(unittest.TestCase):
28 self.url = 'http://www.opnfv.org/'
30 self.dest_path = 'test_path'
31 self.repo_path = 'test_repo_path'
32 self.installer = 'test_installer'
33 self.scenario = 'test_scenario'
34 self.build_tag = 'jenkins-functest-fuel-opnfv-jump-2-daily-master-190'
35 self.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
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'
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 = {'case_name': 'testname',
57 'criteria': self.criteria}
58 self.parameter = 'general.openstack.image_name'
59 self.config_yaml = os.path.normpath(os.path.join(os.path.dirname(
60 os.path.abspath(__file__)), '../../../ci/config_functest.yaml'))
61 self.db_url_env = 'http://foo/testdb'
62 self.testcases_yaml = "test_testcases_yaml"
63 self.file_yaml = {'general': {'openstack': {'image_name':
66 @mock.patch('six.moves.urllib.request.urlopen',
67 side_effect=urllib.error.URLError('no host given'))
68 def test_check_internet_connectivity_failed(self, mock_method):
69 self.assertFalse(functest_utils.check_internet_connectivity())
70 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
72 @mock.patch('six.moves.urllib.request.urlopen')
73 def test_check_internet_connectivity_default(self, mock_method):
74 self.assertTrue(functest_utils.check_internet_connectivity())
75 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
77 @mock.patch('six.moves.urllib.request.urlopen')
78 def test_check_internet_connectivity_debian(self, mock_method):
79 self.url = "https://www.debian.org/"
80 self.assertTrue(functest_utils.check_internet_connectivity(self.url))
81 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
83 @mock.patch('six.moves.urllib.request.urlopen',
84 side_effect=urllib.error.URLError('no host given'))
85 def test_download_url_failed(self, mock_url):
86 self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
88 @mock.patch('six.moves.urllib.request.urlopen')
89 def test_download_url_default(self, mock_url):
90 with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
91 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
93 name = self.url.rsplit('/')[-1]
94 dest = self.dest_path + "/" + name
95 self.assertTrue(functest_utils.download_url(self.url,
97 m.assert_called_once_with(dest, 'wb')
98 self.assertTrue(mock_sh.called)
100 def test_get_git_branch(self):
101 with mock.patch('functest.utils.functest_utils.Repo') as mock_repo:
102 mock_obj2 = mock.Mock()
103 attrs = {'name': 'test_branch'}
104 mock_obj2.configure_mock(**attrs)
106 mock_obj = mock.Mock()
107 attrs = {'active_branch': mock_obj2}
108 mock_obj.configure_mock(**attrs)
110 mock_repo.return_value = mock_obj
111 self.assertEqual(functest_utils.get_git_branch(self.repo_path),
114 @mock.patch('functest.utils.functest_utils.Repo',
115 side_effect=NoSuchPathError)
116 def test_get_git_branch_failed(self, mock_repo):
117 self.assertRaises(NoSuchPathError,
118 lambda: functest_utils.get_git_branch(self.repo_path
121 @mock.patch('functest.utils.functest_utils.logger.error')
122 def test_get_installer_type_failed(self, mock_logger_error):
123 with mock.patch.dict(os.environ,
126 self.assertEqual(functest_utils.get_installer_type(),
128 mock_logger_error.assert_called_once_with("Impossible to retrieve"
129 " the installer type")
131 def test_get_installer_type_default(self):
132 with mock.patch.dict(os.environ,
133 {'INSTALLER_TYPE': 'test_installer'},
135 self.assertEqual(functest_utils.get_installer_type(),
138 @mock.patch('functest.utils.functest_utils.logger.info')
139 def test_get_scenario_failed(self, mock_logger_info):
140 with mock.patch.dict(os.environ,
143 self.assertEqual(functest_utils.get_scenario(),
144 "os-nosdn-nofeature-noha")
145 mock_logger_info.assert_called_once_with("Impossible to retrieve "
148 "os-nosdn-nofeature-noha")
150 def test_get_scenario_default(self):
151 with mock.patch.dict(os.environ,
152 {'DEPLOY_SCENARIO': 'test_scenario'},
154 self.assertEqual(functest_utils.get_scenario(),
157 @mock.patch('functest.utils.functest_utils.get_build_tag')
158 def test_get_version_daily_job(self, mock_get_build_tag):
159 mock_get_build_tag.return_value = self.build_tag
160 self.assertEqual(functest_utils.get_version(), self.version)
162 @mock.patch('functest.utils.functest_utils.get_build_tag')
163 def test_get_version_weekly_job(self, mock_get_build_tag):
164 mock_get_build_tag.return_value = self.build_tag_week
165 self.assertEqual(functest_utils.get_version(), self.version)
167 @mock.patch('functest.utils.functest_utils.get_build_tag')
168 def test_get_version_with_dummy_build_tag(self, mock_get_build_tag):
169 mock_get_build_tag.return_value = 'whatever'
170 self.assertEqual(functest_utils.get_version(), 'unknown')
172 @mock.patch('functest.utils.functest_utils.get_build_tag')
173 def test_get_version_unknown(self, mock_get_build_tag):
174 mock_get_build_tag.return_value = "unknown_build_tag"
175 self.assertEqual(functest_utils.get_version(), "unknown")
177 @mock.patch('functest.utils.functest_utils.logger.info')
178 def test_get_pod_name_failed(self, mock_logger_info):
179 with mock.patch.dict(os.environ,
182 self.assertEqual(functest_utils.get_pod_name(),
184 mock_logger_info.assert_called_once_with("Unable to retrieve "
186 "environment. Using "
187 "pod name 'unknown-pod'")
189 def test_get_pod_name_default(self):
190 with mock.patch.dict(os.environ,
191 {'NODE_NAME': 'test_node_name'},
193 self.assertEqual(functest_utils.get_pod_name(),
196 @mock.patch('functest.utils.functest_utils.logger.info')
197 def test_get_build_tag_failed(self, mock_logger_info):
198 with mock.patch.dict(os.environ,
201 self.assertEqual(functest_utils.get_build_tag(),
203 mock_logger_info.assert_called_once_with("Impossible to retrieve"
206 def test_get_build_tag_default(self):
207 with mock.patch.dict(os.environ,
208 {'BUILD_TAG': self.build_tag},
210 self.assertEqual(functest_utils.get_build_tag(),
213 @mock.patch('functest.utils.functest_utils.logger.info')
214 def test_logger_test_results(self, mock_logger_info):
215 CONST.__setattr__('results_test_db_url', self.db_url)
216 with mock.patch('functest.utils.functest_utils.get_pod_name',
217 return_value=self.node_name), \
218 mock.patch('functest.utils.functest_utils.get_scenario',
219 return_value=self.scenario), \
220 mock.patch('functest.utils.functest_utils.get_version',
221 return_value=self.version), \
222 mock.patch('functest.utils.functest_utils.get_build_tag',
223 return_value=self.build_tag):
224 functest_utils.logger_test_results(self.project, self.case_name,
225 self.status, self.details)
226 mock_logger_info.assert_called_once_with(
228 "****************************************\n"
229 "\t %(p)s/%(n)s results \n\n"
230 "****************************************\n"
236 "build tag:\t%(b)s\n"
238 % {'p': self.project,
240 'db': CONST.__getattribute__('results_test_db_url'),
241 'pod': self.node_name,
248 def _get_env_dict(self, var):
249 dic = {'INSTALLER_TYPE': self.installer,
250 'DEPLOY_SCENARIO': self.scenario,
251 'NODE_NAME': self.node_name,
252 'BUILD_TAG': self.build_tag}
256 def _test_push_results_to_db_missing_env(self, env_var):
257 dic = self._get_env_dict(env_var)
258 CONST.__setattr__('results_test_db_url', self.db_url)
259 with mock.patch.dict(os.environ,
262 mock.patch('functest.utils.functest_utils.logger.error') \
263 as mock_logger_error:
264 functest_utils.push_results_to_db(self.project, self.case_name,
265 self.start_date, self.stop_date,
266 self.result, self.details)
267 mock_logger_error.assert_called_once_with("Please set env var: " +
271 def test_push_results_to_db_missing_installer(self):
272 self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
274 def test_push_results_to_db_missing_scenario(self):
275 self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
277 def test_push_results_to_db_missing_nodename(self):
278 self._test_push_results_to_db_missing_env('NODE_NAME')
280 def test_push_results_to_db_missing_buildtag(self):
281 self._test_push_results_to_db_missing_env('BUILD_TAG')
283 def test_push_results_to_db_request_post_failed(self):
284 dic = self._get_env_dict(None)
285 CONST.__setattr__('results_test_db_url', self.db_url)
286 with mock.patch.dict(os.environ,
289 mock.patch('functest.utils.functest_utils.logger.error') \
290 as mock_logger_error, \
291 mock.patch('functest.utils.functest_utils.requests.post',
292 side_effect=requests.RequestException):
293 self.assertFalse(functest_utils.
294 push_results_to_db(self.project, self.case_name,
297 self.result, self.details))
298 mock_logger_error.assert_called_once_with(test_utils.
299 RegexMatch("Pushing "
305 def test_push_results_to_db_request_post_exception(self):
306 dic = self._get_env_dict(None)
307 CONST.__setattr__('results_test_db_url', self.db_url)
308 with mock.patch.dict(os.environ,
311 mock.patch('functest.utils.functest_utils.logger.error') \
312 as mock_logger_error, \
313 mock.patch('functest.utils.functest_utils.requests.post',
314 side_effect=Exception):
315 self.assertFalse(functest_utils.
316 push_results_to_db(self.project, self.case_name,
319 self.result, self.details))
320 self.assertTrue(mock_logger_error.called)
322 def test_push_results_to_db_default(self):
323 dic = self._get_env_dict(None)
324 CONST.__setattr__('results_test_db_url', self.db_url)
325 with mock.patch.dict(os.environ,
328 mock.patch('functest.utils.functest_utils.requests.post'):
329 self.assertTrue(functest_utils.
330 push_results_to_db(self.project, self.case_name,
333 self.result, self.details))
335 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
339 if FunctestUtilsTesting.readline == \
340 len(FunctestUtilsTesting.test_ip) - 1:
342 FunctestUtilsTesting.readline += 1
343 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
345 # TODO: get_resolvconf_ns
346 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
347 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
348 attrs = {'query.return_value': ["test"]}
349 mock_dns_resolve.configure_mock(**attrs)
352 attrs = {'readline.side_effect': self.readline_side}
353 m.configure_mock(**attrs)
355 with mock.patch("six.moves.builtins.open") as mo:
357 self.assertEqual(functest_utils.get_resolvconf_ns(),
360 def _get_environ(self, var):
361 if var == 'INSTALLER_TYPE':
362 return self.installer
363 elif var == 'DEPLOY_SCENARIO':
367 def test_get_ci_envvars_default(self):
368 with mock.patch('os.environ.get',
369 side_effect=self._get_environ):
370 dic = {"installer": self.installer,
371 "scenario": self.scenario}
372 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
374 def cmd_readline(self):
375 return 'test_value\n'
377 @mock.patch('functest.utils.functest_utils.logger.error')
378 @mock.patch('functest.utils.functest_utils.logger.info')
379 def test_execute_command_args_present_with_error(self, mock_logger_info,
381 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
382 as mock_subproc_open, \
383 mock.patch('six.moves.builtins.open',
384 mock.mock_open()) as mopen:
386 FunctestUtilsTesting.readline = 0
388 mock_obj = mock.Mock()
389 attrs = {'readline.side_effect': self.cmd_readline()}
390 mock_obj.configure_mock(**attrs)
392 mock_obj2 = mock.Mock()
393 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
394 mock_obj2.configure_mock(**attrs)
396 mock_subproc_open.return_value = mock_obj2
398 resp = functest_utils.execute_command(self.cmd, info=True,
399 error_msg=self.error_msg,
401 output_file=self.output_file)
402 self.assertEqual(resp, 1)
403 msg_exec = ("Executing command: '%s'" % self.cmd)
404 mock_logger_info.assert_called_once_with(msg_exec)
405 mopen.assert_called_once_with(self.output_file, "w")
406 mock_logger_error.assert_called_once_with(self.error_msg)
408 @mock.patch('functest.utils.functest_utils.logger.info')
409 def test_execute_command_args_present_with_success(self, mock_logger_info,
411 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
412 as mock_subproc_open, \
413 mock.patch('six.moves.builtins.open',
414 mock.mock_open()) as mopen:
416 FunctestUtilsTesting.readline = 0
418 mock_obj = mock.Mock()
419 attrs = {'readline.side_effect': self.cmd_readline()}
420 mock_obj.configure_mock(**attrs)
422 mock_obj2 = mock.Mock()
423 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
424 mock_obj2.configure_mock(**attrs)
426 mock_subproc_open.return_value = mock_obj2
428 resp = functest_utils.execute_command(self.cmd, info=True,
429 error_msg=self.error_msg,
431 output_file=self.output_file)
432 self.assertEqual(resp, 0)
433 msg_exec = ("Executing command: '%s'" % self.cmd)
434 mock_logger_info.assert_called_once_with(msg_exec)
435 mopen.assert_called_once_with(self.output_file, "w")
437 @mock.patch('sys.stdout')
438 def test_execute_command_args_missing_with_success(self, stdout=None):
439 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
440 as mock_subproc_open:
442 FunctestUtilsTesting.readline = 2
444 mock_obj = mock.Mock()
445 attrs = {'readline.side_effect': self.cmd_readline()}
446 mock_obj.configure_mock(**attrs)
448 mock_obj2 = mock.Mock()
449 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
450 mock_obj2.configure_mock(**attrs)
452 mock_subproc_open.return_value = mock_obj2
454 resp = functest_utils.execute_command(self.cmd, info=False,
458 self.assertEqual(resp, 0)
460 @mock.patch('sys.stdout')
461 def test_execute_command_args_missing_with_error(self, stdout=None):
462 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
463 as mock_subproc_open:
465 FunctestUtilsTesting.readline = 2
466 mock_obj = mock.Mock()
467 attrs = {'readline.side_effect': self.cmd_readline()}
468 mock_obj.configure_mock(**attrs)
470 mock_obj2 = mock.Mock()
471 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
472 mock_obj2.configure_mock(**attrs)
474 mock_subproc_open.return_value = mock_obj2
476 resp = functest_utils.execute_command(self.cmd, info=False,
480 self.assertEqual(resp, 1)
482 def _get_functest_config(self, var):
485 @mock.patch('functest.utils.functest_utils.logger.error')
486 def test_get_dict_by_test(self, mock_logger_error):
487 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
488 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
490 mock.patch('functest.utils.functest_utils.get_testcases_'
492 mock_obj = mock.Mock()
493 attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
494 mock_obj.configure_mock(**attrs)
496 mock_yaml.return_value = mock_obj
498 self.assertDictEqual(functest_utils.
499 get_dict_by_test(self.testname),
502 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
503 def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
504 mock_get_dict_by_test.return_value = self.testcase_dict
505 self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
508 @mock.patch('functest.utils.functest_utils.get_dict_by_test')
509 def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
510 mock_get_dict_by_test.return_value = None
511 self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
513 def test_get_parameter_from_yaml_failed(self):
514 self.file_yaml['general'] = None
515 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
516 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
518 self.assertRaises(ValueError) as excep:
519 mock_yaml.return_value = self.file_yaml
520 functest_utils.get_parameter_from_yaml(self.parameter,
522 self.assertTrue(("The parameter %s is not"
523 " defined in config_functest.yaml" %
524 self.parameter) in excep.exception)
526 def test_get_parameter_from_yaml_default(self):
527 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
528 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
530 mock_yaml.return_value = self.file_yaml
531 self.assertEqual(functest_utils.
532 get_parameter_from_yaml(self.parameter,
536 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
537 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
538 with mock.patch.dict(os.environ,
539 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
540 functest_utils.get_functest_config(self.parameter)
541 mock_get_parameter_from_yaml. \
542 assert_called_once_with(self.parameter,
547 @mock.patch('functest.utils.functest_utils.get_functest_config')
548 def test_get_testcases_file_dir(self, mock_get_functest_config):
549 mock_get_functest_config.return_value = self.testcases_yaml
550 resp = functest_utils.get_testcases_file_dir()
551 self.assertEqual(resp, self.testcases_yaml)
552 mock_get_functest_config.assert_called_once_with(
553 'general.functest.testcases_yaml')
555 def test_get_functest_yaml(self):
556 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
557 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
559 mock_yaml.return_value = self.file_yaml
560 resp = functest_utils.get_functest_yaml()
561 self.assertEqual(resp, self.file_yaml)
563 @mock.patch('functest.utils.functest_utils.logger.info')
564 def test_print_separator(self, mock_logger_info):
565 functest_utils.print_separator()
566 mock_logger_info.assert_called_once_with("======================="
567 "=======================")
570 if __name__ == "__main__":
571 logging.disable(logging.CRITICAL)
572 unittest.main(verbosity=2)