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
|
# -*- coding: utf-8 -*-
#
# Copyright (c) ZeroC, Inc. All rights reserved.
#
#
# Publisher/subscriber test cases, publisher publishes on TestIceStorm1 instance(s) and
# the subscriber subscribes to the TestIceStorm2 instance(s)
#
pub1Sub2Oneway=ClientServerTestCase(client=Publisher("TestIceStorm1"), server=Subscriber("TestIceStorm2"))
pub1Sub2Batch=ClientServerTestCase(client=Publisher("TestIceStorm1"), server=Subscriber("TestIceStorm2", args=["-b"]))
pub1Sub1Oneway=ClientServerTestCase(client=Publisher("TestIceStorm1"), server=Subscriber("TestIceStorm1"))
class IceStormFederation2TestCase(IceStormTestCase):
def runClientSide(self, current):
current.write("setting up the topics... ")
self.runadmin(current, "create TestIceStorm1/fed1 TestIceStorm2/fed1; link TestIceStorm1/fed1 TestIceStorm2/fed1")
current.writeln("ok")
#
# Test oneway subscribers.
#
current.write("testing federation with oneway subscribers... ")
pub1Sub2Oneway.run(current)
current.writeln("ok")
#
# Test batch oneway subscribers.
#
current.write("testing federation with batch subscribers... ")
pub1Sub2Batch.run(current)
current.writeln("ok")
#
# Test #2:
#
# Stop and restart the service and repeat the test. This ensures that
# the database is correct.
#
current.write("restarting services to ensure that the database content is preserved... ")
self.restartIceStorm(current)
current.writeln("ok")
#
# Test oneway subscribers.
#
current.write("retesting federation with oneway subscribers... ")
pub1Sub2Oneway.run(current)
current.writeln("ok")
#
# Test batch oneway subscribers.
#
current.write("retesting federation with batch subscribers... ")
pub1Sub2Batch.run(current)
current.writeln("ok")
#
# Shutdown icestorm.
#
self.stopIceStorm(current)
icestorm1 = [icestorm for icestorm in self.icestorm if icestorm.getInstanceName() == "TestIceStorm1"]
icestorm2 = [icestorm for icestorm in self.icestorm if icestorm.getInstanceName() == "TestIceStorm2"]
#
# Restart the first server and publish some events. Attach a
# subscriber to the channel and make sure the events are received.
#
# Then re-start the linked downstream server and publish the events.
# Ensure they are received by the linked server.
#
if self.getName().find("replicated") == -1:
current.write("restarting only one IceStorm server... ")
icestorm1[0].start(current)
current.writeln("ok")
#
# Test oneway subscribers.
#
current.write("testing that the federation link reports an error... ")
pub1Sub1Oneway.run(current)
# Give some time for the output to be sent.
time.sleep(2)
icestorm1[0].expect(current, "topic.fed1.*subscriber offline")
current.writeln("ok")
current.write("starting downstream icestorm server... ")
icestorm2[0].start(current)
current.writeln("ok")
#
# Need to sleep for at least the discard interval.
#
time.sleep(3)
#
# Test oneway subscribers.
#
current.write("testing link is reestablished... ")
pub1Sub2Oneway.run(current)
current.writeln("ok")
try:
icestorm1[0].expect(current, "topic.fed1.*subscriber offline", timeout=1)
assert False
except Expect.TIMEOUT:
pass
self.stopIceStorm(current)
#
# Test #4:
#
# Trash the TestIceStorm2 database. Then restart the servers and
# verify that the link is removed.
#
current.write("destroying the downstream IceStorm service database... ")
for s in icestorm2:
s.teardown(current, True)
s.setup(current)
current.writeln("ok")
current.write("restarting IceStorm servers... ")
self.startIceStorm(current)
current.writeln("ok")
current.write("checking link still exists... ")
line = self.runadmin(current, "links TestIceStorm1", quiet=True)
if not re.compile("fed1 with cost 0").search(line):
raise RuntimeError("unexpected output (`{0}')".format(line))
current.writeln("ok")
current.write("publishing some events... ")
# The publisher must be run twice because all the events can be
# sent out in one batch to the linked subscriber which means that
# the link is not reaped until the next batch is
# sent. Furthermore, with a replicated IceStorm both sets of
# events must be set to the same replica.
Publisher("TestIceStorm1", args=["--count", 2]).run(current)
current.writeln("ok")
# Give the unsubscription time to propagate.
time.sleep(1)
# Verify that the link has disappeared.
current.write("verifying that the link has been destroyed... ")
line = self.runadmin(current, "links TestIceStorm1")
nRetry = 5
while len(line) > 0 and nRetry > 0:
line = self.runadmin(current, "links TestIceStorm1")
time.sleep(1) # Give more time for unsubscription to propagate.
nRetry -= 1
if len(line) > 0:
raise RuntimeError("unexpected output (`{0}')".format(line))
current.writeln("ok")
#
# Destroy the remaining topic.
#
current.write("destroying topics... ")
self.runadmin(current, "destroy TestIceStorm1/fed1links TestIceStorm1", quiet=True)
current.writeln("ok")
self.stopIceStorm(current)
# Override ReplicatedPublishEndpoints property to empty for testing without replicated publisher
props = { 'IceStorm.Discard.Interval' : 2 }
nonRepProps = { 'IceStorm.Discard.Interval' : 2, 'IceStorm.ReplicatedPublishEndpoints' : '' }
TestSuite(__file__, [
IceStormFederation2TestCase("persistent", icestorm=
[IceStorm("TestIceStorm1", quiet=True, props=props),
IceStorm("TestIceStorm2", quiet=True,portnum=20, props=props)]),
IceStormFederation2TestCase("replicated with non-replicated publisher", icestorm=
[IceStorm("TestIceStorm1", i, 3, quiet=True, props=nonRepProps) for i in range(0,3)] +
[IceStorm("TestIceStorm2", i, 3, quiet=True, portnum=20, props=nonRepProps) for i in range(0,3)]),
IceStormFederation2TestCase("replicated with replicated publisher", icestorm=
[IceStorm("TestIceStorm1", i, 3, quiet=True, props=props) for i in range(0,3)] +
[IceStorm("TestIceStorm2", i, 3, quiet=True, portnum=20, props=props) for i in range(0,3)]),
], multihost=False)
|
