Merge "Replace CONST.* by getattribute/setattr for rally and tempest"
[functest.git] / functest / tests / unit / utils / test_functest_utils.py
1 #!/usr/bin/env python
2
3 # Copyright (c) 2016 Orange and others.
4 #
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
9
10 import logging
11 import os
12 import time
13 import unittest
14
15 from git.exc import NoSuchPathError
16 import mock
17 import requests
18 from six.moves import urllib
19
20 from functest.tests.unit import test_utils
21 from functest.utils import functest_utils
22
23
24 class FunctestUtilsTesting(unittest.TestCase):
25
26     def setUp(self):
27         self.url = 'http://www.opnfv.org/'
28         self.timeout = 5
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'
42         self.criteria = 50
43         self.result = 75
44         self.start_date = 1482624000
45         self.stop_date = 1482624000
46         self.start_time = time.time()
47         self.stop_time = time.time()
48         self.readline = -1
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'
52         self.cmd = 'test_cmd'
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':
61                                                     'test_image_name'}}}
62
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)
68
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)
73
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)
79
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))
84
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')\
89                 as mock_sh:
90             name = self.url.rsplit('/')[-1]
91             dest = self.dest_path + "/" + name
92             self.assertTrue(functest_utils.download_url(self.url,
93                                                         self.dest_path))
94             m.assert_called_once_with(dest, 'wb')
95             self.assertTrue(mock_sh.called)
96
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)
102
103             mock_obj = mock.Mock()
104             attrs = {'active_branch': mock_obj2}
105             mock_obj.configure_mock(**attrs)
106
107             mock_repo.return_value = mock_obj
108             self.assertEqual(functest_utils.get_git_branch(self.repo_path),
109                              'test_branch')
110
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
116                                                                 ))
117
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,
121                              {},
122                              clear=True):
123             self.assertEqual(functest_utils.get_installer_type(),
124                              "Unknown_installer")
125             mock_logger_error.assert_called_once_with("Impossible to retrieve"
126                                                       " the installer type")
127
128     def test_get_installer_type_default(self):
129         with mock.patch.dict(os.environ,
130                              {'INSTALLER_TYPE': 'test_installer'},
131                              clear=True):
132             self.assertEqual(functest_utils.get_installer_type(),
133                              self.installer)
134
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,
138                              {},
139                              clear=True):
140             self.assertEqual(functest_utils.get_scenario(),
141                              "os-nosdn-nofeature-noha")
142             mock_logger_info.assert_called_once_with("Impossible to retrieve "
143                                                      "the scenario.Use "
144                                                      "default "
145                                                      "os-nosdn-nofeature-noha")
146
147     def test_get_scenario_default(self):
148         with mock.patch.dict(os.environ,
149                              {'DEPLOY_SCENARIO': 'test_scenario'},
150                              clear=True):
151             self.assertEqual(functest_utils.get_scenario(),
152                              self.scenario)
153
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)
158
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)
163
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')
168
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")
173
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,
177                              {},
178                              clear=True):
179             self.assertEqual(functest_utils.get_pod_name(),
180                              "unknown-pod")
181             mock_logger_info.assert_called_once_with("Unable to retrieve "
182                                                      "the POD name from "
183                                                      "environment. Using "
184                                                      "pod name 'unknown-pod'")
185
186     def test_get_pod_name_default(self):
187         with mock.patch.dict(os.environ,
188                              {'NODE_NAME': 'test_node_name'},
189                              clear=True):
190             self.assertEqual(functest_utils.get_pod_name(),
191                              self.node_name)
192
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,
196                              {},
197                              clear=True):
198             self.assertEqual(functest_utils.get_build_tag(),
199                              "none")
200             mock_logger_info.assert_called_once_with("Impossible to retrieve"
201                                                      " the build tag")
202
203     def test_get_build_tag_default(self):
204         with mock.patch.dict(os.environ,
205                              {'BUILD_TAG': self.build_tag},
206                              clear=True):
207             self.assertEqual(functest_utils.get_build_tag(),
208                              self.build_tag)
209
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"},
215                              clear=True):
216             self.assertEqual(functest_utils.get_db_url(),
217                              self.db_url_env)
218
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')
224
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(
240                 "\n"
241                 "****************************************\n"
242                 "\t %(p)s/%(n)s results \n\n"
243                 "****************************************\n"
244                 "DB:\t%(db)s\n"
245                 "pod:\t%(pod)s\n"
246                 "version:\t%(v)s\n"
247                 "scenario:\t%(s)s\n"
248                 "status:\t%(c)s\n"
249                 "build tag:\t%(b)s\n"
250                 "details:\t%(d)s\n"
251                 % {'p': self.project,
252                     'n': self.case_name,
253                     'db': self.db_url,
254                     'pod': self.node_name,
255                     'v': self.version,
256                     's': self.scenario,
257                     'c': self.status,
258                     'b': self.build_tag,
259                     'd': self.details})
260
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}
266         dic.pop(var, None)
267         return dic
268
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,
274                                 dic,
275                                 clear=True), \
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: " +
282                                                       str("\'" + env_var +
283                                                           "\'"))
284
285     def test_push_results_to_db_missing_installer(self):
286         self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
287
288     def test_push_results_to_db_missing_scenario(self):
289         self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
290
291     def test_push_results_to_db_missing_nodename(self):
292         self._test_push_results_to_db_missing_env('NODE_NAME')
293
294     def test_push_results_to_db_missing_buildtag(self):
295         self._test_push_results_to_db_missing_env('BUILD_TAG')
296
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,
302                                 dic,
303                                 clear=True), \
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,
310                                                 self.start_date,
311                                                 self.stop_date,
312                                                 self.result, self.details))
313             mock_logger_error.assert_called_once_with(test_utils.
314                                                       RegexMatch("Pushing "
315                                                                  "Result to"
316                                                                  " DB"
317                                                                  "(\S+\s*) "
318                                                                  "failed:"))
319
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,
325                                 dic,
326                                 clear=True), \
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,
333                                                 self.start_date,
334                                                 self.stop_date,
335                                                 self.result, self.details))
336             self.assertTrue(mock_logger_error.called)
337
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,
343                                 dic,
344                                 clear=True), \
345                 mock.patch('functest.utils.functest_utils.requests.post'):
346             self.assertTrue(functest_utils.
347                             push_results_to_db(self.project, self.case_name,
348                                                self.start_date,
349                                                self.stop_date,
350                                                self.result, self.details))
351     readline = 0
352     test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
353
354     @staticmethod
355     def readline_side():
356         if FunctestUtilsTesting.readline == \
357                 len(FunctestUtilsTesting.test_ip) - 1:
358             return False
359         FunctestUtilsTesting.readline += 1
360         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
361
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)
367
368         m = mock.Mock()
369         attrs = {'readline.side_effect': self.readline_side}
370         m.configure_mock(**attrs)
371
372         with mock.patch("six.moves.builtins.open") as mo:
373             mo.return_value = m
374             self.assertEqual(functest_utils.get_resolvconf_ns(),
375                              self.test_ip[1:])
376
377     def _get_environ(self, var):
378         if var == 'INSTALLER_TYPE':
379             return self.installer
380         elif var == 'DEPLOY_SCENARIO':
381             return self.scenario
382         return var
383
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)
390
391     def cmd_readline(self):
392         return 'test_value\n'
393
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,
397                                                      mock_logger_error):
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:
402
403             FunctestUtilsTesting.readline = 0
404
405             mock_obj = mock.Mock()
406             attrs = {'readline.side_effect': self.cmd_readline()}
407             mock_obj.configure_mock(**attrs)
408
409             mock_obj2 = mock.Mock()
410             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
411             mock_obj2.configure_mock(**attrs)
412
413             mock_subproc_open.return_value = mock_obj2
414
415             resp = functest_utils.execute_command(self.cmd, info=True,
416                                                   error_msg=self.error_msg,
417                                                   verbose=True,
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)
424
425     @mock.patch('functest.utils.functest_utils.logger.info')
426     def test_execute_command_args_present_with_success(self, mock_logger_info,
427                                                        ):
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:
432
433             FunctestUtilsTesting.readline = 0
434
435             mock_obj = mock.Mock()
436             attrs = {'readline.side_effect': self.cmd_readline()}
437             mock_obj.configure_mock(**attrs)
438
439             mock_obj2 = mock.Mock()
440             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
441             mock_obj2.configure_mock(**attrs)
442
443             mock_subproc_open.return_value = mock_obj2
444
445             resp = functest_utils.execute_command(self.cmd, info=True,
446                                                   error_msg=self.error_msg,
447                                                   verbose=True,
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")
453
454     @mock.patch('functest.utils.functest_utils.logger.info')
455     def test_execute_command_args_missing_with_success(self, mock_logger_info,
456                                                        ):
457         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
458                 as mock_subproc_open:
459
460             FunctestUtilsTesting.readline = 2
461
462             mock_obj = mock.Mock()
463             attrs = {'readline.side_effect': self.cmd_readline()}
464             mock_obj.configure_mock(**attrs)
465
466             mock_obj2 = mock.Mock()
467             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
468             mock_obj2.configure_mock(**attrs)
469
470             mock_subproc_open.return_value = mock_obj2
471
472             resp = functest_utils.execute_command(self.cmd, info=False,
473                                                   error_msg="",
474                                                   verbose=False,
475                                                   output_file=None)
476             self.assertEqual(resp, 0)
477
478     @mock.patch('functest.utils.functest_utils.logger.error')
479     def test_execute_command_args_missing_with_error(self, mock_logger_error,
480                                                      ):
481         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
482                 as mock_subproc_open:
483
484             FunctestUtilsTesting.readline = 2
485             mock_obj = mock.Mock()
486             attrs = {'readline.side_effect': self.cmd_readline()}
487             mock_obj.configure_mock(**attrs)
488
489             mock_obj2 = mock.Mock()
490             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
491             mock_obj2.configure_mock(**attrs)
492
493             mock_subproc_open.return_value = mock_obj2
494
495             resp = functest_utils.execute_command(self.cmd, info=False,
496                                                   error_msg="",
497                                                   verbose=False,
498                                                   output_file=None)
499             self.assertEqual(resp, 1)
500
501     def _get_functest_config(self, var):
502         return var
503
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') \
508                 as mock_yaml, \
509                 mock.patch('functest.utils.functest_utils.get_testcases_'
510                            'file_dir'):
511             mock_obj = mock.Mock()
512             attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
513             mock_obj.configure_mock(**attrs)
514
515             mock_yaml.return_value = mock_obj
516
517             self.assertDictEqual(functest_utils.
518                                  get_dict_by_test(self.testname),
519                                  self.testcase_dict)
520
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),
525                          self.criteria)
526
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))
531
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') \
536                 as mock_yaml, \
537                 self.assertRaises(ValueError) as excep:
538             mock_yaml.return_value = self.file_yaml
539             functest_utils.get_parameter_from_yaml(self.parameter,
540                                                    self.test_file)
541             self.assertTrue(("The parameter %s is not"
542                              " defined in config_functest.yaml" %
543                              self.parameter) in excep.exception)
544
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') \
548                 as mock_yaml:
549             mock_yaml.return_value = self.file_yaml
550             self.assertEqual(functest_utils.
551                              get_parameter_from_yaml(self.parameter,
552                                                      self.test_file),
553                              'test_image_name')
554
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,
562                                         self.config_yaml)
563
564     # TODO: merge_dicts
565
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")
571
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') \
575                 as mock_yaml:
576             mock_yaml.return_value = self.file_yaml
577             resp = functest_utils.get_functest_yaml()
578             self.assertEqual(resp, self.file_yaml)
579
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                                                  "=======================")
585
586
587 if __name__ == "__main__":
588     logging.disable(logging.CRITICAL)
589     unittest.main(verbosity=2)