ceph/src/pybind/mgr/dashboard/tests/test_tools.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3
   4 import unittest
   5
   6 import cherrypy
   7 from cherrypy.lib.sessions import RamSession
   8 try:
   9     from mock import patch
  10 except ImportError:
  11     from unittest.mock import patch
  12
  13 from . import ControllerTestCase
  14 from ..services.exception import handle_rados_error
  15 from ..controllers import RESTController, ApiController, Controller, \
  16                           BaseController, Proxy
  17 from ..tools import dict_contains_path, json_str_to_object, partial_dict,\
  18                     dict_get, RequestLoggingTool
  19
  20
  21 # pylint: disable=W0613
  22 @Controller('/foo', secure=False)
  23 class FooResource(RESTController):
  24     elems = []
  25
  26     def list(self):
  27         return FooResource.elems
  28
  29     def create(self, a):
  30         FooResource.elems.append({'a': a})
  31         return {'a': a}
  32
  33     def get(self, key):
  34         return {'detail': (key, [])}
  35
  36     def delete(self, key):
  37         del FooResource.elems[int(key)]
  38
  39     def bulk_delete(self):
  40         FooResource.elems = []
  41
  42     def set(self, key, newdata):
  43         FooResource.elems[int(key)] = {'newdata': newdata}
  44         return dict(key=key, newdata=newdata)
  45
  46
  47 @Controller('/foo/:key/:method', secure=False)
  48 class FooResourceDetail(RESTController):
  49     def list(self, key, method):
  50         return {'detail': (key, [method])}
  51
  52
  53 @ApiController('/rgw/proxy', secure=False)
  54 class GenerateControllerRoutesController(BaseController):
  55     @Proxy()
  56     def __call__(self, path, **params):
  57         pass
  58
  59
  60 @ApiController('/fooargs', secure=False)
  61 class FooArgs(RESTController):
  62     def set(self, code, name=None, opt1=None, opt2=None):
  63         return {'code': code, 'name': name, 'opt1': opt1, 'opt2': opt2}
  64
  65     @handle_rados_error('foo')
  66     def create(self, my_arg_name):
  67         return my_arg_name
  68
  69     def list(self):
  70         raise cherrypy.NotFound()
  71
  72
  73 # pylint: disable=blacklisted-name
  74 class Root(object):
  75     foo = FooResource()
  76     fooargs = FooArgs()
  77
  78
  79 class RESTControllerTest(ControllerTestCase):
  80
  81     @classmethod
  82     def setup_server(cls):
  83         cls.setup_controllers(
  84             [FooResource, FooResourceDetail, FooArgs, GenerateControllerRoutesController])
  85
  86     def test_empty(self):
  87         self._delete("/foo")
  88         self.assertStatus(204)
  89         self._get("/foo")
  90         self.assertStatus('200 OK')
  91         self.assertHeader('Content-Type', 'application/json')
  92         self.assertBody('[]')
  93
  94     def test_fill(self):
  95         sess_mock = RamSession()
  96         with patch('cherrypy.session', sess_mock, create=True):
  97             data = {'a': 'b'}
  98             for _ in range(5):
  99                 self._post("/foo", data)
 100                 self.assertJsonBody(data)
 101                 self.assertStatus(201)
 102                 self.assertHeader('Content-Type', 'application/json')
 103
 104             self._get("/foo")
 105             self.assertStatus('200 OK')
 106             self.assertHeader('Content-Type', 'application/json')
 107             self.assertJsonBody([data] * 5)
 108
 109             self._put('/foo/0', {'newdata': 'newdata'})
 110             self.assertStatus('200 OK')
 111             self.assertHeader('Content-Type', 'application/json')
 112             self.assertJsonBody({'newdata': 'newdata', 'key': '0'})
 113
 114     def test_not_implemented(self):
 115         self._put("/foo")
 116         self.assertStatus(404)
 117         body = self.json_body()
 118         self.assertIsInstance(body, dict)
 119         assert body['detail'] == "The path '/foo' was not found."
 120         assert '404' in body['status']
 121
 122     def test_args_from_json(self):
 123         self._put("/api/fooargs/hello", {'name': 'world'})
 124         self.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': None, 'opt2': None})
 125
 126         self._put("/api/fooargs/hello", {'name': 'world', 'opt1': 'opt1'})
 127         self.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': 'opt1', 'opt2': None})
 128
 129         self._put("/api/fooargs/hello", {'name': 'world', 'opt2': 'opt2'})
 130         self.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': None, 'opt2': 'opt2'})
 131
 132     def test_detail_route(self):
 133         self._get('/foo/default')
 134         self.assertJsonBody({'detail': ['default', []]})
 135
 136         self._get('/foo/default/default')
 137         self.assertJsonBody({'detail': ['default', ['default']]})
 138
 139         self._get('/foo/1/detail')
 140         self.assertJsonBody({'detail': ['1', ['detail']]})
 141
 142         self._post('/foo/1/detail', 'post-data')
 143         self.assertStatus(404)
 144
 145     def test_generate_controller_routes(self):
 146         # We just need to add this controller in setup_server():
 147         # noinspection PyStatementEffect
 148         # pylint: disable=pointless-statement
 149         GenerateControllerRoutesController
 150
 151
 152 class RequestLoggingToolTest(ControllerTestCase):
 153
 154     def __init__(self, *args, **kwargs):
 155         cherrypy.tools.request_logging = RequestLoggingTool()
 156         cherrypy.config.update({'tools.request_logging.on': True})
 157         super(RequestLoggingToolTest, self).__init__(*args, **kwargs)
 158
 159     @classmethod
 160     def setup_server(cls):
 161         cls.setup_controllers([FooResource])
 162
 163     def test_is_logged(self):
 164         with patch('logging.Logger.debug') as mock_logger_debug:
 165             self._put('/foo/0', {'newdata': 'xyz'})
 166             self.assertStatus(200)
 167             call_args_list = mock_logger_debug.call_args_list
 168             _, host, _, method, user, path = call_args_list[0][0]
 169             self.assertEqual(host, '127.0.0.1')
 170             self.assertEqual(method, 'PUT')
 171             self.assertIsNone(user)
 172             self.assertEqual(path, '/foo/0')
 173
 174
 175 class TestFunctions(unittest.TestCase):
 176
 177     def test_dict_contains_path(self):
 178         x = {'a': {'b': {'c': 'foo'}}}
 179         self.assertTrue(dict_contains_path(x, ['a', 'b', 'c']))
 180         self.assertTrue(dict_contains_path(x, ['a', 'b', 'c']))
 181         self.assertTrue(dict_contains_path(x, ['a']))
 182         self.assertFalse(dict_contains_path(x, ['a', 'c']))
 183         self.assertTrue(dict_contains_path(x, []))
 184
 185     def test_json_str_to_object(self):
 186         expected_result = {'a': 1, 'b': 'bbb'}
 187         self.assertEqual(expected_result, json_str_to_object('{"a": 1, "b": "bbb"}'))
 188         self.assertEqual(expected_result, json_str_to_object(b'{"a": 1, "b": "bbb"}'))
 189         self.assertEqual('', json_str_to_object(''))
 190         self.assertRaises(TypeError, json_str_to_object, None)
 191
 192     def test_partial_dict(self):
 193         expected_result = {'a': 1, 'c': 3}
 194         self.assertEqual(expected_result, partial_dict({'a': 1, 'b': 2, 'c': 3}, ['a', 'c']))
 195         self.assertEqual({}, partial_dict({'a': 1, 'b': 2, 'c': 3}, []))
 196         self.assertEqual({}, partial_dict({}, []))
 197         self.assertRaises(KeyError, partial_dict, {'a': 1, 'b': 2, 'c': 3}, ['d'])
 198         self.assertRaises(TypeError, partial_dict, None, ['a'])
 199         self.assertRaises(TypeError, partial_dict, {'a': 1, 'b': 2, 'c': 3}, None)
 200
 201     def test_dict_get(self):
 202         self.assertFalse(dict_get({'foo': {'bar': False}}, 'foo.bar'))
 203         self.assertIsNone(dict_get({'foo': {'bar': False}}, 'foo.bar.baz'))
 204         self.assertEqual(dict_get({'foo': {'bar': False}, 'baz': 'xyz'}, 'baz'), 'xyz')