Merge "Enable tempest offline by use_custom_images=True"
[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 from functest.utils.constants import CONST
23
24
25 class FunctestUtilsTesting(unittest.TestCase):
26
27     def setUp(self):
28         self.url = 'http://www.opnfv.org/'
29         self.timeout = 5
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'
43         self.criteria = 50
44         self.result = 75
45         self.start_date = 1482624000
46         self.stop_date = 1482624000
47         self.start_time = time.time()
48         self.stop_time = time.time()
49         self.readline = -1
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'
53         self.cmd = 'test_cmd'
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':
64                                                     'test_image_name'}}}
65
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)
71
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)
76
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)
82
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))
87
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')\
92                 as mock_sh:
93             name = self.url.rsplit('/')[-1]
94             dest = self.dest_path + "/" + name
95             self.assertTrue(functest_utils.download_url(self.url,
96                                                         self.dest_path))
97             m.assert_called_once_with(dest, 'wb')
98             self.assertTrue(mock_sh.called)
99
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)
105
106             mock_obj = mock.Mock()
107             attrs = {'active_branch': mock_obj2}
108             mock_obj.configure_mock(**attrs)
109
110             mock_repo.return_value = mock_obj
111             self.assertEqual(functest_utils.get_git_branch(self.repo_path),
112                              'test_branch')
113
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
119                                                                 ))
120
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,
124                              {},
125                              clear=True):
126             self.assertEqual(functest_utils.get_installer_type(),
127                              "Unknown_installer")
128             mock_logger_error.assert_called_once_with("Impossible to retrieve"
129                                                       " the installer type")
130
131     def test_get_installer_type_default(self):
132         with mock.patch.dict(os.environ,
133                              {'INSTALLER_TYPE': 'test_installer'},
134                              clear=True):
135             self.assertEqual(functest_utils.get_installer_type(),
136                              self.installer)
137
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,
141                              {},
142                              clear=True):
143             self.assertEqual(functest_utils.get_scenario(),
144                              "os-nosdn-nofeature-noha")
145             mock_logger_info.assert_called_once_with("Impossible to retrieve "
146                                                      "the scenario.Use "
147                                                      "default "
148                                                      "os-nosdn-nofeature-noha")
149
150     def test_get_scenario_default(self):
151         with mock.patch.dict(os.environ,
152                              {'DEPLOY_SCENARIO': 'test_scenario'},
153                              clear=True):
154             self.assertEqual(functest_utils.get_scenario(),
155                              self.scenario)
156
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)
161
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)
166
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')
171
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")
176
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,
180                              {},
181                              clear=True):
182             self.assertEqual(functest_utils.get_pod_name(),
183                              "unknown-pod")
184             mock_logger_info.assert_called_once_with("Unable to retrieve "
185                                                      "the POD name from "
186                                                      "environment. Using "
187                                                      "pod name 'unknown-pod'")
188
189     def test_get_pod_name_default(self):
190         with mock.patch.dict(os.environ,
191                              {'NODE_NAME': 'test_node_name'},
192                              clear=True):
193             self.assertEqual(functest_utils.get_pod_name(),
194                              self.node_name)
195
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,
199                              {},
200                              clear=True):
201             self.assertEqual(functest_utils.get_build_tag(),
202                              "none")
203             mock_logger_info.assert_called_once_with("Impossible to retrieve"
204                                                      " the build tag")
205
206     def test_get_build_tag_default(self):
207         with mock.patch.dict(os.environ,
208                              {'BUILD_TAG': self.build_tag},
209                              clear=True):
210             self.assertEqual(functest_utils.get_build_tag(),
211                              self.build_tag)
212
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(
227                 "\n"
228                 "****************************************\n"
229                 "\t %(p)s/%(n)s results \n\n"
230                 "****************************************\n"
231                 "DB:\t%(db)s\n"
232                 "pod:\t%(pod)s\n"
233                 "version:\t%(v)s\n"
234                 "scenario:\t%(s)s\n"
235                 "status:\t%(c)s\n"
236                 "build tag:\t%(b)s\n"
237                 "details:\t%(d)s\n"
238                 % {'p': self.project,
239                     'n': self.case_name,
240                     'db': CONST.__getattribute__('results_test_db_url'),
241                     'pod': self.node_name,
242                     'v': self.version,
243                     's': self.scenario,
244                     'c': self.status,
245                     'b': self.build_tag,
246                     'd': self.details})
247
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}
253         dic.pop(var, None)
254         return dic
255
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,
260                              dic,
261                              clear=True), \
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: " +
268                                                       str("\'" + env_var +
269                                                           "\'"))
270
271     def test_push_results_to_db_missing_installer(self):
272         self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
273
274     def test_push_results_to_db_missing_scenario(self):
275         self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
276
277     def test_push_results_to_db_missing_nodename(self):
278         self._test_push_results_to_db_missing_env('NODE_NAME')
279
280     def test_push_results_to_db_missing_buildtag(self):
281         self._test_push_results_to_db_missing_env('BUILD_TAG')
282
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,
287                              dic,
288                              clear=True), \
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,
295                                                 self.start_date,
296                                                 self.stop_date,
297                                                 self.result, self.details))
298             mock_logger_error.assert_called_once_with(test_utils.
299                                                       RegexMatch("Pushing "
300                                                                  "Result to"
301                                                                  " DB"
302                                                                  "(\S+\s*) "
303                                                                  "failed:"))
304
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,
309                              dic,
310                              clear=True), \
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,
317                                                 self.start_date,
318                                                 self.stop_date,
319                                                 self.result, self.details))
320             self.assertTrue(mock_logger_error.called)
321
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,
326                              dic,
327                              clear=True), \
328                 mock.patch('functest.utils.functest_utils.requests.post'):
329             self.assertTrue(functest_utils.
330                             push_results_to_db(self.project, self.case_name,
331                                                self.start_date,
332                                                self.stop_date,
333                                                self.result, self.details))
334     readline = 0
335     test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
336
337     @staticmethod
338     def readline_side():
339         if FunctestUtilsTesting.readline == \
340                 len(FunctestUtilsTesting.test_ip) - 1:
341             return False
342         FunctestUtilsTesting.readline += 1
343         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
344
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)
350
351         m = mock.Mock()
352         attrs = {'readline.side_effect': self.readline_side}
353         m.configure_mock(**attrs)
354
355         with mock.patch("six.moves.builtins.open") as mo:
356             mo.return_value = m
357             self.assertEqual(functest_utils.get_resolvconf_ns(),
358                              self.test_ip[1:])
359
360     def _get_environ(self, var):
361         if var == 'INSTALLER_TYPE':
362             return self.installer
363         elif var == 'DEPLOY_SCENARIO':
364             return self.scenario
365         return var
366
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)
373
374     def cmd_readline(self):
375         return 'test_value\n'
376
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,
380                                                      mock_logger_error):
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:
385
386             FunctestUtilsTesting.readline = 0
387
388             mock_obj = mock.Mock()
389             attrs = {'readline.side_effect': self.cmd_readline()}
390             mock_obj.configure_mock(**attrs)
391
392             mock_obj2 = mock.Mock()
393             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
394             mock_obj2.configure_mock(**attrs)
395
396             mock_subproc_open.return_value = mock_obj2
397
398             resp = functest_utils.execute_command(self.cmd, info=True,
399                                                   error_msg=self.error_msg,
400                                                   verbose=True,
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)
407
408     @mock.patch('functest.utils.functest_utils.logger.info')
409     def test_execute_command_args_present_with_success(self, mock_logger_info,
410                                                        ):
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:
415
416             FunctestUtilsTesting.readline = 0
417
418             mock_obj = mock.Mock()
419             attrs = {'readline.side_effect': self.cmd_readline()}
420             mock_obj.configure_mock(**attrs)
421
422             mock_obj2 = mock.Mock()
423             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
424             mock_obj2.configure_mock(**attrs)
425
426             mock_subproc_open.return_value = mock_obj2
427
428             resp = functest_utils.execute_command(self.cmd, info=True,
429                                                   error_msg=self.error_msg,
430                                                   verbose=True,
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")
436
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:
441
442             FunctestUtilsTesting.readline = 2
443
444             mock_obj = mock.Mock()
445             attrs = {'readline.side_effect': self.cmd_readline()}
446             mock_obj.configure_mock(**attrs)
447
448             mock_obj2 = mock.Mock()
449             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
450             mock_obj2.configure_mock(**attrs)
451
452             mock_subproc_open.return_value = mock_obj2
453
454             resp = functest_utils.execute_command(self.cmd, info=False,
455                                                   error_msg="",
456                                                   verbose=False,
457                                                   output_file=None)
458             self.assertEqual(resp, 0)
459
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:
464
465             FunctestUtilsTesting.readline = 2
466             mock_obj = mock.Mock()
467             attrs = {'readline.side_effect': self.cmd_readline()}
468             mock_obj.configure_mock(**attrs)
469
470             mock_obj2 = mock.Mock()
471             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
472             mock_obj2.configure_mock(**attrs)
473
474             mock_subproc_open.return_value = mock_obj2
475
476             resp = functest_utils.execute_command(self.cmd, info=False,
477                                                   error_msg="",
478                                                   verbose=False,
479                                                   output_file=None)
480             self.assertEqual(resp, 1)
481
482     def _get_functest_config(self, var):
483         return var
484
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') \
489                 as mock_yaml, \
490                 mock.patch('functest.utils.functest_utils.get_testcases_'
491                            'file_dir'):
492             mock_obj = mock.Mock()
493             attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
494             mock_obj.configure_mock(**attrs)
495
496             mock_yaml.return_value = mock_obj
497
498             self.assertDictEqual(functest_utils.
499                                  get_dict_by_test(self.testname),
500                                  self.testcase_dict)
501
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),
506                          self.criteria)
507
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))
512
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') \
517                 as mock_yaml, \
518                 self.assertRaises(ValueError) as excep:
519             mock_yaml.return_value = self.file_yaml
520             functest_utils.get_parameter_from_yaml(self.parameter,
521                                                    self.test_file)
522             self.assertTrue(("The parameter %s is not"
523                              " defined in config_functest.yaml" %
524                              self.parameter) in excep.exception)
525
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') \
529                 as mock_yaml:
530             mock_yaml.return_value = self.file_yaml
531             self.assertEqual(functest_utils.
532                              get_parameter_from_yaml(self.parameter,
533                                                      self.test_file),
534                              'test_image_name')
535
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,
543                                         self.config_yaml)
544
545     # TODO: merge_dicts
546
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')
554
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') \
558                 as mock_yaml:
559             mock_yaml.return_value = self.file_yaml
560             resp = functest_utils.get_functest_yaml()
561             self.assertEqual(resp, self.file_yaml)
562
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                                                  "=======================")
568
569
570 if __name__ == "__main__":
571     logging.disable(logging.CRITICAL)
572     unittest.main(verbosity=2)