[Buildbot-commits] buildbot/buildbot/test test_transfer.py, NONE, 1.1 runutils.py, 1.13, 1.14
Brian Warner
warner at users.sourceforge.net
Fri Sep 15 14:49:35 UTC 2006
Update of /cvsroot/buildbot/buildbot/buildbot/test
In directory sc8-pr-cvs3.sourceforge.net:/tmp/cvs-serv21663/buildbot/test
Modified Files:
runutils.py
Added Files:
test_transfer.py
Log Message:
[project @ filetransfer: add a unit test, docs, do some cleanup]
Original author: warner at lothar.com
Date: 2006-09-09 06:11:59
Index: runutils.py
===================================================================
RCS file: /cvsroot/buildbot/buildbot/buildbot/test/runutils.py,v
retrieving revision 1.13
retrieving revision 1.14
diff -u -d -r1.13 -r1.14
--- runutils.py 21 Aug 2006 00:43:20 -0000 1.13
+++ runutils.py 15 Sep 2006 14:49:33 -0000 1.14
@@ -319,3 +319,94 @@
if "stderr" in u:
print u["stderr"]
+# ----------------------------------------
+from twisted.spread.util import LocalAsRemote
+
+class LocalWrapper:
+ # r = pb.Referenceable()
+ # w = LocalWrapper(r)
+ # now you can do things like w.callRemote()
+ def __init__(self, target):
+ self.target = target
+
+ def callRemote(self, name, *args, **kwargs):
+ d = defer.maybeDeferred(self._callRemote, name, *args, **kwargs)
+ return d
+
+ def _callRemote(self, name, *args, **kwargs):
+ method = getattr(self.target, "remote_"+name)
+ return method(*args, **kwargs)
+
+ def notifyOnDisconnect(self, observer):
+ pass
+ def dontNotifyOnDisconnect(self, observer):
+ pass
+
+
+class LocalSlaveBuilder(bot.SlaveBuilder):
+ """I am object that behaves like a pb.RemoteReference, but in fact I
+ invoke methods locally."""
+ _arg_filter = None
+
+ def setArgFilter(self, filter):
+ self._arg_filter = filter
+
+ def remote_startCommand(self, stepref, stepId, command, args):
+ if self._arg_filter:
+ args = self._arg_filter(args)
+ # stepref should be a RemoteReference to the RemoteCommand
+ return bot.SlaveBuilder.remote_startCommand(self,
+ LocalWrapper(stepref),
+ stepId, command, args)
+
+class StepTester:
+ """Utility class to exercise BuildSteps and RemoteCommands, without
+ really using a Build or a Bot. No networks are used.
+
+ Use this as follows::
+
+ class MyTest(StepTester, unittest.TestCase):
+ def testOne(self):
+ self.slavebase = 'testOne.slave'
+ self.masterbase = 'testOne.master'
+ sb = self.makeSlaveBuilder()
+ step = self.makeStep(stepclass, **kwargs)
+ d = self.runStep(step)
+ d.addCallback(_checkResults)
+ return d
+ """
+
+ #slavebase = "slavebase"
+ slavebuilderbase = "slavebuilderbase"
+ #masterbase = "masterbase"
+
+ def makeSlaveBuilder(self):
+ os.mkdir(self.slavebase)
+ os.mkdir(os.path.join(self.slavebase, self.slavebuilderbase))
+ b = bot.Bot(self.slavebase, False)
+ b.startService()
+ sb = LocalSlaveBuilder("slavebuildername", False)
+ sb.setArgFilter(self.filterArgs)
+ sb.usePTY = False
+ sb.setServiceParent(b)
+ sb.setBuilddir(self.slavebuilderbase)
+ self.remote = LocalWrapper(sb)
+ return sb
+
+ workdir = "build"
+ def makeStep(self, factory, **kwargs):
+ if not kwargs.has_key("workdir"):
+ kwargs['workdir'] = self.workdir
+ step = makeBuildStep(self.masterbase, factory, **kwargs)
+ return step
+
+ def runStep(self, step):
+ d = defer.maybeDeferred(step.startStep, self.remote)
+ return d
+
+ def wrap(self, target):
+ return LocalWrapper(target)
+
+ def filterArgs(self, args):
+ # this can be overridden
+ return args
--- NEW FILE: test_transfer.py ---
# -*- test-case-name: buildbot.test.test_transfer -*-
import os
from twisted.trial import unittest
from twisted.internet import defer
from buildbot.twcompat import maybeWait
from buildbot.steps.transfer import FileUpload, FileDownload
from buildbot.test.runutils import StepTester
from buildbot.status.builder import SUCCESS, FAILURE
# these steps pass a pb.Referenceable inside their arguments, so we have to
# catch and wrap them. If the LocalAsRemote wrapper were a proper membrane,
# we wouldn't have to do this.
class Upload(StepTester, unittest.TestCase):
def filterArgs(self, args):
if "writer" in args:
args["writer"] = self.wrap(args["writer"])
return args
def testSuccess(self):
self.slavebase = "testUpload.slave"
self.masterbase = "testUpload.master"
sb = self.makeSlaveBuilder()
os.mkdir(os.path.join(self.slavebase, self.slavebuilderbase,
"build"))
# the buildmaster normally runs chdir'ed into masterbase, so uploaded
# files will appear there. Under trial, we're chdir'ed into
# _trial_temp instead, so use a different masterdest= to keep the
# uploaded file in a test-local directory
masterdest = os.path.join(self.masterbase, "dest.text")
step = self.makeStep(FileUpload,
slavesrc="source.txt",
masterdest=masterdest)
slavesrc = os.path.join(self.slavebase,
self.slavebuilderbase,
"build",
"source.txt")
contents = "this is the source file\n"
open(slavesrc, "w").write(contents)
f = open(masterdest, "w")
f.write("overwrite me\n")
f.close()
d = self.runStep(step)
def _checkUpload(results):
step_status = step.step_status
#l = step_status.getLogs()
#if l:
# logtext = l[0].getText()
# print logtext
self.failUnlessEqual(results, SUCCESS)
self.failUnless(os.path.exists(masterdest))
masterdest_contents = open(masterdest, "r").read()
self.failUnlessEqual(masterdest_contents, contents)
d.addCallback(_checkUpload)
return maybeWait(d)
def testMissingFile(self):
self.slavebase = "testUploadMissingFile.slave"
self.masterbase = "testUploadMissingFile.master"
sb = self.makeSlaveBuilder()
step = self.makeStep(FileUpload,
slavesrc="MISSING.txt",
masterdest="dest.txt")
masterdest = os.path.join(self.masterbase, "dest.txt")
d = self.runStep(step)
def _checkUpload(results):
step_status = step.step_status
self.failUnlessEqual(results, FAILURE)
self.failIf(os.path.exists(masterdest))
l = step_status.getLogs()
logtext = l[0].getText().strip()
self.failUnless(logtext.startswith("Cannot open file"))
self.failUnless(logtext.endswith("for upload"))
d.addCallback(_checkUpload)
return maybeWait(d)
class Download(StepTester, unittest.TestCase):
def filterArgs(self, args):
if "reader" in args:
args["reader"] = self.wrap(args["reader"])
return args
def testSuccess(self):
self.slavebase = "testDownload.slave"
self.masterbase = "testDownload.master"
sb = self.makeSlaveBuilder()
os.mkdir(os.path.join(self.slavebase, self.slavebuilderbase,
"build"))
mastersrc = os.path.join(self.masterbase, "source.text")
slavedest = os.path.join(self.slavebase,
self.slavebuilderbase,
"build",
"dest.txt")
step = self.makeStep(FileDownload,
mastersrc=mastersrc,
slavedest="dest.txt")
contents = "this is the source file\n"
open(mastersrc, "w").write(contents)
f = open(slavedest, "w")
f.write("overwrite me\n")
f.close()
d = self.runStep(step)
def _checkDownload(results):
step_status = step.step_status
self.failUnlessEqual(results, SUCCESS)
self.failUnless(os.path.exists(slavedest))
slavedest_contents = open(slavedest, "r").read()
self.failUnlessEqual(slavedest_contents, contents)
d.addCallback(_checkDownload)
return maybeWait(d)
def testMissingFile(self):
self.slavebase = "testDownloadMissingFile.slave"
self.masterbase = "testDownloadMissingFile.master"
sb = self.makeSlaveBuilder()
os.mkdir(os.path.join(self.slavebase, self.slavebuilderbase,
"build"))
mastersrc = os.path.join(self.masterbase, "MISSING.text")
slavedest = os.path.join(self.slavebase,
self.slavebuilderbase,
"build",
"dest.txt")
step = self.makeStep(FileDownload,
mastersrc=mastersrc,
slavedest="dest.txt")
d = self.runStep(step)
def _checkDownload(results):
step_status = step.step_status
self.failUnlessEqual(results, FAILURE)
self.failIf(os.path.exists(slavedest))
l = step_status.getLogs()
logtext = l[0].getText().strip()
self.failUnless(logtext.endswith(" not available at master"))
d.addCallback(_checkDownload)
return maybeWait(d)
# TODO:
# test relative paths
# need to implement expanduser() for slave-side
# test error message when master-side file is in a missing directory
# remove workdir= default?
# clean up command start/finish code
More information about the Commits
mailing list