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
|
import time
from twisted.internet import defer
from twisted.trial import unittest
from buildbot.changes.changes import Change
from buildbot.changes.p4poller import P4Source, get_simple_split
first_p4changes = \
"""Change 1 on 2006/04/13 by slamb@testclient 'first rev'
"""
second_p4changes = \
"""Change 3 on 2006/04/13 by bob@testclient 'short desc truncated'
Change 2 on 2006/04/13 by slamb@testclient 'bar'
"""
third_p4changes = \
"""Change 5 on 2006/04/13 by mpatel@testclient 'first rev'
"""
change_4_log = \
"""Change 4 by mpatel@testclient on 2006/04/13 21:55:39
short desc truncated because this is a long description.
"""
change_3_log = \
"""Change 3 by bob@testclient on 2006/04/13 21:51:39
short desc truncated because this is a long description.
"""
change_2_log = \
"""Change 2 by slamb@testclient on 2006/04/13 21:46:23
creation
"""
p4change = {
3: change_3_log +
"""Affected files ...
... //depot/myproject/branch_b/branch_b_file#1 add
... //depot/myproject/branch_b/whatbranch#1 branch
... //depot/myproject/branch_c/whatbranch#1 branch
""",
2: change_2_log +
"""Affected files ...
... //depot/myproject/trunk/whatbranch#1 add
... //depot/otherproject/trunk/something#1 add
""",
5: change_4_log +
"""Affected files ...
... //depot/myproject/branch_b/branch_b_file#1 add
... //depot/myproject/branch_b#75 edit
... //depot/myproject/branch_c/branch_c_file#1 add
""",
}
class MockP4Source(P4Source):
"""Test P4Source which doesn't actually invoke p4."""
invocation = 0
def __init__(self, p4changes, p4change, *args, **kwargs):
P4Source.__init__(self, *args, **kwargs)
self.p4changes = p4changes
self.p4change = p4change
def _get_changes(self):
assert self.working
result = self.p4changes[self.invocation]
self.invocation += 1
return defer.succeed(result)
def _get_describe(self, dummy, num):
assert self.working
return defer.succeed(self.p4change[num])
class TestP4Poller(unittest.TestCase):
def setUp(self):
self.changes = []
self.addChange = self.changes.append
def failUnlessIn(self, substr, string):
# this is for compatibility with python2.2
if isinstance(string, str):
self.failUnless(string.find(substr) != -1)
else:
self.assertIn(substr, string)
def testCheck(self):
"""successful checks"""
self.t = MockP4Source(p4changes=[first_p4changes, second_p4changes],
p4change=p4change,
p4port=None, p4user=None,
p4base='//depot/myproject/',
split_file=lambda x: x.split('/', 1))
self.t.parent = self
# The first time, it just learns the change to start at.
self.assert_(self.t.last_change is None)
self.assert_(not self.t.working)
return self.t.checkp4().addCallback(self._testCheck2)
def _testCheck2(self, res):
self.assertEquals(self.changes, [])
self.assertEquals(self.t.last_change, 1)
# Subsequent times, it returns Change objects for new changes.
return self.t.checkp4().addCallback(self._testCheck3)
def _testCheck3(self, res):
self.assertEquals(len(self.changes), 3)
self.assertEquals(self.t.last_change, 3)
self.assert_(not self.t.working)
# They're supposed to go oldest to newest, so this one must be first.
self.assertEquals(self.changes[0].asText(),
Change(who='slamb',
files=['whatbranch'],
comments=change_2_log,
revision='2',
when=self.makeTime("2006/04/13 21:46:23"),
branch='trunk').asText())
# These two can happen in either order, since they're from the same
# Perforce change.
self.failUnlessIn(
Change(who='bob',
files=['branch_b_file',
'whatbranch'],
comments=change_3_log,
revision='3',
when=self.makeTime("2006/04/13 21:51:39"),
branch='branch_b').asText(),
[c.asText() for c in self.changes])
self.failUnlessIn(
Change(who='bob',
files=['whatbranch'],
comments=change_3_log,
revision='3',
when=self.makeTime("2006/04/13 21:51:39"),
branch='branch_c').asText(),
[c.asText() for c in self.changes])
def makeTime(self, timestring):
datefmt = '%Y/%m/%d %H:%M:%S'
when = time.mktime(time.strptime(timestring, datefmt))
return when
def testFailedChanges(self):
"""'p4 changes' failure is properly ignored"""
self.t = MockP4Source(p4changes=['Perforce client error:\n...'],
p4change={},
p4port=None, p4user=None)
self.t.parent = self
d = self.t.checkp4()
d.addCallback(self._testFailedChanges2)
return d
def _testFailedChanges2(self, f):
self.failUnlessEqual(f, None)
self.assert_(not self.t.working)
def testFailedDescribe(self):
"""'p4 describe' failure is properly ignored"""
c = dict(p4change)
c[3] = 'Perforce client error:\n...'
self.t = MockP4Source(p4changes=[first_p4changes, second_p4changes],
p4change=c, p4port=None, p4user=None)
self.t.parent = self
d = self.t.checkp4()
d.addCallback(self._testFailedDescribe2)
return d
def _testFailedDescribe2(self, res):
# first time finds nothing; check again.
return self.t.checkp4().addCallback(self._testFailedDescribe3)
def _testFailedDescribe3(self, f):
self.failUnlessEqual(f, None)
self.assert_(not self.t.working)
self.assertEquals(self.t.last_change, 2)
def testAlreadyWorking(self):
"""don't launch a new poll while old is still going"""
self.t = P4Source()
self.t.working = True
self.assert_(self.t.last_change is None)
d = self.t.checkp4()
d.addCallback(self._testAlreadyWorking2)
def _testAlreadyWorking2(self, res):
self.assert_(self.t.last_change is None)
def testSplitFile(self):
"""Make sure split file works on branch only changes"""
self.t = MockP4Source(p4changes=[third_p4changes],
p4change=p4change,
p4port=None, p4user=None,
p4base='//depot/myproject/',
split_file=get_simple_split)
self.t.parent = self
self.t.last_change = 50
d = self.t.checkp4()
d.addCallback(self._testSplitFile)
def _testSplitFile(self, res):
self.assertEquals(len(self.changes), 2)
self.assertEquals(self.t.last_change, 5)
|
