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
|
# -*- test-case-name: buildbot.test.test_control -*-
import sys, os, signal, shutil, time, errno
from twisted.trial import unittest
from twisted.internet import defer, reactor
from buildbot import master, interfaces
from buildbot.sourcestamp import SourceStamp
from buildbot.twcompat import providedBy, maybeWait
from buildbot.slave import bot
from buildbot.status import builder
from buildbot.status.builder import SUCCESS
from buildbot.process import base
config = """
from buildbot.process import factory, step
def s(klass, **kwargs):
return (klass, kwargs)
f1 = factory.BuildFactory([
s(step.Dummy, timeout=1),
])
c = {}
c['bots'] = [['bot1', 'sekrit']]
c['sources'] = []
c['schedulers'] = []
c['builders'] = [{'name': 'force', 'slavename': 'bot1',
'builddir': 'force-dir', 'factory': f1}]
c['slavePortnum'] = 0
BuildmasterConfig = c
"""
class FakeBuilder:
name = "fake"
def getSlaveCommandVersion(self, command, oldversion=None):
return "1.10"
class SignalMixin:
sigchldHandler = None
def setUpClass(self):
# make sure SIGCHLD handler is installed, as it should be on
# reactor.run(). problem is reactor may not have been run when this
# test runs.
if hasattr(reactor, "_handleSigchld") and hasattr(signal, "SIGCHLD"):
self.sigchldHandler = signal.signal(signal.SIGCHLD,
reactor._handleSigchld)
def tearDownClass(self):
if self.sigchldHandler:
signal.signal(signal.SIGCHLD, self.sigchldHandler)
class Force(unittest.TestCase):
def rmtree(self, d):
try:
shutil.rmtree(d, ignore_errors=1)
except OSError, e:
# stupid 2.2 appears to ignore ignore_errors
if e.errno != errno.ENOENT:
raise
def setUp(self):
self.master = None
self.slave = None
self.rmtree("control_basedir")
os.mkdir("control_basedir")
self.master = master.BuildMaster("control_basedir")
self.slavebase = os.path.abspath("control_slavebase")
self.rmtree(self.slavebase)
os.mkdir("control_slavebase")
def connectSlave(self):
port = self.master.slavePort._port.getHost().port
slave = bot.BuildSlave("localhost", port, "bot1", "sekrit",
self.slavebase, keepalive=0, usePTY=1)
self.slave = slave
slave.startService()
d = self.master.botmaster.waitUntilBuilderAttached("force")
return d
def tearDown(self):
dl = []
if self.slave:
dl.append(self.master.botmaster.waitUntilBuilderDetached("force"))
dl.append(defer.maybeDeferred(self.slave.stopService))
if self.master:
dl.append(defer.maybeDeferred(self.master.stopService))
return maybeWait(defer.DeferredList(dl))
def testForce(self):
# TODO: since BuilderControl.forceBuild has been deprecated, this
# test is scheduled to be removed soon
m = self.master
m.loadConfig(config)
m.startService()
d = self.connectSlave()
d.addCallback(self._testForce_1)
return maybeWait(d)
def _testForce_1(self, res):
c = interfaces.IControl(self.master)
builder_control = c.getBuilder("force")
d = builder_control.forceBuild("bob", "I was bored")
d.addCallback(self._testForce_2)
return d
def _testForce_2(self, build_control):
self.failUnless(providedBy(build_control, interfaces.IBuildControl))
d = build_control.getStatus().waitUntilFinished()
d.addCallback(self._testForce_3)
return d
def _testForce_3(self, bs):
self.failUnless(providedBy(bs, interfaces.IBuildStatus))
self.failUnless(bs.isFinished())
self.failUnlessEqual(bs.getResults(), SUCCESS)
#self.failUnlessEqual(bs.getResponsibleUsers(), ["bob"]) # TODO
self.failUnlessEqual(bs.getChanges(), [])
#self.failUnlessEqual(bs.getReason(), "forced") # TODO
def testRequest(self):
m = self.master
m.loadConfig(config)
m.startService()
d = self.connectSlave()
d.addCallback(self._testRequest_1)
return maybeWait(d)
def _testRequest_1(self, res):
c = interfaces.IControl(self.master)
req = base.BuildRequest("I was bored", SourceStamp())
builder_control = c.getBuilder("force")
d = defer.Deferred()
req.subscribe(d.callback)
builder_control.requestBuild(req)
d.addCallback(self._testForce_2)
# we use the same check-the-results code as testForce
return d
|
