← Back to branch summary

~mapclient.devs/mapclient/stable

« back to all changes in this revision

Viewing changes to tests/core/test_workflowscene.py

  • Committer: musculoskeletal
  • Date: 2014-06-25 05:38:05 UTC
  • mfrom: (1.6.35 testing)
  • Revision ID: musculoskeletal@bioeng1033-20140625053805-jkqhi5oq74vmlntl
Merging testing into stable.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/core/test_workflowscene.py
 
1
'''
 
2
Created on Mar 14, 2013
 
3
 
 
4
@author: hsorby
 
5
'''
 
6
import os, unittest
 
7
 
 
8
from PySide import QtCore
 
9
 
 
10
from mapclient.core.workflowscene import WorkflowScene, WorkflowDependencyGraph, MetaStep, Connection
 
11
 
 
12
from tests import utils
 
13
 
 
14
test_path = os.path.join(os.path.dirname(utils.__file__), 'test_resources', 'core_test')
 
15
 
 
16
class DumbManager(object):
 
17
 
 
18
    def location(self):
 
19
        return self._location
 
20
 
 
21
 
 
22
class DumbStep(object):
 
23
 
 
24
    _ports = ['one', 'two', 'three']
 
25
 
 
26
    def isConfigured(self):
 
27
        return True
 
28
 
 
29
    def getName(self):
 
30
        return self._name
 
31
 
 
32
    def getIdentifier(self):
 
33
        return self._identifier
 
34
 
 
35
    def serialize(self, location):
 
36
        pass
 
37
 
 
38
    def deserialize(self, location):
 
39
        pass
 
40
 
 
41
    def execute(self):
 
42
        pass
 
43
 
 
44
    def setPortData(self, index, data):
 
45
        pass
 
46
 
 
47
    def getPortData(self, index):
 
48
        return None
 
49
 
 
50
 
 
51
class WorkflowSceneTestCase(unittest.TestCase):
 
52
 
 
53
 
 
54
    def assertIn(self, a, b, *args, **kwargs):
 
55
        ''''Python < v2.7 compatibility.  Assert "a" in "b"'''
 
56
        try:
 
57
            f = super(WorkflowSceneTestCase, self).assertIn
 
58
        except AttributeError:
 
59
            self.assertTrue(a in b, *args, **kwargs)
 
60
        else:
 
61
            f(a, b, *args, **kwargs)
 
62
 
 
63
    def assertNotIn(self, a, b, *args, **kwargs):
 
64
        ''''Python < v2.7 compatibility.  Assert "a" NOT in "b"'''
 
65
        try:
 
66
            f = super(WorkflowSceneTestCase, self).assertNotIn
 
67
        except AttributeError:
 
68
            self.assertFalse(a in b, *args, **kwargs)
 
69
        else:
 
70
            f(a, b, *args, **kwargs)
 
71
 
 
72
    def testCreate(self):
 
73
 
 
74
        s = WorkflowScene(DumbManager())
 
75
        self.assertTrue(s != None)
 
76
 
 
77
    def testItemAPI(self):
 
78
 
 
79
        item = MetaStep(DumbStep())
 
80
        s = WorkflowScene(DumbManager())
 
81
        s.addItem(item)
 
82
        self.assertEqual(len(s._items), 1)
 
83
        s.setItemPos(item, QtCore.QPoint(344, 404))
 
84
        self.assertEqual(s._items[item]._pos.x(), 344)
 
85
        self.assertEqual(s._items[item]._pos.y(), 404)
 
86
        s.setItemSelected(item, False)
 
87
        self.assertFalse(s._items[item]._selected)
 
88
        s.removeItem(item)
 
89
        self.assertEqual(len(s._items), 0)
 
90
 
 
91
    def testSaveLoad(self):
 
92
        test_conf = os.path.join(test_path, 'test.conf')
 
93
        dm = DumbManager()
 
94
        dm._location = test_conf
 
95
 
 
96
        s = WorkflowScene(dm)
 
97
        ws = QtCore.QSettings(test_conf, QtCore.QSettings.IniFormat)
 
98
        ds1 = DumbStep()
 
99
        ds1._identifier = '1'
 
100
        ds1._name = 'a'
 
101
        ds2 = DumbStep()
 
102
        ds2._identifier = '2'
 
103
        ds2._name = 'b'
 
104
        item1 = MetaStep(ds1)
 
105
        item2 = MetaStep(ds2)
 
106
        s.addItem(item1)
 
107
        s.addItem(item2)
 
108
        s.saveState(ws)
 
109
        del ws
 
110
 
 
111
        self.assertTrue(os.path.exists(test_conf))
 
112
        file_content = open(test_conf).read()
 
113
        self.assertIn('Point(0 0)', file_content)
 
114
        self.assertIn('name=a', file_content)
 
115
        self.assertIn('name=b', file_content)
 
116
        self.assertNotIn('selected=false', file_content)
 
117
 
 
118
        os.remove(test_conf)
 
119
 
 
120
 
 
121
class WorkflowDependencyGraphTestCase(unittest.TestCase):
 
122
 
 
123
 
 
124
    def assertIn(self, a, b, *args, **kwargs):
 
125
        ''''Python < v2.7 compatibility.  Assert "a" in "b"'''
 
126
        try:
 
127
            f = super(WorkflowDependencyGraphTestCase, self).assertIn
 
128
        except AttributeError:
 
129
            self.assertTrue(a in b, *args, **kwargs)
 
130
        else:
 
131
            f(a, b, *args, **kwargs)
 
132
 
 
133
    def assertNotIn(self, a, b, *args, **kwargs):
 
134
        ''''Python < v2.7 compatibility.  Assert "a" NOT in "b"'''
 
135
        try:
 
136
            f = super(WorkflowDependencyGraphTestCase, self).assertNotIn
 
137
        except AttributeError:
 
138
            self.assertFalse(a in b, *args, **kwargs)
 
139
        else:
 
140
            f(a, b, *args, **kwargs)
 
141
 
 
142
    def assertLess(self, a, b, *args, **kwargs):
 
143
        ''''Python < v2.7 compatibility.  Assert "a" less "b"'''
 
144
        try:
 
145
            f = super(WorkflowDependencyGraphTestCase, self).assertLess
 
146
        except AttributeError:
 
147
            self.assertTrue(a < b, *args, **kwargs)
 
148
        else:
 
149
            f(a, b, *args, **kwargs)
 
150
 
 
151
    def setUp(self):
 
152
        self._s = WorkflowScene(DumbManager())
 
153
        self._nodes = []
 
154
        self._nodes.append(MetaStep(DumbStep()))
 
155
        self._nodes.append(MetaStep(DumbStep()))
 
156
        self._nodes.append(MetaStep(DumbStep()))
 
157
        self._nodes.append(MetaStep(DumbStep()))
 
158
        self._nodes.append(MetaStep(DumbStep()))
 
159
        self._nodes.append(MetaStep(DumbStep()))
 
160
        self._nodes.append(MetaStep(DumbStep()))
 
161
 
 
162
    def tearDown(self):
 
163
        self._s.clear()
 
164
        self._nodes = []
 
165
 
 
166
    def testCreate(self):
 
167
        g = WorkflowDependencyGraph(self._s)
 
168
        self.assertTrue(g != None)
 
169
 
 
170
    def testGraph1(self):
 
171
        g = WorkflowDependencyGraph(self._s)
 
172
        c1 = Connection(self._nodes[0], 0, self._nodes[1], 0)
 
173
        self._s.addItem(self._nodes[0])
 
174
        self._s.addItem(self._nodes[1])
 
175
        self._s.addItem(c1)
 
176
 
 
177
        self.assertTrue(g.canExecute())
 
178
        self.assertEqual(len(g._topologicalOrder), 2)
 
179
        self.assertEqual(g._topologicalOrder[0], self._nodes[0])
 
180
        self.assertEqual(g._topologicalOrder[1], self._nodes[1])
 
181
 
 
182
    def testGraph2(self):
 
183
        g = WorkflowDependencyGraph(self._s)
 
184
        c1 = Connection(self._nodes[0], 0, self._nodes[1], 0)
 
185
        c2 = Connection(self._nodes[1], 0, self._nodes[2], 0)
 
186
        self._s.addItem(self._nodes[0])
 
187
        self._s.addItem(self._nodes[1])
 
188
        self._s.addItem(self._nodes[2])
 
189
        self._s.addItem(c1)
 
190
        self._s.addItem(c2)
 
191
 
 
192
        self.assertTrue(g.canExecute())
 
193
        self.assertEqual(len(g._topologicalOrder), 3)
 
194
        self.assertEqual(g._topologicalOrder[0], self._nodes[0])
 
195
        self.assertEqual(g._topologicalOrder[1], self._nodes[1])
 
196
        self.assertEqual(g._topologicalOrder[2], self._nodes[2])
 
197
 
 
198
    def testGraph3(self):
 
199
        g = WorkflowDependencyGraph(self._s)
 
200
        c1 = Connection(self._nodes[0], 0, self._nodes[1], 0)
 
201
        c2 = Connection(self._nodes[1], 0, self._nodes[2], 0)
 
202
        c3 = Connection(self._nodes[2], 0, self._nodes[3], 0)
 
203
        self._s.addItem(self._nodes[0])
 
204
        self._s.addItem(self._nodes[1])
 
205
        self._s.addItem(self._nodes[2])
 
206
        self._s.addItem(self._nodes[3])
 
207
        self._s.addItem(c3)
 
208
        self._s.addItem(c1)
 
209
        self._s.addItem(c2)
 
210
 
 
211
        self.assertTrue(g.canExecute())
 
212
        self.assertEqual(len(g._topologicalOrder), 4)
 
213
        self.assertEqual(g._topologicalOrder[0], self._nodes[0])
 
214
        self.assertEqual(g._topologicalOrder[1], self._nodes[1])
 
215
        self.assertEqual(g._topologicalOrder[2], self._nodes[2])
 
216
        self.assertEqual(g._topologicalOrder[3], self._nodes[3])
 
217
 
 
218
    def testGraph4(self):
 
219
        g = WorkflowDependencyGraph(self._s)
 
220
        c1 = Connection(self._nodes[0], 0, self._nodes[1], 0)
 
221
        c2 = Connection(self._nodes[1], 0, self._nodes[2], 0)
 
222
        c3 = Connection(self._nodes[2], 0, self._nodes[3], 0)
 
223
        self._s.addItem(self._nodes[0])
 
224
        self._s.addItem(self._nodes[1])
 
225
        self._s.addItem(self._nodes[2])
 
226
        self._s.addItem(self._nodes[3])
 
227
        self._s.addItem(self._nodes[4])
 
228
        self._s.addItem(self._nodes[5])
 
229
        self._s.addItem(c3)
 
230
        self._s.addItem(c1)
 
231
        self._s.addItem(c2)
 
232
 
 
233
        nodes = g._findAllConnectedNodes()
 
234
        self.assertEqual(4, len(nodes))
 
235
        self.assertIn(self._nodes[0], nodes)
 
236
        self.assertIn(self._nodes[1], nodes)
 
237
        self.assertIn(self._nodes[2], nodes)
 
238
        self.assertIn(self._nodes[3], nodes)
 
239
        self.assertNotIn(self._nodes[4], nodes)
 
240
        self.assertNotIn(self._nodes[5], nodes)
 
241
 
 
242
    def testGraph5(self):
 
243
        g = WorkflowDependencyGraph(self._s)
 
244
        c1 = Connection(self._nodes[0], 0, self._nodes[2], 0)
 
245
        c2 = Connection(self._nodes[1], 0, self._nodes[2], 0)
 
246
        c3 = Connection(self._nodes[2], 0, self._nodes[3], 0)
 
247
        c4 = Connection(self._nodes[2], 0, self._nodes[4], 0)
 
248
        self._s.addItem(self._nodes[0])
 
249
        self._s.addItem(self._nodes[1])
 
250
        self._s.addItem(self._nodes[2])
 
251
        self._s.addItem(self._nodes[3])
 
252
        self._s.addItem(self._nodes[4])
 
253
        self._s.addItem(self._nodes[5])
 
254
        self._s.addItem(c1)
 
255
        self._s.addItem(c2)
 
256
        self._s.addItem(c3)
 
257
        self._s.addItem(c4)
 
258
 
 
259
        nodes = g._findAllConnectedNodes()
 
260
        self.assertEqual(5, len(nodes))
 
261
        self.assertIn(self._nodes[0], nodes)
 
262
        self.assertIn(self._nodes[1], nodes)
 
263
        self.assertIn(self._nodes[2], nodes)
 
264
        self.assertIn(self._nodes[3], nodes)
 
265
        self.assertIn(self._nodes[4], nodes)
 
266
        self.assertNotIn(self._nodes[5], nodes)
 
267
 
 
268
        graph = g._calculateDependencyGraph()
 
269
        self.assertFalse(g._nodeIsDestination(graph, self._nodes[0]))
 
270
        self.assertFalse(g._nodeIsDestination(graph, self._nodes[1]))
 
271
        self.assertTrue(g._nodeIsDestination(graph, self._nodes[2]))
 
272
        self.assertTrue(g._nodeIsDestination(graph, self._nodes[3]))
 
273
        self.assertTrue(g._nodeIsDestination(graph, self._nodes[4]))
 
274
        self.assertFalse(g._nodeIsDestination(graph, self._nodes[5]))
 
275
 
 
276
        starting_set = g._findStartingSet(graph, nodes)
 
277
 
 
278
        self.assertEqual(2, len(starting_set))
 
279
        self.assertIn(self._nodes[0], starting_set)
 
280
        self.assertIn(self._nodes[1], starting_set)
 
281
        self.assertNotIn(self._nodes[2], starting_set)
 
282
        self.assertNotIn(self._nodes[3], starting_set)
 
283
        self.assertNotIn(self._nodes[4], starting_set)
 
284
        self.assertNotIn(self._nodes[5], starting_set)
 
285
 
 
286
        order = g._determineTopologicalOrder(graph, starting_set)
 
287
        self.assertEqual(5, len(order))
 
288
        index0 = order.index(self._nodes[0])
 
289
        index1 = order.index(self._nodes[1])
 
290
        index2 = order.index(self._nodes[2])
 
291
        index3 = order.index(self._nodes[3])
 
292
        index4 = order.index(self._nodes[4])
 
293
        self.assertLess(index1, index2)
 
294
        self.assertLess(index0, index2)
 
295
        self.assertLess(index2, index3)
 
296
        self.assertLess(index2, index4)
 
297
 
 
298
    def testGraph6(self):
 
299
        g = WorkflowDependencyGraph(self._s)
 
300
        c1 = Connection(self._nodes[0], 0, self._nodes[2], 0)
 
301
        c2 = Connection(self._nodes[1], 0, self._nodes[2], 0)
 
302
        c3 = Connection(self._nodes[2], 0, self._nodes[3], 0)
 
303
        c4 = Connection(self._nodes[2], 0, self._nodes[4], 0)
 
304
        c5 = Connection(self._nodes[6], 0, self._nodes[0], 0)
 
305
        c6 = Connection(self._nodes[6], 0, self._nodes[1], 0)
 
306
        self._s.addItem(self._nodes[0])
 
307
        self._s.addItem(self._nodes[1])
 
308
        self._s.addItem(self._nodes[2])
 
309
        self._s.addItem(self._nodes[3])
 
310
        self._s.addItem(self._nodes[4])
 
311
        self._s.addItem(self._nodes[5])
 
312
        self._s.addItem(self._nodes[6])
 
313
        self._s.addItem(c1)
 
314
        self._s.addItem(c2)
 
315
        self._s.addItem(c3)
 
316
        self._s.addItem(c4)
 
317
        self._s.addItem(c5)
 
318
        self._s.addItem(c6)
 
319
 
 
320
        nodes = g._findAllConnectedNodes()
 
321
        self.assertEqual(6, len(nodes))
 
322
        graph = g._calculateDependencyGraph()
 
323
        starting_set = g._findStartingSet(graph, nodes)
 
324
        self.assertEqual(1, len(starting_set))
 
325
        order = g._determineTopologicalOrder(graph, starting_set)
 
326
        self.assertEqual(6, len(order))
 
327
        index0 = order.index(self._nodes[6])
 
328
        index1 = order.index(self._nodes[1])
 
329
        self.assertLess(index0, index1)
 
330
 
 
331
    def testGraph7(self):
 
332
        '''
 
333
        Testing independent graphs
 
334
        '''
 
335
        g = WorkflowDependencyGraph(self._s)
 
336
        c1 = Connection(self._nodes[0], 0, self._nodes[2], 0)
 
337
        c2 = Connection(self._nodes[1], 0, self._nodes[2], 0)
 
338
        c3 = Connection(self._nodes[2], 0, self._nodes[3], 0)
 
339
        c4 = Connection(self._nodes[2], 0, self._nodes[4], 0)
 
340
        c5 = Connection(self._nodes[6], 0, self._nodes[5], 0)
 
341
        self._s.addItem(self._nodes[0])
 
342
        self._s.addItem(self._nodes[1])
 
343
        self._s.addItem(self._nodes[2])
 
344
        self._s.addItem(self._nodes[3])
 
345
        self._s.addItem(self._nodes[4])
 
346
        self._s.addItem(self._nodes[5])
 
347
        self._s.addItem(self._nodes[6])
 
348
        self._s.addItem(c1)
 
349
        self._s.addItem(c2)
 
350
        self._s.addItem(c3)
 
351
        self._s.addItem(c4)
 
352
        self._s.addItem(c5)
 
353
 
 
354
        nodes = g._findAllConnectedNodes()
 
355
        self.assertEqual(7, len(nodes))
 
356
        graph = g._calculateDependencyGraph()
 
357
        starting_set = g._findStartingSet(graph, nodes)
 
358
        self.assertEqual(3, len(starting_set))
 
359
        order = g._determineTopologicalOrder(graph, starting_set)
 
360
        self.assertEqual(7, len(order))
 
361
 
 
362
    def testGraph8(self):
 
363
        '''
 
364
        Testing graph with loop
 
365
        '''
 
366
        g = WorkflowDependencyGraph(self._s)
 
367
        c1 = Connection(self._nodes[0], 0, self._nodes[1], 0)
 
368
        c2 = Connection(self._nodes[1], 0, self._nodes[2], 0)
 
369
        c3 = Connection(self._nodes[2], 0, self._nodes[3], 0)
 
370
        c4 = Connection(self._nodes[3], 0, self._nodes[4], 0)
 
371
        c5 = Connection(self._nodes[3], 0, self._nodes[5], 0)
 
372
        c6 = Connection(self._nodes[5], 0, self._nodes[6], 0)
 
373
        c7 = Connection(self._nodes[6], 0, self._nodes[2], 0)
 
374
        self._s.addItem(self._nodes[0])
 
375
        self._s.addItem(self._nodes[1])
 
376
        self._s.addItem(self._nodes[2])
 
377
        self._s.addItem(self._nodes[3])
 
378
        self._s.addItem(self._nodes[4])
 
379
        self._s.addItem(self._nodes[5])
 
380
        self._s.addItem(self._nodes[6])
 
381
        self._s.addItem(c1)
 
382
        self._s.addItem(c2)
 
383
        self._s.addItem(c3)
 
384
        self._s.addItem(c4)
 
385
        self._s.addItem(c5)
 
386
        self._s.addItem(c6)
 
387
        self._s.addItem(c7)
 
388
 
 
389
        nodes = g._findAllConnectedNodes()
 
390
        self.assertEqual(7, len(nodes))
 
391
        graph = g._calculateDependencyGraph()
 
392
        starting_set = g._findStartingSet(graph, nodes)
 
393
        self.assertEqual(1, len(starting_set))
 
394
        order = g._determineTopologicalOrder(graph, starting_set)
 
395
        self.assertEqual(0, len(order))
 
396
 
 
397
    def testExecute(self):
 
398
        g = WorkflowDependencyGraph(self._s)
 
399
        c1 = Connection(self._nodes[0], 0, self._nodes[2], 0)
 
400
        c2 = Connection(self._nodes[1], 0, self._nodes[2], 1)
 
401
        c3 = Connection(self._nodes[1], 1, self._nodes[2], 2)
 
402
        c4 = Connection(self._nodes[2], 0, self._nodes[3], 0)
 
403
        self._s.addItem(self._nodes[0])
 
404
        self._s.addItem(self._nodes[1])
 
405
        self._s.addItem(self._nodes[2])
 
406
        self._s.addItem(self._nodes[3])
 
407
        self._s.addItem(c1)
 
408
        self._s.addItem(c2)
 
409
        self._s.addItem(c3)
 
410
        self._s.addItem(c4)
 
411
        g.canExecute()
 
412
 
 
413
        for _ in range(len(g._topologicalOrder)):
 
414
            g.execute()
 
415
 
 
416
class DictUtilsTestCase(unittest.TestCase):
 
417
 
 
418
 
 
419
    def assertIn(self, a, b, *args, **kwargs):
 
420
        ''''Python < v2.7 compatibility.  Assert "a" in "b"'''
 
421
        try:
 
422
            f = super(DictUtilsTestCase, self).assertIn
 
423
        except AttributeError:
 
424
            self.assertTrue(a in b, *args, **kwargs)
 
425
        else:
 
426
            f(a, b, *args, **kwargs)
 
427
 
 
428
    def testReverseDict(self):
 
429
        d = {'a': ['1', '2'], 'b': ['2', '3']}
 
430
#         from collections import defaultdict
 
431
        rd = {}  # defaultdict(list)
 
432
        for k, v in d.items():
 
433
            for rk in v:
 
434
                rd[rk] = rd.get(rk, [])
 
435
                rd[rk].append(k)
 
436
 
 
437
        self.assertIn('1', rd)
 
438
        self.assertIn('2', rd)
 
439
        self.assertIn('3', rd)
 
440
        self.assertEqual(['a'], rd['1'])
 
441
        self.assertIn('a', rd['2'])
 
442
        self.assertIn('b', rd['2'])
 
443
 
 
444
if __name__ == "__main__":
 
445
    # import sys;sys.argv = ['', 'Test.testCreate']
 
446
    unittest.main()