Merge "Stop printing in console when testing functest_utils.py"
[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 = os.path.normpath(os.path.join(os.path.dirname(
59             os.path.abspath(__file__)), '../../../ci/config_functest.yaml'))
60         self.db_url_env = 'http://foo/testdb'
61         self.file_yaml = {'general': {'openstack': {'image_name':
62                                                     'test_image_name'}}}
63
64     @mock.patch('six.moves.urllib.request.urlopen',
65                 side_effect=urllib.error.URLError('no host given'))
66     def test_check_internet_connectivity_failed(self, mock_method):
67         self.assertFalse(functest_utils.check_internet_connectivity())
68         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
69
70     @mock.patch('six.moves.urllib.request.urlopen')
71     def test_check_internet_connectivity_default(self, mock_method):
72         self.assertTrue(functest_utils.check_internet_connectivity())
73         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
74
75     @mock.patch('six.moves.urllib.request.urlopen')
76     def test_check_internet_connectivity_debian(self, mock_method):
77         self.url = "https://www.debian.org/"
78         self.assertTrue(functest_utils.check_internet_connectivity(self.url))
79         mock_method.assert_called_once_with(self.url, timeout=self.timeout)
80
81     @mock.patch('six.moves.urllib.request.urlopen',
82                 side_effect=urllib.error.URLError('no host given'))
83     def test_download_url_failed(self, mock_url):
84         self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
85
86     @mock.patch('six.moves.urllib.request.urlopen')
87     def test_download_url_default(self, mock_url):
88         with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
89                 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
90                 as mock_sh:
91             name = self.url.rsplit('/')[-1]
92             dest = self.dest_path + "/" + name
93             self.assertTrue(functest_utils.download_url(self.url,
94                                                         self.dest_path))
95             m.assert_called_once_with(dest, 'wb')
96             self.assertTrue(mock_sh.called)
97
98     def test_get_git_branch(self):
99         with mock.patch('functest.utils.functest_utils.Repo') as mock_repo:
100             mock_obj2 = mock.Mock()
101             attrs = {'name': 'test_branch'}
102             mock_obj2.configure_mock(**attrs)
103
104             mock_obj = mock.Mock()
105             attrs = {'active_branch': mock_obj2}
106             mock_obj.configure_mock(**attrs)
107
108             mock_repo.return_value = mock_obj
109             self.assertEqual(functest_utils.get_git_branch(self.repo_path),
110                              'test_branch')
111
112     @mock.patch('functest.utils.functest_utils.Repo',
113                 side_effect=NoSuchPathError)
114     def test_get_git_branch_failed(self, mock_repo):
115         self.assertRaises(NoSuchPathError,
116                           lambda: functest_utils.get_git_branch(self.repo_path
117                                                                 ))
118
119     @mock.patch('functest.utils.functest_utils.logger.error')
120     def test_get_installer_type_failed(self, mock_logger_error):
121         with mock.patch.dict(os.environ,
122                              {},
123                              clear=True):
124             self.assertEqual(functest_utils.get_installer_type(),
125                              "Unknown_installer")
126             mock_logger_error.assert_called_once_with("Impossible to retrieve"
127                                                       " the installer type")
128
129     def test_get_installer_type_default(self):
130         with mock.patch.dict(os.environ,
131                              {'INSTALLER_TYPE': 'test_installer'},
132                              clear=True):
133             self.assertEqual(functest_utils.get_installer_type(),
134                              self.installer)
135
136     @mock.patch('functest.utils.functest_utils.logger.info')
137     def test_get_scenario_failed(self, mock_logger_info):
138         with mock.patch.dict(os.environ,
139                              {},
140                              clear=True):
141             self.assertEqual(functest_utils.get_scenario(),
142                              "os-nosdn-nofeature-noha")
143             mock_logger_info.assert_called_once_with("Impossible to retrieve "
144                                                      "the scenario.Use "
145                                                      "default "
146                                                      "os-nosdn-nofeature-noha")
147
148     def test_get_scenario_default(self):
149         with mock.patch.dict(os.environ,
150                              {'DEPLOY_SCENARIO': 'test_scenario'},
151                              clear=True):
152             self.assertEqual(functest_utils.get_scenario(),
153                              self.scenario)
154
155     @mock.patch('functest.utils.functest_utils.get_build_tag')
156     def test_get_version_daily_job(self, mock_get_build_tag):
157         mock_get_build_tag.return_value = self.build_tag
158         self.assertEqual(functest_utils.get_version(), self.version)
159
160     @mock.patch('functest.utils.functest_utils.get_build_tag')
161     def test_get_version_weekly_job(self, mock_get_build_tag):
162         mock_get_build_tag.return_value = self.build_tag_week
163         self.assertEqual(functest_utils.get_version(), self.version)
164
165     @mock.patch('functest.utils.functest_utils.get_build_tag')
166     def test_get_version_with_dummy_build_tag(self, mock_get_build_tag):
167         mock_get_build_tag.return_value = 'whatever'
168         self.assertEqual(functest_utils.get_version(), 'unknown')
169
170     @mock.patch('functest.utils.functest_utils.get_build_tag')
171     def test_get_version_unknown(self, mock_get_build_tag):
172         mock_get_build_tag.return_value = "unknown_build_tag"
173         self.assertEqual(functest_utils.get_version(), "unknown")
174
175     @mock.patch('functest.utils.functest_utils.logger.info')
176     def test_get_pod_name_failed(self, mock_logger_info):
177         with mock.patch.dict(os.environ,
178                              {},
179                              clear=True):
180             self.assertEqual(functest_utils.get_pod_name(),
181                              "unknown-pod")
182             mock_logger_info.assert_called_once_with("Unable to retrieve "
183                                                      "the POD name from "
184                                                      "environment. Using "
185                                                      "pod name 'unknown-pod'")
186
187     def test_get_pod_name_default(self):
188         with mock.patch.dict(os.environ,
189                              {'NODE_NAME': 'test_node_name'},
190                              clear=True):
191             self.assertEqual(functest_utils.get_pod_name(),
192                              self.node_name)
193
194     @mock.patch('functest.utils.functest_utils.logger.info')
195     def test_get_build_tag_failed(self, mock_logger_info):
196         with mock.patch.dict(os.environ,
197                              {},
198                              clear=True):
199             self.assertEqual(functest_utils.get_build_tag(),
200                              "none")
201             mock_logger_info.assert_called_once_with("Impossible to retrieve"
202                                                      " the build tag")
203
204     def test_get_build_tag_default(self):
205         with mock.patch.dict(os.environ,
206                              {'BUILD_TAG': self.build_tag},
207                              clear=True):
208             self.assertEqual(functest_utils.get_build_tag(),
209                              self.build_tag)
210
211     def test_get_db_url_env_var(self):
212         with mock.patch.dict(os.environ,
213                              {'TEST_DB_URL': self.db_url_env,
214                               'CONFIG_FUNCTEST_YAML':
215                               "./functest/ci/config_functest.yaml"},
216                              clear=True):
217             self.assertEqual(functest_utils.get_db_url(),
218                              self.db_url_env)
219
220     @mock.patch('functest.utils.functest_utils.get_functest_config')
221     def test_get_db_url_default(self, mock_get_functest_config):
222         mock_get_functest_config.return_value = self.db_url
223         self.assertEqual(functest_utils.get_db_url(), self.db_url)
224         mock_get_functest_config.assert_called_once_with('results.test_db_url')
225
226     @mock.patch('functest.utils.functest_utils.logger.info')
227     def test_logger_test_results(self, mock_logger_info):
228         with mock.patch('functest.utils.functest_utils.get_pod_name',
229                         return_value=self.node_name), \
230                 mock.patch('functest.utils.functest_utils.get_scenario',
231                            return_value=self.scenario), \
232                 mock.patch('functest.utils.functest_utils.get_version',
233                            return_value=self.version), \
234                 mock.patch('functest.utils.functest_utils.get_build_tag',
235                            return_value=self.build_tag), \
236                 mock.patch('functest.utils.functest_utils.get_db_url',
237                            return_value=self.db_url):
238             functest_utils.logger_test_results(self.project, self.case_name,
239                                                self.status, self.details)
240             mock_logger_info.assert_called_once_with(
241                 "\n"
242                 "****************************************\n"
243                 "\t %(p)s/%(n)s results \n\n"
244                 "****************************************\n"
245                 "DB:\t%(db)s\n"
246                 "pod:\t%(pod)s\n"
247                 "version:\t%(v)s\n"
248                 "scenario:\t%(s)s\n"
249                 "status:\t%(c)s\n"
250                 "build tag:\t%(b)s\n"
251                 "details:\t%(d)s\n"
252                 % {'p': self.project,
253                     'n': self.case_name,
254                     'db': self.db_url,
255                     'pod': self.node_name,
256                     'v': self.version,
257                     's': self.scenario,
258                     'c': self.status,
259                     'b': self.build_tag,
260                     'd': self.details})
261
262     def _get_env_dict(self, var):
263         dic = {'INSTALLER_TYPE': self.installer,
264                'DEPLOY_SCENARIO': self.scenario,
265                'NODE_NAME': self.node_name,
266                'BUILD_TAG': self.build_tag}
267         dic.pop(var, None)
268         return dic
269
270     def _test_push_results_to_db_missing_env(self, env_var):
271         dic = self._get_env_dict(env_var)
272         with mock.patch('functest.utils.functest_utils.get_db_url',
273                         return_value=self.db_url), \
274                 mock.patch.dict(os.environ,
275                                 dic,
276                                 clear=True), \
277                 mock.patch('functest.utils.functest_utils.logger.error') \
278                 as mock_logger_error:
279             functest_utils.push_results_to_db(self.project, self.case_name,
280                                               self.start_date, self.stop_date,
281                                               self.result, self.details)
282             mock_logger_error.assert_called_once_with("Please set env var: " +
283                                                       str("\'" + env_var +
284                                                           "\'"))
285
286     def test_push_results_to_db_missing_installer(self):
287         self._test_push_results_to_db_missing_env('INSTALLER_TYPE')
288
289     def test_push_results_to_db_missing_scenario(self):
290         self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO')
291
292     def test_push_results_to_db_missing_nodename(self):
293         self._test_push_results_to_db_missing_env('NODE_NAME')
294
295     def test_push_results_to_db_missing_buildtag(self):
296         self._test_push_results_to_db_missing_env('BUILD_TAG')
297
298     def test_push_results_to_db_request_post_failed(self):
299         dic = self._get_env_dict(None)
300         with mock.patch('functest.utils.functest_utils.get_db_url',
301                         return_value=self.db_url), \
302                 mock.patch.dict(os.environ,
303                                 dic,
304                                 clear=True), \
305                 mock.patch('functest.utils.functest_utils.logger.error') \
306                 as mock_logger_error, \
307                 mock.patch('functest.utils.functest_utils.requests.post',
308                            side_effect=requests.RequestException):
309             self.assertFalse(functest_utils.
310                              push_results_to_db(self.project, self.case_name,
311                                                 self.start_date,
312                                                 self.stop_date,
313                                                 self.result, self.details))
314             mock_logger_error.assert_called_once_with(test_utils.
315                                                       RegexMatch("Pushing "
316                                                                  "Result to"
317                                                                  " DB"
318                                                                  "(\S+\s*) "
319                                                                  "failed:"))
320
321     def test_push_results_to_db_request_post_exception(self):
322         dic = self._get_env_dict(None)
323         with mock.patch('functest.utils.functest_utils.get_db_url',
324                         return_value=self.db_url), \
325                 mock.patch.dict(os.environ,
326                                 dic,
327                                 clear=True), \
328                 mock.patch('functest.utils.functest_utils.logger.error') \
329                 as mock_logger_error, \
330                 mock.patch('functest.utils.functest_utils.requests.post',
331                            side_effect=Exception):
332             self.assertFalse(functest_utils.
333                              push_results_to_db(self.project, self.case_name,
334                                                 self.start_date,
335                                                 self.stop_date,
336                                                 self.result, self.details))
337             self.assertTrue(mock_logger_error.called)
338
339     def test_push_results_to_db_default(self):
340         dic = self._get_env_dict(None)
341         with mock.patch('functest.utils.functest_utils.get_db_url',
342                         return_value=self.db_url), \
343                 mock.patch.dict(os.environ,
344                                 dic,
345                                 clear=True), \
346                 mock.patch('functest.utils.functest_utils.requests.post'):
347             self.assertTrue(functest_utils.
348                             push_results_to_db(self.project, self.case_name,
349                                                self.start_date,
350                                                self.stop_date,
351                                                self.result, self.details))
352     readline = 0
353     test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
354
355     @staticmethod
356     def readline_side():
357         if FunctestUtilsTesting.readline == \
358                 len(FunctestUtilsTesting.test_ip) - 1:
359             return False
360         FunctestUtilsTesting.readline += 1
361         return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
362
363     # TODO: get_resolvconf_ns
364     @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
365     def test_get_resolvconf_ns_default(self, mock_dns_resolve):
366         attrs = {'query.return_value': ["test"]}
367         mock_dns_resolve.configure_mock(**attrs)
368
369         m = mock.Mock()
370         attrs = {'readline.side_effect': self.readline_side}
371         m.configure_mock(**attrs)
372
373         with mock.patch("six.moves.builtins.open") as mo:
374             mo.return_value = m
375             self.assertEqual(functest_utils.get_resolvconf_ns(),
376                              self.test_ip[1:])
377
378     def _get_environ(self, var):
379         if var == 'INSTALLER_TYPE':
380             return self.installer
381         elif var == 'DEPLOY_SCENARIO':
382             return self.scenario
383         return var
384
385     def test_get_ci_envvars_default(self):
386         with mock.patch('os.environ.get',
387                         side_effect=self._get_environ):
388             dic = {"installer": self.installer,
389                    "scenario": self.scenario}
390             self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
391
392     def cmd_readline(self):
393         return 'test_value\n'
394
395     @mock.patch('functest.utils.functest_utils.logger.error')
396     @mock.patch('functest.utils.functest_utils.logger.info')
397     def test_execute_command_args_present_with_error(self, mock_logger_info,
398                                                      mock_logger_error):
399         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
400                 as mock_subproc_open, \
401                 mock.patch('six.moves.builtins.open',
402                            mock.mock_open()) as mopen:
403
404             FunctestUtilsTesting.readline = 0
405
406             mock_obj = mock.Mock()
407             attrs = {'readline.side_effect': self.cmd_readline()}
408             mock_obj.configure_mock(**attrs)
409
410             mock_obj2 = mock.Mock()
411             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
412             mock_obj2.configure_mock(**attrs)
413
414             mock_subproc_open.return_value = mock_obj2
415
416             resp = functest_utils.execute_command(self.cmd, info=True,
417                                                   error_msg=self.error_msg,
418                                                   verbose=True,
419                                                   output_file=self.output_file)
420             self.assertEqual(resp, 1)
421             msg_exec = ("Executing command: '%s'" % self.cmd)
422             mock_logger_info.assert_called_once_with(msg_exec)
423             mopen.assert_called_once_with(self.output_file, "w")
424             mock_logger_error.assert_called_once_with(self.error_msg)
425
426     @mock.patch('functest.utils.functest_utils.logger.info')
427     def test_execute_command_args_present_with_success(self, mock_logger_info,
428                                                        ):
429         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
430                 as mock_subproc_open, \
431                 mock.patch('six.moves.builtins.open',
432                            mock.mock_open()) as mopen:
433
434             FunctestUtilsTesting.readline = 0
435
436             mock_obj = mock.Mock()
437             attrs = {'readline.side_effect': self.cmd_readline()}
438             mock_obj.configure_mock(**attrs)
439
440             mock_obj2 = mock.Mock()
441             attrs = {'stdout': mock_obj, 'wait.return_value': 0}
442             mock_obj2.configure_mock(**attrs)
443
444             mock_subproc_open.return_value = mock_obj2
445
446             resp = functest_utils.execute_command(self.cmd, info=True,
447                                                   error_msg=self.error_msg,
448                                                   verbose=True,
449                                                   output_file=self.output_file)
450             self.assertEqual(resp, 0)
451             msg_exec = ("Executing command: '%s'" % self.cmd)
452             mock_logger_info.assert_called_once_with(msg_exec)
453             mopen.assert_called_once_with(self.output_file, "w")
454
455     @mock.patch('sys.stdout')
456     def test_execute_command_args_missing_with_success(self, stdout=None):
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('sys.stdout')
479     def test_execute_command_args_missing_with_error(self, stdout=None):
480         with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
481                 as mock_subproc_open:
482
483             FunctestUtilsTesting.readline = 2
484             mock_obj = mock.Mock()
485             attrs = {'readline.side_effect': self.cmd_readline()}
486             mock_obj.configure_mock(**attrs)
487
488             mock_obj2 = mock.Mock()
489             attrs = {'stdout': mock_obj, 'wait.return_value': 1}
490             mock_obj2.configure_mock(**attrs)
491
492             mock_subproc_open.return_value = mock_obj2
493
494             resp = functest_utils.execute_command(self.cmd, info=False,
495                                                   error_msg="",
496                                                   verbose=False,
497                                                   output_file=None)
498             self.assertEqual(resp, 1)
499
500     def _get_functest_config(self, var):
501         return var
502
503     @mock.patch('functest.utils.functest_utils.logger.error')
504     def test_get_dict_by_test(self, mock_logger_error):
505         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
506                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
507                 as mock_yaml, \
508                 mock.patch('functest.utils.functest_utils.get_testcases_'
509                            'file_dir'):
510             mock_obj = mock.Mock()
511             attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]}
512             mock_obj.configure_mock(**attrs)
513
514             mock_yaml.return_value = mock_obj
515
516             self.assertDictEqual(functest_utils.
517                                  get_dict_by_test(self.testname),
518                                  self.testcase_dict)
519
520     @mock.patch('functest.utils.functest_utils.get_dict_by_test')
521     def test_get_criteria_by_test_default(self, mock_get_dict_by_test):
522         mock_get_dict_by_test.return_value = self.testcase_dict
523         self.assertEqual(functest_utils.get_criteria_by_test(self.testname),
524                          self.criteria)
525
526     @mock.patch('functest.utils.functest_utils.get_dict_by_test')
527     def test_get_criteria_by_test_failed(self, mock_get_dict_by_test):
528         mock_get_dict_by_test.return_value = None
529         self.assertIsNone(functest_utils.get_criteria_by_test(self.testname))
530
531     def test_get_parameter_from_yaml_failed(self):
532         self.file_yaml['general'] = None
533         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
534                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
535                 as mock_yaml, \
536                 self.assertRaises(ValueError) as excep:
537             mock_yaml.return_value = self.file_yaml
538             functest_utils.get_parameter_from_yaml(self.parameter,
539                                                    self.test_file)
540             self.assertTrue(("The parameter %s is not"
541                              " defined in config_functest.yaml" %
542                              self.parameter) in excep.exception)
543
544     def test_get_parameter_from_yaml_default(self):
545         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
546                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
547                 as mock_yaml:
548             mock_yaml.return_value = self.file_yaml
549             self.assertEqual(functest_utils.
550                              get_parameter_from_yaml(self.parameter,
551                                                      self.test_file),
552                              'test_image_name')
553
554     @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
555     def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
556         with mock.patch.dict(os.environ,
557                              {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
558             functest_utils.get_functest_config(self.parameter)
559             mock_get_parameter_from_yaml. \
560                 assert_called_once_with(self.parameter,
561                                         self.config_yaml)
562
563     # TODO: merge_dicts
564
565     def test_get_testcases_file_dir(self):
566         resp = functest_utils.get_testcases_file_dir()
567         self.assertEqual(resp,
568                          "/home/opnfv/repos/functest/"
569                          "functest/ci/testcases.yaml")
570
571     def test_get_functest_yaml(self):
572         with mock.patch('six.moves.builtins.open', mock.mock_open()), \
573                 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
574                 as mock_yaml:
575             mock_yaml.return_value = self.file_yaml
576             resp = functest_utils.get_functest_yaml()
577             self.assertEqual(resp, self.file_yaml)
578
579     @mock.patch('functest.utils.functest_utils.logger.info')
580     def test_print_separator(self, mock_logger_info):
581         functest_utils.print_separator()
582         mock_logger_info.assert_called_once_with("======================="
583                                                  "=======================")
584
585
586 if __name__ == "__main__":
587     logging.disable(logging.CRITICAL)
588     unittest.main(verbosity=2)