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
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
|
#########################################################################
#
# Date: Aug. 2003 Author: Daniel Stoffler
#
# stoffler@scripps.edu
#
# Copyright: Daniel Stoffler and TSRI
#
#########################################################################
#NOTE: after every test, the editor gets destroyed and restarts
import sys
from time import sleep
from NetworkEditor.net import Network
from NetworkEditor.simpleNE import NetworkBuilder
from NetworkEditor.items import NetworkNode
from NetworkEditor.Tests.nodes import DialNode, PassNode, CounterNode, \
EvalNode, PrintNode, IterateNode
ed = None
def setUp():
global ed
ed = NetworkBuilder("test builder1", withShell=0,
visibleWidth=400, visibleHeight=300)
ed.master.update()
ed.configure(withThreads=0)
def tearDown():
ed.exit_cb()
import gc
gc.collect()
def pause(sleepTime=None):
if sleepTime is None:
from NetworkEditor.Tests import pauseLength as sleepTime
ed.master.update()
sleep(sleepTime)
def test_01_runNetwork():
net = ed.currentNetwork
node1 = DialNode()
net.addNode(node1, 100, 50)
pause()
assert node1.outputPorts[0].data == 'no data yet',\
"Expected 'no data yet', got %s"%node1.outputPorts[0].data
net.run()
assert node1.outputPorts[0].data != 'no data yet',\
"data is %s"%node1.outputPorts[0].data
def test_02_runNode():
net = ed.currentNetwork
node1 = DialNode()
net.addNode(node1, 100, 50)
pause()
assert node1.outputPorts[0].data == 'no data yet',\
"Expected 'no data yet', got %s"%node1.outputPorts[0].data
node1.run() # should not run
assert node1.outputPorts[0].data != 'no data yet',\
"data is %s"%node1.outputPorts[0].data
## def test_scheduleNode():
## net = ed.currentNetwork
## node1 = OutNode()
## node2 = CounterNode()
## net.addNode(node1, 50, 50)
## net.addNode(node2, 50, 150)
## net.connectNodes(node1, node2, 0, 0)
## pause()
## assert node2.counter == 0
## net.run() # run once
## print node2.counter
## assert node2.counter == 1
## node1.schedule()
## assert node2.counter == 2
#def test_03_freezeNetwork():
# net = Network('freezeNetwork')
# ed.addNetwork(net)
# node1 = EvalNode()
# net.addNode(node1, 50, 50)
# node1.toggleNodeExpand_cb()
# # freeze network
# net.freeze()
# pause()
#
# net.run()
# node1.inputPorts[0].widget.set('range(2000)')
# assert node1.outputPorts[0].data == 'no data yet',\
# "Expected 'no data yet', got %s"%node1.outputPorts[0].data
# net.unfreeze()
# pause()
# assert net.execStatus == 'waiting',\
# "Expected 'waiting', got '%s'"%net.execStatus
# assert node1.outputPorts[0].data == 'no data yet',\
# "Expected 'no data yet', got %s"%node1.outputPorts[0].data
#
# net.freeze(updateGUI=1)
# fill = node1.network.canvas.itemconfig(node1.id)['fill'][-1]
# assert fill =='#b6d3f6',"Expected '#b6d3f6', got '%s'"%fill
# net.unfreeze(updateGUI=1)
# fill = node1.network.canvas.itemconfig(node1.id)['fill'][-1]
# assert fill =='gray85', "Expected 'gray85', got '%s'"%fill
#
# # use editor call to freeze
# ed.freezeNetwork()
# pause()
# assert ed.isFrozenTk.get() == 1,"Expected 1, got %s"%ed.isFrozenTk.get()
# fill = node1.network.canvas.itemconfig(node1.id)['fill'][-1]
# assert fill =='#b6d3f6', "Expected '#b6d3f6', got '%s'"%fill
# node1.inputPorts[0].widget.set('range(200)')
# assert node1.outputPorts[0].data == 'no data yet',\
# "Expected 'no data yet', got %s"%node1.outputPorts[0].data
# ed.unfreezeNetwork()
# pause()
# assert ed.isFrozenTk.get() == 0,"Expected 0, got %s"%ed.isFrozenTk.get()
# fill = node1.network.canvas.itemconfig(node1.id)['fill'][-1]
# assert fill =='gray85', "Expected 'gray85', got '%s'"%fill
# #This triggers the execution
# assert node1.outputPorts[0].data == 'no data yet',\
# "Expected 'no data yet', got %s"%node1.outputPorts[0].data
#
# # add second network
# ed.freezeNetwork()
# assert ed.isFrozenTk.get()==1,\
# "Expected 1, got %s"%str(ed.isFrozenTk.get())
# pause()
# net1 = Network('freeze1Network')
# net1.runOnNewData.value = True
#
# ed.addNetwork(net1)
# assert ed.isFrozenTk.get() == 0,"Expected 0, got %s"%ed.isFrozenTk.get()
#
# node2 = DialNode()
# net1.addNode(node2, 50, 50)
# node2.inputPorts[0].widget.set(100, run=1)
# pause()
# assert node2.outputPorts[0].data == 100,\
# "Expected 100, got %s"%node2.outputPorts[0].data
# ed.deleteNetwork(net1)
#
# # got back to first network
# pause()
# ed.setNetwork(net)
# assert ed.isFrozenTk.get()==1,\
# "Expected 1, got %s"%str(ed.isFrozenTk.get())
# ed.deleteNetwork(net)
def test_04_setWidgetValue():
# add a dial node, set it's value to 10.0 and see if the node outputs
net = ed.currentNetwork
net.runOnNewData.value = True
node1 = DialNode()
net.addNode(node1, 100, 75)
pause()
out = node1.outputPorts[0]
# should not output any data right now
assert out.data == 'no data yet',"Expected 'no data yet', got %s"%out.data
dial1 = node1.inputPorts[0].widget
dial1.set(10.0)
pause()
# now the output port should have data
assert out.data == 10.0, "Expected 10.0, got %s"%out.data
def test_05_runChildUponConnect():
# add a dial node, set it's value to 10.0, connect a Pass node and see if
# the Pass node outputs 10.0
net = ed.currentNetwork
ed.currentNetwork.runOnNewData.value = True
node1 = DialNode()
net.addNode(node1, 50, 50)
dial1 = node1.inputPorts[0].widget
dial1.set(10.0)
# now connect a Pass node and check if its output port has 10.0
node2 = PassNode()
net.addNode(node2, 50, 150)
net.connectNodes(node1, node2, 0, 0)
pause()
assert node2.outputPorts[0].data == 10.0,\
"Expected 10.0, got %s"%node2.outputPorts[0].data
def test_06_TriggerOutputPort():
# test execution of special ports: a network with a dial1 node and
# a dial2 node connected to a Counter node. Then we connect special
# outputport of dial1 to the special input port of Counter and run
# dial1 (not dial2), which should trigger the counter node to run again
net = ed.currentNetwork
ed.currentNetwork.runOnNewData.value = True
node1 = DialNode()
net.addNode(node1, 50, 50)
pause()
dial1 = node1.inputPorts[0].widget
dial1.set(10.0)
node2 = DialNode()
net.addNode(node2, 150, 50)
node3 = CounterNode()
net.addNode(node3, 150, 190)
net.connectNodes(node2, node3, 0, 0)
dial2 = node2.inputPorts[0].widget
# set dial2, which should increase the counter node by 1
dial2.set(20.0)
assert node3.counter == 1, "Expected 1, got %s"%node3.counter
pause()
# show special ports on dial 1 and counter
node1.showSpecialPorts()
node3.showSpecialPorts()
# connect dial1 with counter
net.specialConnectNodes(node1, node3, 0, 0)
pause()
# counter should still be 1
assert node3.counter == 1, "Expected 1, got %s"%node3.counter
# now change value in dial1 which should trigger specialoutputport
dial1.set(15.0)
assert node3.counter == 2, "Expected 2, got %s"%node3.counter
def test_07_basicScheduling():
net = Network('basicScheduling')
ed.addNetwork(net)
net = ed.currentNetwork
node1 = EvalNode()
net.addNode(node1, 50, 50)
node2 = IterateNode()
net.addNode(node2, 50, 150)
node3 = PrintNode()
net.addNode(node3, 50, 250)
net.connectNodes(node1, node2, 0, 0 )
net.connectNodes(node2, node3, 0, 0 )
# stop node execution
net.stop()
assert net.execStatus == 'stop',"Expected 'stop', got '%s'"%net.execStatus
# pause node execution
net.pause()
assert net.execStatus == 'pause',\
"Expected 'pause', got '%s'"%net.execStatus
# resume node execution
net.resume()
assert net.execStatus == 'running',\
"Expected 'running', got '%s'"%net.execStatus
# start node execution
net.start()
assert net.execStatus == 'pending',\
"Expected 'pending', got '%s'"%net.execStatus
# show the widget in node1
node1.toggleNodeExpand_cb()
node1.inputPorts[0].widget.set('range(20)')
# run network that fails
net.run()
# run network that should run
node1.inputPorts[0].widget.set('range(20)')
# run long execution
node1.inputPorts[0].widget.set('range(400)', run=0)
net.run()
pause(0.1)
net.pause()
pause(0.1)
net.resume()
pause(0.1)
net.stop()
ed.deleteNetwork(net)
def test_08_macroSchedulingNoInputValues():
"""Create a macro in which the inputPort node has no children (i,e,
just a Dial connected to output. Run parent network and verify dial
value is output
"""
from NetworkEditor.macros import MacroNode, MacroInputNode, \
MacroOutputNode, MacroNetwork
net = Network('macroSchedulingNoInputValues')
net.runOnNewData.value = True
ed.addNetwork(net)
# add macro node
node0 = MacroNode(name='test macro')
net.addNode(node0, 200, 200)
macronet = node0.macroNetwork
macronet.runOnNewData.value = True
opnode = macronet.nodes[1]
# dial node connected with output
node1 = DialNode()
macronet.addNode(node1, 150, 150)
macronet.connectNodes(node1, opnode, 'value', 'new')
node0.shrink()
# check output port of macro for nothing yet
assert net.nodes[0].outputPorts[0].data == 'no data yet'
# run Network
net.run()
# check output port of macro for 0.0
assert net.nodes[0].outputPorts[0].data==0.0
# set widget to 1.0 and check macro output port for value 1
node1.inputPorts[0].widget.set(1.0)
assert net.nodes[0].outputPorts[0].data==1.0
|
