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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
|
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2012 OpenStack, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Unit Tests for 'common' functons used through rpc code.
"""
import logging
import sys
from openstack.common import cfg
from openstack.common import exception
from openstack.common import importutils
from openstack.common import jsonutils
from openstack.common import rpc
from openstack.common.rpc import common as rpc_common
from tests import utils as test_utils
FLAGS = cfg.CONF
LOG = logging.getLogger(__name__)
def raise_exception():
raise Exception("test")
class FakeUserDefinedException(Exception):
def __init__(self):
Exception.__init__(self, "Test Message")
class RpcCommonTestCase(test_utils.BaseTestCase):
def test_serialize_remote_exception(self):
expected = {
'class': 'Exception',
'module': 'exceptions',
'message': 'test',
}
try:
raise_exception()
except Exception as exc:
failure = rpc_common.serialize_remote_exception(sys.exc_info())
failure = jsonutils.loads(failure)
#assure the traceback was added
self.assertEqual(expected['class'], failure['class'])
self.assertEqual(expected['module'], failure['module'])
self.assertEqual(expected['message'], failure['message'])
def test_serialize_remote_custom_exception(self):
def raise_custom_exception():
raise exception.OpenstackException()
expected = {
'class': 'OpenstackException',
'module': 'openstack.common.exception',
'message': exception.OpenstackException.message,
}
try:
raise_custom_exception()
except Exception as exc:
failure = rpc_common.serialize_remote_exception(sys.exc_info())
failure = jsonutils.loads(failure)
#assure the traceback was added
self.assertEqual(expected['class'], failure['class'])
self.assertEqual(expected['module'], failure['module'])
self.assertEqual(expected['message'], failure['message'])
def test_deserialize_remote_exception(self):
failure = {
'class': 'NotImplementedError',
'module': 'exceptions',
'message': '',
'tb': ['raise NotImplementedError'],
}
serialized = jsonutils.dumps(failure)
after_exc = rpc_common.deserialize_remote_exception(FLAGS, serialized)
self.assertTrue(isinstance(after_exc, NotImplementedError))
#assure the traceback was added
self.assertTrue('raise NotImplementedError' in unicode(after_exc))
def test_deserialize_remote_custom_exception(self):
failure = {
'class': 'OpenstackException',
'module': 'openstack.common.exception',
'message': exception.OpenstackException.message,
'tb': ['raise OpenstackException'],
}
serialized = jsonutils.dumps(failure)
after_exc = rpc_common.deserialize_remote_exception(FLAGS, serialized)
self.assertTrue(isinstance(after_exc, exception.OpenstackException))
self.assertTrue('An unknown' in unicode(after_exc))
#assure the traceback was added
self.assertTrue('raise OpenstackException' in unicode(after_exc))
def test_deserialize_remote_exception_bad_module(self):
failure = {
'class': 'popen2',
'module': 'os',
'kwargs': {'cmd': '/bin/echo failed'},
'message': 'foo',
}
serialized = jsonutils.dumps(failure)
after_exc = rpc_common.deserialize_remote_exception(FLAGS, serialized)
self.assertTrue(isinstance(after_exc, rpc_common.RemoteError))
def test_deserialize_remote_exception_user_defined_exception(self):
"""Ensure a user defined exception can be deserialized."""
self.config(allowed_rpc_exception_modules=[self.__class__.__module__])
failure = {
'class': 'FakeUserDefinedException',
'module': self.__class__.__module__,
'tb': ['raise FakeUserDefinedException'],
}
serialized = jsonutils.dumps(failure)
after_exc = rpc_common.deserialize_remote_exception(FLAGS, serialized)
self.assertTrue(isinstance(after_exc, FakeUserDefinedException))
#assure the traceback was added
self.assertTrue('raise FakeUserDefinedException' in unicode(after_exc))
def test_deserialize_remote_exception_cannot_recreate(self):
"""Ensure a RemoteError is returned on initialization failure.
If an exception cannot be recreated with it's original class then a
RemoteError with the exception informations should still be returned.
"""
self.config(allowed_rpc_exception_modules=[self.__class__.__module__])
failure = {
'class': 'FakeIDontExistException',
'module': self.__class__.__module__,
'tb': ['raise FakeIDontExistException'],
}
serialized = jsonutils.dumps(failure)
after_exc = rpc_common.deserialize_remote_exception(FLAGS, serialized)
self.assertTrue(isinstance(after_exc, rpc_common.RemoteError))
#assure the traceback was added
self.assertTrue('raise FakeIDontExistException' in unicode(after_exc))
def test_loading_old_nova_config(self):
self.config(rpc_backend='nova.rpc.impl_qpid')
rpc._RPCIMPL = None
self.mod = None
def fake_import_module(m):
if not self.mod:
# The first time import_module is called, before the replace()
self.mod = m
raise ImportError
self.mod = m
orig_import_module = importutils.import_module
importutils.import_module = fake_import_module
rpc._get_impl()
importutils.import_module = orig_import_module
self.assertEqual(self.mod, 'nova.openstack.common.rpc.impl_qpid')
def test_queue_get_for(self):
self.assertEqual(rpc.queue_get_for(None, 'a', 'b'), 'a.b')
self.assertEqual(rpc.queue_get_for(None, 'a', None), 'a')
def test_client_exception(self):
e = None
try:
try:
raise ValueError()
except Exception:
raise rpc_common.ClientException()
except rpc_common.ClientException, e:
pass
self.assertTrue(isinstance(e, rpc_common.ClientException))
self.assertTrue(e._exc_info[1], ValueError)
def test_catch_client_exception(self):
def naughty(param):
int(param)
e = None
try:
rpc_common.catch_client_exception([ValueError], naughty, 'a')
except rpc_common.ClientException, e:
pass
self.assertTrue(isinstance(e, rpc_common.ClientException))
self.assertTrue(isinstance(e._exc_info[1], ValueError))
def test_catch_client_exception_other(self):
class FooException(Exception):
pass
def naughty():
raise FooException()
e = None
self.assertRaises(FooException,
rpc_common.catch_client_exception,
[ValueError], naughty)
def test_client_exceptions_decorator(self):
class FooException(Exception):
pass
@rpc_common.client_exceptions(FooException)
def naughty():
raise FooException()
@rpc_common.client_exceptions(FooException)
def really_naughty():
raise ValueError()
self.assertRaises(rpc_common.ClientException, naughty)
self.assertRaises(ValueError, really_naughty)
|
