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
|
#
# Copyright (c) ZeroC, Inc. All rights reserved.
#
import os, sys, traceback, time
import Ice, Test
def test(b):
if not b:
raise RuntimeError('test assertion failed')
class TestI(Test.TestIntf):
def requestFailedException(self, current=None):
return None
def unknownUserException(self, current=None):
return None
def unknownLocalException(self, current=None):
return None
def unknownException(self, current=None):
return None
def localException(self, current=None):
return None
def userException(self, current=None):
return None
def pythonException(self, current=None):
return None
def unknownExceptionWithServantException(self, current=None):
f = Ice.Future()
f.set_exception(Ice.ObjectNotExistException())
return f
def impossibleException(self, throw, current=None):
f = Ice.Future()
if throw:
f.set_exception(Test.TestImpossibleException())
else:
#
# Return a value so we can be sure that the stream position
# is reset correctly if finished() throws.
#
f.set_result("Hello")
return f
def intfUserException(self, throw, current=None):
f = Ice.Future()
if throw:
f.set_exception(Test.TestIntfUserException())
else:
#
# Return a value so we can be sure that the stream position
# is reset correctly if finished() throws.
#
f.set_result("Hello")
return f
def asyncResponse(self, current=None):
#
# We can't do this with futures.
#
#return Ice.Future.completed(None)
raise Ice.ObjectNotExistException()
def asyncException(self, current=None):
#
# We can't do this with futures.
#
#f = Ice.Future()
#f.set_exception(Test.TestIntfUserException())
#return f
raise Ice.ObjectNotExistException()
def shutdown(self, current=None):
current.adapter.deactivate()
class CookieI(Test.Cookie):
def message(self):
return 'blahblah'
class ServantLocatorI(Ice.ServantLocator):
def __init__(self, category):
self._deactivated = False
self._category = category
self._requestId = -1
def __del__(self):
test(self._deactivated)
def locate(self, current):
test(not self._deactivated)
test(current.id.category == self._category or self._category == "")
if current.id.name == "unknown":
return None
if current.id.name == "invalidReturnValue":
return (45, 12)
if current.id.name == "invalidReturnType":
return "invalid"
test(current.id.name == "locate" or current.id.name == "finished")
if current.id.name == "locate":
self.exception(current)
#
# Ensure locate() is only called once per request.
#
test(self._requestId == -1)
self._requestId = current.requestId
return (TestI(), CookieI())
def finished(self, current, servant, cookie):
test(not self._deactivated)
#
# Ensure finished() is only called once per request.
#
test(self._requestId == current.requestId)
self._requestId = -1
test(current.id.category == self._category or self._category == "")
test(current.id.name == "locate" or current.id.name == "finished")
if current.id.name == "finished":
self.exception(current)
test(isinstance(cookie, Test.Cookie))
test(cookie.message() == 'blahblah')
def deactivate(self, category):
test(not self._deactivated)
self._deactivated = True
def exception(self, current):
if current.operation == "ice_ids":
raise Test.TestIntfUserException()
elif current.operation == "requestFailedException":
raise Ice.ObjectNotExistException()
elif current.operation == "unknownUserException":
raise Ice.UnknownUserException("reason")
elif current.operation == "unknownLocalException":
raise Ice.UnknownLocalException("reason")
elif current.operation == "unknownException":
raise Ice.UnknownException("reason")
elif current.operation == "userException":
raise Test.TestIntfUserException()
elif current.operation == "localException":
raise Ice.SocketException(0)
elif current.operation == "pythonException":
raise RuntimeError("message")
elif current.operation == "unknownExceptionWithServantException":
raise Ice.UnknownException("reason")
elif current.operation == "impossibleException":
raise Test.TestIntfUserException() # Yes, it really is meant to be TestIntfUserException.
elif current.operation == "intfUserException":
raise Test.TestImpossibleException() # Yes, it really is meant to be TestImpossibleException.
elif current.operation == "asyncResponse":
raise Test.TestImpossibleException()
elif current.operation == "asyncException":
raise Test.TestImpossibleException()
|
