1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
# -*- coding: utf-8 -*-
from __future__ import absolute_import
import time
import rados
from . import ControllerTestCase
from ..services.ceph_service import SendCommandError
from ..controllers import RESTController, Controller, Task, Endpoint
from ..services.exception import handle_rados_error, handle_send_command_error, \
serialize_dashboard_exception
from ..tools import ViewCache, TaskManager, NotificationQueue
# pylint: disable=W0613
@Controller('foo', secure=False)
class FooResource(RESTController):
@Endpoint()
@handle_rados_error('foo')
def no_exception(self, param1, param2):
return [param1, param2]
@Endpoint()
@handle_rados_error('foo')
def error_foo_controller(self):
raise rados.OSError('hi', errno=-42)
@Endpoint()
@handle_send_command_error('foo')
def error_send_command(self):
raise SendCommandError('hi', 'prefix', {}, -42)
@Endpoint()
def error_generic(self):
raise rados.Error('hi')
@Endpoint()
def vc_no_data(self):
@ViewCache(timeout=0)
def _no_data():
time.sleep(0.2)
_no_data()
assert False
@handle_rados_error('foo')
@Endpoint()
def vc_exception(self):
@ViewCache(timeout=10)
def _raise():
raise rados.OSError('hi', errno=-42)
_raise()
assert False
@Endpoint()
def internal_server_error(self):
return 1/0
@handle_send_command_error('foo')
def list(self):
raise SendCommandError('list', 'prefix', {}, -42)
@Endpoint()
@Task('task_exceptions/task_exception', {1: 2}, 1.0,
exception_handler=serialize_dashboard_exception)
@handle_rados_error('foo')
def task_exception(self):
raise rados.OSError('hi', errno=-42)
@Endpoint()
def wait_task_exception(self):
ex, _ = TaskManager.list('task_exceptions/task_exception')
return bool(len(ex))
# pylint: disable=C0102
class Root(object):
foo = FooResource()
class RESTControllerTest(ControllerTestCase):
@classmethod
def setup_server(cls):
NotificationQueue.start_queue()
TaskManager.init()
cls.setup_controllers([FooResource])
def test_no_exception(self):
self._get('/foo/no_exception/a/b')
self.assertStatus(200)
self.assertJsonBody(
['a', 'b']
)
def test_error_foo_controller(self):
self._get('/foo/error_foo_controller')
self.assertStatus(400)
self.assertJsonBody(
{'detail': '[errno -42] hi', 'code': "42", 'component': 'foo'}
)
def test_error_send_command(self):
self._get('/foo/error_send_command')
self.assertStatus(400)
self.assertJsonBody(
{'detail': '[errno -42] hi', 'code': "42", 'component': 'foo'}
)
def test_error_send_command_list(self):
self._get('/foo/')
self.assertStatus(400)
self.assertJsonBody(
{'detail': '[errno -42] list', 'code': "42", 'component': 'foo'}
)
def test_error_foo_generic(self):
self._get('/foo/error_generic')
self.assertJsonBody({'detail': 'hi', 'code': 'Error', 'component': None})
self.assertStatus(400)
def test_viewcache_no_data(self):
self._get('/foo/vc_no_data')
self.assertStatus(200)
self.assertJsonBody({'status': ViewCache.VALUE_NONE, 'value': None})
def test_viewcache_exception(self):
self._get('/foo/vc_exception')
self.assertStatus(400)
self.assertJsonBody(
{'detail': '[errno -42] hi', 'code': "42", 'component': 'foo'}
)
def test_task_exception(self):
self._get('/foo/task_exception')
self.assertStatus(400)
self.assertJsonBody(
{'detail': '[errno -42] hi', 'code': "42", 'component': 'foo',
'task': {'name': 'task_exceptions/task_exception', 'metadata': {'1': 2}}}
)
self._get('/foo/wait_task_exception')
while self.jsonBody():
time.sleep(0.5)
self._get('/foo/wait_task_exception')
def test_internal_server_error(self):
self._get('/foo/internal_server_error')
self.assertStatus(500)
self.assertIn('unexpected condition', self.jsonBody()['detail'])
def test_404(self):
self._get('/foonot_found')
self.assertStatus(404)
self.assertIn('detail', self.jsonBody())
|
