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.core import testcase
  22 from functest.utils import decorators
  23
  24 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
  25
  26 DIR = '/dev'
  27 FILE = '{}/null'.format(DIR)
  28 URL = 'file://{}'.format(FILE)
  29
  30
  31 class DecoratorsTesting(unittest.TestCase):
  32     # pylint: disable=missing-docstring
  33
  34     _case_name = 'base'
  35     _project_name = 'functest'
  36     _start_time = 1.0
  37     _stop_time = 2.0
  38     _result = 'PASS'
  39     _version = 'unknown'
  40     _build_tag = 'none'
  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(testcase.TestCase.push_to_db.__name__,
  53                          "push_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': self._version,
  65                 'details': {}, 'criteria': self._result}
  66         return json.dumps(data, sort_keys=True)
  67
  68     def _get_testcase(self):
  69         test = testcase.TestCase(
  70             project_name=self._project_name, case_name=self._case_name)
  71         test.start_time = self._start_time
  72         test.stop_time = self._stop_time
  73         test.result = 100
  74         test.details = {}
  75         return test
  76
  77     @mock.patch('requests.post')
  78     def test_http_shema(self, *args):
  79         os.environ['TEST_DB_URL'] = 'http://127.0.0.1'
  80         test = self._get_testcase()
  81         self.assertEqual(test.push_to_db(), testcase.TestCase.EX_OK)
  82         args[0].assert_called_once_with(
  83             'http://127.0.0.1', data=self._get_json(),
  84             headers={'Content-Type': 'application/json'})
  85
  86     def test_wrong_shema(self):
  87         os.environ['TEST_DB_URL'] = '/dev/null'
  88         test = self._get_testcase()
  89         self.assertEqual(
  90             test.push_to_db(), testcase.TestCase.EX_PUSH_TO_DB_ERROR)
  91
  92     def _test_dump(self):
  93         os.environ['TEST_DB_URL'] = URL
  94         with mock.patch.object(decorators, 'open', mock.mock_open(),
  95                                create=True) as mock_open:
  96             test = self._get_testcase()
  97             self.assertEqual(test.push_to_db(), testcase.TestCase.EX_OK)
  98         mock_open.assert_called_once_with(FILE, 'a')
  99         handle = mock_open()
 100         call_args, _ = handle.write.call_args
 101         self.assertIn('POST', call_args[0])
 102         self.assertIn(self._get_json(), call_args[0])
 103
 104     @mock.patch('os.makedirs')
 105     def test_default_dump(self, mock_method=None):
 106         self._test_dump()
 107         mock_method.assert_called_once_with(DIR)
 108
 109     @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, ''))
 110     def test_makedirs_dir_exists(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)
 115     def test_makedirs_exc(self, *args):
 116         os.environ['TEST_DB_URL'] = URL
 117         test = self._get_testcase()
 118         self.assertEqual(
 119             test.push_to_db(), testcase.TestCase.EX_PUSH_TO_DB_ERROR)
 120         args[0].assert_called_once_with(DIR)
 121
 122
 123 if __name__ == "__main__":
 124     logging.disable(logging.CRITICAL)
 125     unittest.main(verbosity=2)