Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_notification_successToException(self):
self.do_x_to_y_notification_test(notify="successToException", previous_result=builder.SUCCESS, new_result=builder.EXCEPTION,
expected_msg="build #862 of builder834 is complete: Exception [step1 step2] Build details are at http://myserver/mypath?build=765" )
self.do_x_to_y_notification_test(notify="successToException", previous_result=builder.SUCCESS, new_result=builder.SUCCESS,
expected_msg = "" )
self.do_x_to_y_notification_test(notify="successToException", previous_result=builder.SUCCESS, new_result=builder.FAILURE,
expected_msg = "" )
self.do_x_to_y_notification_test(notify="successToException", previous_result=builder.SUCCESS, new_result=builder.WARNINGS,
expected_msg = "" )
def testBasic(self):
# test basic substitution with WithProperties
self.props.setProperty("revision", "47", "test")
command = WithProperties("build-%s.tar.gz", "revision")
self.failUnlessEqual(self.props.render(command),
"build-47.tar.gz")
def testReadlines(self):
l = MyLog(self.basedir, "chunks1")
l.addHeader("HEADER\n") # should be ignored
l.addStdout("Some text\n")
l.addStdout("Some More Text\nAnd Some More\n")
l.addStderr("Some Stderr\n")
l.addStdout("Last line\n")
l.finish()
alllines = list(l.readlines())
self.failUnlessEqual(len(alllines), 4)
self.failUnlessEqual(alllines[0], "Some text\n")
self.failUnlessEqual(alllines[2], "And Some More\n")
self.failUnlessEqual(alllines[3], "Last line\n")
stderr = list(l.readlines(interfaces.LOG_CHANNEL_STDERR))
self.failUnlessEqual(len(stderr), 1)
self.failUnlessEqual(stderr[0], "Some Stderr\n")
lines = l.readlines()
if False: # TODO: l.readlines() is not yet an iterator
# verify that it really is an iterator
line0 = lines.next()
self.failUnlessEqual(line0, "Some text\n")
line1 = lines.next()
del line1
line2 = lines.next()
self.failUnlessEqual(line2, "And Some More\n")
self.slaves = {}
if self.basedir is None:
self.basedir = self.mktemp()
basedir = self.basedir
os.makedirs(basedir)
self.master = master.BuildMaster(basedir, **kwargs)
spec = dbspec.DBSpec.from_url("sqlite:///state.sqlite", basedir=basedir)
sm = schema.DBSchemaManager(spec, basedir)
sm.upgrade(quiet=True)
self.master.loadDatabase(spec)
self.master.readConfig = True
self.master.startService()
self.status = self.master.getStatus()
self.control = interfaces.IControl(self.master)
def makeGenerator(self, reasons=[], minTime=0):
b = MyBuilder("bname")
b1 = b.newBuild()
b1.setSourceStamp(SourceStamp(changes=[Change("foo", [], "")]))
b1.buildStarted(b1, 1)
b2 = b.newBuild()
b2.setSourceStamp(SourceStamp(changes=[Change("bar", [], "")]))
b2.buildStarted(b2, 2)
return b.eventGenerator([], [], reasons, minTime)
def test_notification_set_in_config(self):
irc = MyContact(channel = MyChannel(notify_events = {'success': 1}))
my_builder = MyBuilder("builder834")
my_build = MyIrcBuild(my_builder, 862, builder.SUCCESS)
my_build.changes = (
Change(who = 'author1', files = ['file1'], comments = 'comment1', revision = "943"),
)
irc.message = ""
irc.buildFinished(my_builder.getName(), my_build, None)
self.failUnlessEqual(irc.message, "build #862 of builder834 is complete: Success [step1 step2] Build details are at http://myserver/mypath?build=765", "Finish notification generated on success with notify_events=['success']")
os.mkdir(basedir)
mailer = MyMailer(fromaddr="buildbot@example.com", mode="problem",
extraRecipients=["recip@example.com",
"recip2@example.com"],
lookup=MyLookup(),
customMesg=customTextMailMessage)
mailer.parent = self
mailer.master_status = self
self.messages = []
b1 = self.makeBuild(4, builder.FAILURE)
b1.setProperty('buildnumber', 1, 'Build')
b1.setText(["snarkleack", "polarization", "failed"])
b1.blamelist = ["dev3", "dev3", "dev3", "dev4",
"Thomas_Walters"]
b1.source.changes = (Change(who = 'author1', files = ['file1'], comments = 'comment1', revision = "123"),
Change(who = 'author2', files = ['file2'], comments = 'comment2', revision = "456"))
b1.testlogs = [MyLog(basedir, 'compile', "Compile log here\n"),
MyLog(basedir, 'test', "Test log here\nTest 1 failed\nTest 2 failed\nTest 3 failed\nTest 4 failed\n")]
mailer.buildFinished("builder1", b1, b1.results)
m,r = self.messages.pop()
t = m.as_string()
#
# Uncomment to review custom message
#
#self.fail(t)
self.failUnlessIn("comment1", t)
self.failUnlessIn("comment2", t)
self.failUnlessIn("author2", t)
self.failUnlessIn("Test 4 failed", t)
self.failUnlessIn("number was: 1", t)
spec.create_db()
the_db = spec.open_db()
self.dbs.add(the_db)
runner.migrate_changes_pickle_to_db(fn, the_db, silent=True)
m = ChangeManager()
m.parent = FakeMaster()
m.parent.db = the_db
c1 = m.getChangeNumberedNow(1)
self.failUnlessEqual(c1.who, "brian")
self.failUnlessEqual(c1.files, sorted(["foo.c", "subdir/bar.c"]))
self.failUnlessEqual(c1.comments, "first change")
self.failUnlessEqual(c1.revision, "1234")
# more..
c3 = Change("alice", ["baz.c"], "third change", revision="abcd")
m.addChange(c3)
def testMerge(self):
R = BuildRequest
S = sourcestamp.SourceStamp
N = 'test_builder'
b1 = R("why", S("branch1", None, None, None), N)
b1r1 = R("why2", S("branch1", "rev1", None, None), N)
b1r1a = R("why not", S("branch1", "rev1", None, None), N)
b1r2 = R("why3", S("branch1", "rev2", None, None), N)
b2r2 = R("why4", S("branch2", "rev2", None, None), N)
b1r1p1 = R("why5", S("branch1", "rev1", (3, "diff"), None), N)
c1 = Change("alice", [], "changed stuff", branch="branch1")
c2 = Change("alice", [], "changed stuff", branch="branch1")
c3 = Change("alice", [], "changed stuff", branch="branch1")
c4 = Change("alice", [], "changed stuff", branch="branch1")
c5 = Change("alice", [], "changed stuff", branch="branch1")
c6 = Change("alice", [], "changed stuff", branch="branch1")
b1c1 = R("changes", S("branch1", None, None, [c1,c2,c3]), N)
b1c2 = R("changes", S("branch1", None, None, [c4,c5,c6]), N)
self.failUnless(b1.canBeMergedWith(b1))
self.failIf(b1.canBeMergedWith(b1r1))
self.failIf(b1.canBeMergedWith(b2r2))
self.failIf(b1.canBeMergedWith(b1r1p1))
self.failIf(b1.canBeMergedWith(b1c1))
self.failIf(b1r1.canBeMergedWith(b1))
self.failUnless(b1r1.canBeMergedWith(b1r1))
self.failIf(b1r1.canBeMergedWith(b2r2))
def create_pickle(self):
# ugh, what was I thinking? pickles? with class instances? ick.
cm = OldChangeMaster()
cm.basedir = "db/migrate"
os.makedirs(cm.basedir)
c1 = Change(who="brian", files=["foo.c", "subdir/bar.c"],
comments="first change",
revision="1234")
del c1.revlink
c2 = Change(who="brian", files=["foo.c"],
comments="second change",
revision="1235", branch="release",
links=["url1", "url2"], revlink="url3",
properties={"who": "what"},
category="nifty")
cm.addChange(c1)
cm.addChange(c2)
cm.saveYourself() # db/migrate/changes.pck
return os.path.join(cm.basedir, "changes.pck")