functest/tests/unit/utils/test_decorators.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 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 """Define the class required to fully cover decorators."""
  11
  12 from datetime import datetime
  13 import errno
  14 import json
  15 import logging
  16 import os
  17 import unittest
  18
  19 import mock
  20
  21 from functest.utils import decorators
  22 from functest.utils import functest_utils
  23
  24 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
  25
  26 VERSION = 'master'
  27 DIR = '/dev'
  28 FILE = '{}/null'.format(DIR)
  29 URL = 'file://{}'.format(FILE)
  30
  31
  32 class DecoratorsTesting(unittest.TestCase):
  33     # pylint: disable=missing-docstring
  34
  35     _case_name = 'base'
  36     _project_name = 'functest'
  37     _start_time = 1.0
  38     _stop_time = 2.0
  39     _result = 'PASS'
  40     _build_tag = VERSION
  41     _node_name = 'bar'
  42     _deploy_scenario = 'foo'
  43     _installer_type = 'debian'
  44
  45     def setUp(self):
  46         os.environ['INSTALLER_TYPE'] = self._installer_type
  47         os.environ['DEPLOY_SCENARIO'] = self._deploy_scenario
  48         os.environ['NODE_NAME'] = self._node_name
  49         os.environ['BUILD_TAG'] = self._build_tag
  50
  51     def test_wraps(self):
  52         self.assertEqual(functest_utils.push_results_to_db.__name__,
  53                          "push_results_to_db")
  54
  55     def _get_json(self):
  56         stop_time = datetime.fromtimestamp(self._stop_time).strftime(
  57             '%Y-%m-%d %H:%M:%S')
  58         start_time = datetime.fromtimestamp(self._start_time).strftime(
  59             '%Y-%m-%d %H:%M:%S')
  60         data = {'project_name': self._project_name,
  61                 'stop_date': stop_time, 'start_date': start_time,
  62                 'case_name': self._case_name, 'build_tag': self._build_tag,
  63                 'pod_name': self._node_name, 'installer': self._installer_type,
  64                 'scenario': self._deploy_scenario, 'version': VERSION,
  65                 'details': {}, 'criteria': self._result}
  66         return json.dumps(data, sort_keys=True)
  67
  68     @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
  69                 return_value='http://127.0.0.1')
  70     @mock.patch('{}.get_version'.format(functest_utils.__name__),
  71                 return_value=VERSION)
  72     @mock.patch('requests.post')
  73     def test_http_shema(self, *args):
  74         self.assertTrue(functest_utils.push_results_to_db(
  75             self._project_name, self._case_name, self._start_time,
  76             self._stop_time, self._result, {}))
  77         args[1].assert_called_once_with()
  78         args[2].assert_called_once_with()
  79         args[0].assert_called_once_with(
  80             'http://127.0.0.1', data=self._get_json(),
  81             headers={'Content-Type': 'application/json'})
  82
  83     @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
  84                 return_value="/dev/null")
  85     def test_wrong_shema(self, mock_method=None):
  86         self.assertFalse(functest_utils.push_results_to_db(
  87             self._project_name, self._case_name, self._start_time,
  88             self._stop_time, self._result, {}))
  89         mock_method.assert_called_once_with()
  90
  91     @mock.patch('{}.get_version'.format(functest_utils.__name__),
  92                 return_value=VERSION)
  93     @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
  94                 return_value=URL)
  95     def _test_dump(self, *args):
  96         with mock.patch.object(decorators, 'open', mock.mock_open(),
  97                                create=True) as mock_open:
  98             self.assertTrue(functest_utils.push_results_to_db(
  99                 self._project_name, self._case_name, self._start_time,
 100                 self._stop_time, self._result, {}))
 101         mock_open.assert_called_once_with(FILE, 'a')
 102         handle = mock_open()
 103         call_args, _ = handle.write.call_args
 104         self.assertIn('POST', call_args[0])
 105         self.assertIn(self._get_json(), call_args[0])
 106         args[0].assert_called_once_with()
 107         args[1].assert_called_once_with()
 108
 109     @mock.patch('os.makedirs')
 110     def test_default_dump(self, mock_method=None):
 111         self._test_dump()
 112         mock_method.assert_called_once_with(DIR)
 113
 114     @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, ''))
 115     def test_makedirs_dir_exists(self, mock_method=None):
 116         self._test_dump()
 117         mock_method.assert_called_once_with(DIR)
 118
 119     @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
 120                 return_value=URL)
 121     @mock.patch('os.makedirs', side_effect=OSError)
 122     def test_makedirs_exc(self, *args):
 123         self.assertFalse(
 124             functest_utils.push_results_to_db(
 125                 self._project_name, self._case_name, self._start_time,
 126                 self._stop_time, self._result, {}))
 127         args[0].assert_called_once_with(DIR)
 128         args[1].assert_called_once_with()
 129
 130
 131 if __name__ == "__main__":
 132     logging.disable(logging.CRITICAL)
 133     unittest.main(verbosity=2)