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
|
# This file is part of Buildbot. Buildbot is free software: you can
# redistribute it and/or modify it under the terms of the GNU General Public
# License as published by the Free Software Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright Buildbot Team Members
from twisted.trial import unittest
from twisted.internet import defer
from twisted.python import log
from buildbot.util import eventual
class Eventually(unittest.TestCase):
def setUp(self):
# reset the queue to its base state
eventual._theSimpleQueue = eventual._SimpleCallQueue()
self.old_log_err = log.err
self.results = []
def tearDown(self):
log.err = self.old_log_err
return eventual.flushEventualQueue()
# utility callback
def cb(self, *args, **kwargs):
r = args
if kwargs: r = r + (kwargs,)
self.results.append(r)
# flush the queue and assert results
def assertResults(self, exp):
d = eventual.flushEventualQueue()
def cb(_):
self.assertEqual(self.results, exp)
d.addCallback(cb)
return d
## tests
def test_eventually_calls(self):
eventual.eventually(self.cb)
return self.assertResults([()])
def test_eventually_args(self):
eventual.eventually(self.cb, 1, 2, a='a')
return self.assertResults([(1, 2, dict(a='a'))])
def test_eventually_err(self):
# monkey-patch log.err; this is restored by tearDown
log.err = lambda : self.results.append("err")
def cb_fails():
raise RuntimeError("should not cause test failure")
eventual.eventually(cb_fails)
return self.assertResults(['err'])
def test_eventually_butNotNow(self):
eventual.eventually(self.cb, 1)
self.failIf(self.results != [])
return self.assertResults([(1,)])
def test_eventually_order(self):
eventual.eventually(self.cb, 1)
eventual.eventually(self.cb, 2)
eventual.eventually(self.cb, 3)
return self.assertResults([(1,), (2,), (3,)])
def test_flush_waitForChainedEventuallies(self):
def chain(n):
self.results.append(n)
if n <= 0: return
eventual.eventually(chain, n-1)
chain(3)
# (the flush this tests is implicit in assertResults)
return self.assertResults([3, 2, 1, 0])
def test_flush_waitForTreeEventuallies(self):
# a more complex set of eventualities
def tree(n):
self.results.append(n)
if n <= 0: return
eventual.eventually(tree, n-1)
eventual.eventually(tree, n-1)
tree(2)
# (the flush this tests is implicit in assertResults)
return self.assertResults([2, 1, 1, 0, 0, 0, 0])
def test_flush_duringTurn(self):
testd = defer.Deferred()
def cb():
d = eventual.flushEventualQueue()
d.addCallback(testd.callback)
eventual.eventually(cb)
return testd
def test_fireEventually_call(self):
d = eventual.fireEventually(13)
d.addCallback(self.cb)
return self.assertResults([(13,)])
|
