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
|
##
# Copyright (c) 2012-2016 Apple Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##
"""
Tests for L{twext.python.parallel}.
"""
from twisted.internet.defer import Deferred
from twext.python.parallel import Parallelizer
from twisted.trial.unittest import TestCase
class ParallelizerTests(TestCase):
"""
Tests for L{Parallelizer}.
"""
def test_doAndDone(self):
"""
Blanket catch-all test. (TODO: split this up into more nice
fine-grained tests.)
"""
d1 = Deferred()
d2 = Deferred()
d3 = Deferred()
d4 = Deferred()
doing = []
done = []
allDone = []
p = Parallelizer(['a', 'b', 'c'])
p.do(lambda a: doing.append(a) or d1).addCallback(done.append)
p.do(lambda b: doing.append(b) or d2).addCallback(done.append)
p.do(lambda c: doing.append(c) or d3).addCallback(done.append)
p.do(lambda b1: doing.append(b1) or d4).addCallback(done.append)
p.done().addCallback(allDone.append)
self.assertEqual(allDone, [])
self.assertEqual(doing, ['a', 'b', 'c'])
self.assertEqual(done, [None, None, None])
d2.callback(1)
self.assertEqual(doing, ['a', 'b', 'c', 'b'])
self.assertEqual(done, [None, None, None, None])
self.assertEqual(allDone, [])
d3.callback(2)
d4.callback(3)
d1.callback(4)
self.assertEqual(done, [None, None, None, None])
self.assertEqual(allDone, [None])
|
