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