buildbot/test/test_control.py

   1 # -*- test-case-name: buildbot.test.test_control -*-
   2
   3 import os
   4
   5 from twisted.trial import unittest
   6 from twisted.internet import defer
   7
   8 from buildbot import master, interfaces
   9 from buildbot.sourcestamp import SourceStamp
  10 from buildbot.twcompat import providedBy
  11 from buildbot.slave import bot
  12 from buildbot.status.builder import SUCCESS
  13 from buildbot.process import base
  14 from buildbot.test.runutils import rmtree
  15
  16 config = """
  17 from buildbot.process import factory
  18 from buildbot.steps import dummy
  19
  20 def s(klass, **kwargs):
  21     return (klass, kwargs)
  22
  23 f1 = factory.BuildFactory([
  24     s(dummy.Dummy, timeout=1),
  25     ])
  26 c = {}
  27 c['bots'] = [['bot1', 'sekrit']]
  28 c['sources'] = []
  29 c['schedulers'] = []
  30 c['builders'] = [{'name': 'force', 'slavename': 'bot1',
  31                   'builddir': 'force-dir', 'factory': f1}]
  32 c['slavePortnum'] = 0
  33 BuildmasterConfig = c
  34 """
  35
  36 class FakeBuilder:
  37     name = "fake"
  38     def getSlaveCommandVersion(self, command, oldversion=None):
  39         return "1.10"
  40
  41
  42 class Force(unittest.TestCase):
  43
  44     def rmtree(self, d):
  45         rmtree(d)
  46
  47     def setUp(self):
  48         self.master = None
  49         self.slave = None
  50         self.rmtree("control_basedir")
  51         os.mkdir("control_basedir")
  52         self.master = master.BuildMaster("control_basedir")
  53         self.slavebase = os.path.abspath("control_slavebase")
  54         self.rmtree(self.slavebase)
  55         os.mkdir("control_slavebase")
  56
  57     def connectSlave(self):
  58         port = self.master.slavePort._port.getHost().port
  59         slave = bot.BuildSlave("localhost", port, "bot1", "sekrit",
  60                                self.slavebase, keepalive=0, usePTY=1)
  61         self.slave = slave
  62         slave.startService()
  63         d = self.master.botmaster.waitUntilBuilderAttached("force")
  64         return d
  65
  66     def tearDown(self):
  67         dl = []
  68         if self.slave:
  69             dl.append(self.master.botmaster.waitUntilBuilderDetached("force"))
  70             dl.append(defer.maybeDeferred(self.slave.stopService))
  71         if self.master:
  72             dl.append(defer.maybeDeferred(self.master.stopService))
  73         return defer.DeferredList(dl)
  74
  75     def testRequest(self):
  76         m = self.master
  77         m.loadConfig(config)
  78         m.startService()
  79         d = self.connectSlave()
  80         d.addCallback(self._testRequest_1)
  81         return d
  82     def _testRequest_1(self, res):
  83         c = interfaces.IControl(self.master)
  84         req = base.BuildRequest("I was bored", SourceStamp())
  85         builder_control = c.getBuilder("force")
  86         d = defer.Deferred()
  87         req.subscribe(d.callback)
  88         builder_control.requestBuild(req)
  89         d.addCallback(self._testRequest_2)
  90         # we use the same check-the-results code as testForce
  91         return d
  92
  93     def _testRequest_2(self, build_control):
  94         self.failUnless(providedBy(build_control, interfaces.IBuildControl))
  95         d = build_control.getStatus().waitUntilFinished()
  96         d.addCallback(self._testRequest_3)
  97         return d
  98
  99     def _testRequest_3(self, bs):
 100         self.failUnless(providedBy(bs, interfaces.IBuildStatus))
 101         self.failUnless(bs.isFinished())
 102         self.failUnlessEqual(bs.getResults(), SUCCESS)
 103         #self.failUnlessEqual(bs.getResponsibleUsers(), ["bob"]) # TODO
 104         self.failUnlessEqual(bs.getChanges(), [])
 105         #self.failUnlessEqual(bs.getReason(), "forced") # TODO