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
|
#!/usr/bin/env python3
#
# Copyright (c) ZeroC, Inc. All rights reserved.
#
import sys
import threading
import Ice
import Glacier2
from TestHelper import TestHelper
TestHelper.loadSlice("Callback.ice")
import Test
def test(b):
if not b:
raise RuntimeError('test assertion failed')
class CallbackReceiverI(Test.CallbackReceiver):
def __init__(self):
self._received = False
self._cond = threading.Condition()
def callback(self, current):
with self._cond:
self._received = True
self._cond.notify()
def waitForCallback(self):
with self._cond:
while not self._received:
self._cond.wait()
self._received = False;
class Application(Glacier2.Application):
def __init__(self, helper):
Glacier2.Application.__init__(self)
self._restart = 0
self._destroyed = False
self._receiver = CallbackReceiverI()
self._helper = helper
def createSession(self):
return Glacier2.SessionPrx.uncheckedCast(self.router().createSession("userid", "abc123"))
def runWithSession(self, args):
test(self.router());
test(self.categoryForClient());
test(self.objectAdapter());
if self._restart == 0:
sys.stdout.write("testing Glacier2::Application restart... ")
sys.stdout.flush()
base = self.communicator().stringToProxy("callback:{0}".format(
self._helper.getTestEndpoint(properties=self.communicator().getProperties())));
callback = Test.CallbackPrx.uncheckedCast(base)
self._restart += 1
if self._restart < 5:
receiver = Test.CallbackReceiverPrx.uncheckedCast(self.addWithUUID(self._receiver))
callback.initiateCallback(receiver)
self._receiver.waitForCallback()
self.restart()
print("ok")
sys.stdout.write("testing server shutdown... ")
callback.shutdown()
print("ok")
return 0
def sessionDestroyed(self):
self._destroyed = True
class Client(TestHelper):
def run(self, args):
initData = Ice.InitializationData()
initData.properties = self.createTestProperties(args)
initData.properties.setProperty("Ice.Default.Router", "Glacier2/router:{0}".format(
self.getTestEndpoint(properties=initData.properties, num=50)))
app = Application(self)
status = app.main(sys.argv, initData=initData)
test(status == 0)
test(app._restart == 5)
test(app._destroyed)
initData.properties.setProperty("Ice.Default.Router", "")
with self.initialize(initData=initData) as communicator:
sys.stdout.write("testing stringToProxy for process object... ")
sys.stdout.flush()
processBase = communicator.stringToProxy("Glacier2/admin -f Process:{0}".format(
self.getTestEndpoint(properties=initData.properties, num=51)))
print("ok")
sys.stdout.write("testing checked cast for admin object... ")
sys.stdout.flush()
process = Ice.ProcessPrx.checkedCast(processBase)
test(process)
print("ok")
sys.stdout.write("testing Glacier2 shutdown... ")
sys.stdout.flush()
process.shutdown()
try:
process.ice_ping()
test(False)
except Ice.LocalException:
print("ok")
|
