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
158
159
160
|
# -*- coding: utf-8 -*-
import time
import rados
from ..controllers import Endpoint, RESTController, Router, Task
from ..services.ceph_service import SendCommandError
from ..services.exception import handle_rados_error, \
handle_send_command_error, serialize_dashboard_exception
from ..tests import ControllerTestCase
from ..tools import NotificationQueue, TaskManager, ViewCache
# pylint: disable=W0613
@Router('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])
@classmethod
def tearDownClass(cls):
NotificationQueue.stop()
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.json_body():
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.json_body()['detail'])
def test_404(self):
self._get('/foonot_found')
self.assertStatus(404)
self.assertIn('detail', self.json_body())
|
