1 # -*- coding: utf-8 -*-
6 from cherrypy
.lib
.sessions
import RamSession
11 from unittest
.mock
import patch
13 from ..controllers
import APIRouter
, BaseController
, Proxy
, RESTController
, Router
14 from ..controllers
._version
import APIVersion
15 from ..services
.exception
import handle_rados_error
16 from ..tests
import ControllerTestCase
17 from ..tools
import dict_contains_path
, dict_get
, json_str_to_object
, \
18 merge_list_of_dicts_by_key
, partial_dict
21 # pylint: disable=W0613
22 @Router('/foo', secure
=False)
23 class FooResource(RESTController
):
27 return FooResource
.elems
30 FooResource
.elems
.append({'a': a
})
34 return {'detail': (key
, [])}
36 def delete(self
, key
):
37 del FooResource
.elems
[int(key
)]
39 def bulk_delete(self
):
40 FooResource
.elems
= []
42 def set(self
, key
, newdata
):
43 FooResource
.elems
[int(key
)] = {'newdata': newdata
}
44 return dict(key
=key
, newdata
=newdata
)
47 @Router('/foo/:key/:method', secure
=False)
48 class FooResourceDetail(RESTController
):
49 def list(self
, key
, method
):
50 return {'detail': (key
, [method
])}
53 @APIRouter('/rgw/proxy', secure
=False)
54 class GenerateControllerRoutesController(BaseController
):
56 def __call__(self
, path
, **params
):
60 @APIRouter('/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
}
65 @handle_rados_error('foo')
66 def create(self
, my_arg_name
):
70 raise cherrypy
.NotFound()
74 foo_resource
= FooResource()
78 class RESTControllerTest(ControllerTestCase
):
81 def setup_server(cls
):
82 cls
.setup_controllers(
83 [FooResource
, FooResourceDetail
, FooArgs
, GenerateControllerRoutesController
])
87 self
.assertStatus(204)
89 self
.assertStatus('200 OK')
90 self
.assertHeader('Content-Type', APIVersion
.DEFAULT
.to_mime_type())
94 sess_mock
= RamSession()
95 with
patch('cherrypy.session', sess_mock
, create
=True):
98 self
._post
("/foo", data
)
99 self
.assertJsonBody(data
)
100 self
.assertStatus(201)
101 self
.assertHeader('Content-Type', APIVersion
.DEFAULT
.to_mime_type())
104 self
.assertStatus('200 OK')
105 self
.assertHeader('Content-Type', APIVersion
.DEFAULT
.to_mime_type())
106 self
.assertJsonBody([data
] * 5)
108 self
._put
('/foo/0', {'newdata': 'newdata'})
109 self
.assertStatus('200 OK')
110 self
.assertHeader('Content-Type', APIVersion
.DEFAULT
.to_mime_type())
111 self
.assertJsonBody({'newdata': 'newdata', 'key': '0'})
113 def test_not_implemented(self
):
115 self
.assertStatus(404)
116 body
= self
.json_body()
117 self
.assertIsInstance(body
, dict)
118 assert body
['detail'] == "The path '/foo' was not found."
119 assert '404' in body
['status']
121 def test_args_from_json(self
):
122 self
._put
("/api/fooargs/hello", {'name': 'world'})
123 self
.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': None, 'opt2': None})
125 self
._put
("/api/fooargs/hello", {'name': 'world', 'opt1': 'opt1'})
126 self
.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': 'opt1', 'opt2': None})
128 self
._put
("/api/fooargs/hello", {'name': 'world', 'opt2': 'opt2'})
129 self
.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': None, 'opt2': 'opt2'})
131 def test_detail_route(self
):
132 self
._get
('/foo/default')
133 self
.assertJsonBody({'detail': ['default', []]})
135 self
._get
('/foo/default/default')
136 self
.assertJsonBody({'detail': ['default', ['default']]})
138 self
._get
('/foo/1/detail')
139 self
.assertJsonBody({'detail': ['1', ['detail']]})
141 self
._post
('/foo/1/detail', 'post-data')
142 self
.assertStatus(404)
144 def test_generate_controller_routes(self
):
145 # We just need to add this controller in setup_server():
146 # noinspection PyStatementEffect
147 # pylint: disable=pointless-statement
148 GenerateControllerRoutesController
151 class RequestLoggingToolTest(ControllerTestCase
):
153 _request_logging
= True
156 def setup_server(cls
):
157 cls
.setup_controllers([FooResource
])
159 def test_is_logged(self
):
160 with
patch('logging.Logger.debug') as mock_logger_debug
:
161 self
._put
('/foo/0', {'newdata': 'xyz'})
162 self
.assertStatus(200)
163 call_args_list
= mock_logger_debug
.call_args_list
164 _
, host
, _
, method
, user
, path
= call_args_list
[0][0]
165 self
.assertEqual(host
, '127.0.0.1')
166 self
.assertEqual(method
, 'PUT')
167 self
.assertIsNone(user
)
168 self
.assertEqual(path
, '/foo/0')
171 class TestFunctions(unittest
.TestCase
):
173 def test_dict_contains_path(self
):
174 x
= {'a': {'b': {'c': 'foo'}}}
175 self
.assertTrue(dict_contains_path(x
, ['a', 'b', 'c']))
176 self
.assertTrue(dict_contains_path(x
, ['a', 'b', 'c']))
177 self
.assertTrue(dict_contains_path(x
, ['a']))
178 self
.assertFalse(dict_contains_path(x
, ['a', 'c']))
179 self
.assertTrue(dict_contains_path(x
, []))
181 def test_json_str_to_object(self
):
182 expected_result
= {'a': 1, 'b': 'bbb'}
183 self
.assertEqual(expected_result
, json_str_to_object('{"a": 1, "b": "bbb"}'))
184 self
.assertEqual(expected_result
, json_str_to_object(b
'{"a": 1, "b": "bbb"}'))
185 self
.assertEqual('', json_str_to_object(''))
186 self
.assertRaises(TypeError, json_str_to_object
, None)
188 def test_partial_dict(self
):
189 expected_result
= {'a': 1, 'c': 3}
190 self
.assertEqual(expected_result
, partial_dict({'a': 1, 'b': 2, 'c': 3}, ['a', 'c']))
191 self
.assertEqual({}, partial_dict({'a': 1, 'b': 2, 'c': 3}, []))
192 self
.assertEqual({}, partial_dict({}, []))
193 self
.assertRaises(KeyError, partial_dict
, {'a': 1, 'b': 2, 'c': 3}, ['d'])
194 self
.assertRaises(TypeError, partial_dict
, None, ['a'])
195 self
.assertRaises(TypeError, partial_dict
, {'a': 1, 'b': 2, 'c': 3}, None)
197 def test_dict_get(self
):
198 self
.assertFalse(dict_get({'foo': {'bar': False}}, 'foo.bar'))
199 self
.assertIsNone(dict_get({'foo': {'bar': False}}, 'foo.bar.baz'))
200 self
.assertEqual(dict_get({'foo': {'bar': False}, 'baz': 'xyz'}, 'baz'), 'xyz')
202 def test_merge_list_of_dicts_by_key(self
):
203 expected_result
= [{'a': 1, 'b': 2, 'c': 3}, {'a': 4, 'b': 5, 'c': 6}]
204 self
.assertEqual(expected_result
, merge_list_of_dicts_by_key(
205 [{'a': 1, 'b': 2}, {'a': 4, 'b': 5}], [{'a': 1, 'c': 3}, {'a': 4, 'c': 6}], 'a'))
207 expected_result
= [{'a': 1, 'b': 2}, {'a': 4, 'b': 5, 'c': 6}]
208 self
.assertEqual(expected_result
, merge_list_of_dicts_by_key(
209 [{'a': 1, 'b': 2}, {'a': 4, 'b': 5}], [{}, {'a': 4, 'c': 6}], 'a'))
210 self
.assertRaises(TypeError, merge_list_of_dicts_by_key
, None)