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
|
# -*- test-case-name: buildbot.test.test_buildstep -*-
# test cases for buildbot.process.buildstep
from twisted.trial import unittest
from buildbot import interfaces
from buildbot.process import buildstep
# have to subclass LogObserver in order to test it, since the default
# implementations of outReceived() and errReceived() do nothing
class MyLogObserver(buildstep.LogObserver):
def __init__(self):
self._out = [] # list of chunks
self._err = []
def outReceived(self, data):
self._out.append(data)
def errReceived(self, data):
self._err.append(data)
class ObserverTestCase(unittest.TestCase):
observer_cls = None # must be set by subclass
def setUp(self):
self.observer = self.observer_cls()
def _logStdout(self, chunk):
# why does LogObserver.logChunk() take 'build', 'step', and
# 'log' arguments when it clearly doesn't use them for anything?
self.observer.logChunk(None, None, None, interfaces.LOG_CHANNEL_STDOUT, chunk)
def _logStderr(self, chunk):
self.observer.logChunk(None, None, None, interfaces.LOG_CHANNEL_STDERR, chunk)
def _assertStdout(self, expect_lines):
self.assertEqual(self.observer._out, expect_lines)
def _assertStderr(self, expect_lines):
self.assertEqual(self.observer._err, expect_lines)
class LogObserver(ObserverTestCase):
observer_cls = MyLogObserver
def testLogChunk(self):
self._logStdout("foo")
self._logStderr("argh")
self._logStdout(" wubba\n")
self._logStderr("!!!\n")
self._assertStdout(["foo", " wubba\n"])
self._assertStderr(["argh", "!!!\n"])
# again, have to subclass LogLineObserver in order to test it, because the
# default implementations of data-receiving methods are empty
class MyLogLineObserver(buildstep.LogLineObserver):
def __init__(self):
#super(MyLogLineObserver, self).__init__()
buildstep.LogLineObserver.__init__(self)
self._out = [] # list of lines
self._err = []
def outLineReceived(self, line):
self._out.append(line)
def errLineReceived(self, line):
self._err.append(line)
class LogLineObserver(ObserverTestCase):
observer_cls = MyLogLineObserver
def testLineBuffered(self):
# no challenge here: we feed it chunks that are already lines
# (like a program writing to stdout in line-buffered mode)
self._logStdout("stdout line 1\n")
self._logStdout("stdout line 2\n")
self._logStderr("stderr line 1\n")
self._logStdout("stdout line 3\n")
self._assertStdout(["stdout line 1",
"stdout line 2",
"stdout line 3"])
self._assertStderr(["stderr line 1"])
def testShortBrokenLines(self):
self._logStdout("stdout line 1 starts ")
self._logStderr("an intervening line of error\n")
self._logStdout("and continues ")
self._logStdout("but finishes here\n")
self._logStderr("more error\n")
self._logStdout("and another line of stdout\n")
self._assertStdout(["stdout line 1 starts and continues but finishes here",
"and another line of stdout"])
self._assertStderr(["an intervening line of error",
"more error"])
def testLongLine(self):
chunk = "." * 1024
self._logStdout(chunk)
self._logStdout(chunk)
self._logStdout(chunk)
self._logStdout(chunk)
self._logStdout(chunk)
self._logStdout("\n")
self._assertStdout([chunk * 5])
self._assertStderr([])
def testBigChunk(self):
chunk = "." * 5000
self._logStdout(chunk)
self._logStdout("\n")
self._assertStdout([chunk])
self._assertStderr([])
def testReallyLongLine(self):
# A single line of > 16384 bytes is dropped on the floor (bug #201).
# In real life, I observed such a line being broken into chunks of
# 4095 bytes, so that's how I'm breaking it here.
self.observer.setMaxLineLength(65536)
chunk = "." * 4095
self._logStdout(chunk)
self._logStdout(chunk)
self._logStdout(chunk)
self._logStdout(chunk) # now we're up to 16380 bytes
self._logStdout("12345\n")
self._assertStdout([chunk*4 + "12345"])
self._assertStderr([])
class RemoteShellTest(unittest.TestCase):
def testRepr(self):
# Test for #352
rsc = buildstep.RemoteShellCommand('.', ('sh', 'make'))
testval = repr(rsc)
rsc = buildstep.RemoteShellCommand('.', ['sh', 'make'])
testval = repr(rsc)
rsc = buildstep.RemoteShellCommand('.', 'make')
testval = repr(rsc)
|
