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     logging.disable(logging.CRITICAL)
  36
  37     _case_name = 'base'
  38     _project_name = 'functest'
  39     _start_time = 1.0
  40     _stop_time = 2.0
  41     _result = 'PASS'
  42     _build_tag = VERSION
  43     _node_name = 'bar'
  44     _deploy_scenario = 'foo'
  45     _installer_type = 'debian'
  46
  47     def setUp(self):
  48         os.environ['INSTALLER_TYPE'] = self._installer_type
  49         os.environ['DEPLOY_SCENARIO'] = self._deploy_scenario
  50         os.environ['NODE_NAME'] = self._node_name
  51         os.environ['BUILD_TAG'] = self._build_tag
  52
  53     def test_wraps(self):
  54         self.assertEqual(functest_utils.push_results_to_db.__name__,
  55                          "push_results_to_db")
  56
  57     def _get_json(self):
  58         stop_time = datetime.fromtimestamp(self._stop_time).strftime(
  59             '%Y-%m-%d %H:%M:%S')
  60         start_time = datetime.fromtimestamp(self._start_time).strftime(
  61             '%Y-%m-%d %H:%M:%S')
  62         data = {'project_name': self._project_name,
  63                 'stop_date': stop_time, 'start_date': start_time,
  64                 'case_name': self._case_name, 'build_tag': self._build_tag,
  65                 'pod_name': self._node_name, 'installer': self._installer_type,
  66                 'scenario': self._deploy_scenario, 'version': VERSION,
  67                 'details': {}, 'criteria': self._result}
  68         return json.dumps(data)
  69
  70     @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
  71                 return_value='http://127.0.0.1')
  72     @mock.patch('{}.get_version'.format(functest_utils.__name__),
  73                 return_value=VERSION)
  74     @mock.patch('requests.post')
  75     def test_http_shema(self, *args):
  76         self.assertTrue(functest_utils.push_results_to_db(
  77             self._project_name, self._case_name, self._start_time,
  78             self._stop_time, self._result, {}))
  79         args[1].assert_called_once_with()
  80         args[2].assert_called_once_with()
  81         args[0].assert_called_once_with(
  82             'http://127.0.0.1', data=self._get_json(),
  83             headers={'Content-Type': 'application/json'})
  84
  85     @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
  86                 return_value="/dev/null")
  87     def test_wrong_shema(self, mock_method=None):
  88         self.assertFalse(functest_utils.push_results_to_db(
  89             self._project_name, self._case_name, self._start_time,
  90             self._stop_time, self._result, {}))
  91         mock_method.assert_called_once_with()
  92
  93     @mock.patch('{}.get_version'.format(functest_utils.__name__),
  94                 return_value=VERSION)
  95     @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
  96                 return_value=URL)
  97     def _test_dump(self, *args):
  98         with mock.patch.object(decorators, 'open', mock.mock_open(),
  99                                create=True) as mock_open:
 100             self.assertTrue(functest_utils.push_results_to_db(
 101                 self._project_name, self._case_name, self._start_time,
 102                 self._stop_time, self._result, {}))
 103         mock_open.assert_called_once_with(FILE, 'a')
 104         handle = mock_open()
 105         call_args, _ = handle.write.call_args
 106         self.assertIn('POST', call_args[0])
 107         self.assertIn(self._get_json(), call_args[0])
 108         args[0].assert_called_once_with()
 109         args[1].assert_called_once_with()
 110
 111     @mock.patch('os.makedirs')
 112     def test_default_dump(self, mock_method=None):
 113         self._test_dump()
 114         mock_method.assert_called_once_with(DIR)
 115
 116     @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, ''))
 117     def test_makedirs_dir_exists(self, mock_method=None):
 118         self._test_dump()
 119         mock_method.assert_called_once_with(DIR)
 120
 121     @mock.patch('{}.get_db_url'.format(functest_utils.__name__),
 122                 return_value=URL)
 123     @mock.patch('os.makedirs', side_effect=OSError)
 124     def test_makedirs_exc(self, *args):
 125         self.assertFalse(
 126             functest_utils.push_results_to_db(
 127                 self._project_name, self._case_name, self._start_time,
 128                 self._stop_time, self._result, {}))
 129         args[0].assert_called_once_with(DIR)
 130         args[1].assert_called_once_with()
 131
 132
 133 if __name__ == "__main__":
 134     logging.basicConfig()
 135     unittest.main(verbosity=2)