Bye bye buildbot 0.7.12.

third_party/buildbot_7_12/buildbot/test/test_p4poller.py

-import time
-
-from twisted.internet import defer
-from twisted.trial import unittest
-
-from buildbot.changes.changes import Change
-from buildbot.changes.p4poller import P4Source, get_simple_split
-
-first_p4changes = \
-"""Change 1 on 2006/04/13 by slamb@testclient 'first rev'
-"""
-
-second_p4changes = \
-"""Change 3 on 2006/04/13 by bob@testclient 'short desc truncated'
-Change 2 on 2006/04/13 by slamb@testclient 'bar'
-"""
-
-third_p4changes = \
-"""Change 5 on 2006/04/13 by mpatel@testclient 'first rev'
-"""
-
-change_4_log = \
-"""Change 4 by mpatel@testclient on 2006/04/13 21:55:39
-
-        short desc truncated because this is a long description.
-"""
-change_3_log = \
-"""Change 3 by bob@testclient on 2006/04/13 21:51:39
-
-        short desc truncated because this is a long description.
-"""
-
-change_2_log = \
-"""Change 2 by slamb@testclient on 2006/04/13 21:46:23
-
-        creation
-"""
-
-p4change = {
-    3: change_3_log +
-"""Affected files ...
-
-... //depot/myproject/branch_b/branch_b_file#1 add
-... //depot/myproject/branch_b/whatbranch#1 branch
-... //depot/myproject/branch_c/whatbranch#1 branch
-""",
-    2: change_2_log +
-"""Affected files ...
-
-... //depot/myproject/trunk/whatbranch#1 add
-... //depot/otherproject/trunk/something#1 add
-""",
-    5: change_4_log +
-"""Affected files ...
-
-... //depot/myproject/branch_b/branch_b_file#1 add
-... //depot/myproject/branch_b#75 edit
-... //depot/myproject/branch_c/branch_c_file#1 add
-""",
-}
-
-
-class MockP4Source(P4Source):
-    """Test P4Source which doesn't actually invoke p4."""
-    invocation = 0
-
-    def __init__(self, p4changes, p4change, *args, **kwargs):
-        P4Source.__init__(self, *args, **kwargs)
-        self.p4changes = p4changes
-        self.p4change = p4change
-
-    def _get_changes(self):
-        assert self.working
-        result = self.p4changes[self.invocation]
-        self.invocation += 1
-        return defer.succeed(result)
-
-    def _get_describe(self, dummy, num):
-        assert self.working
-        return defer.succeed(self.p4change[num])
-
-class TestP4Poller(unittest.TestCase):
-    def setUp(self):
-        self.changes = []
-        self.addChange = self.changes.append
-
-    def failUnlessIn(self, substr, string):
-        # this is for compatibility with python2.2
-        if isinstance(string, str):
-            self.failUnless(string.find(substr) != -1)
-        else:
-            self.assertIn(substr, string)
-
-    def testCheck(self):
-        """successful checks"""
-        self.t = MockP4Source(p4changes=[first_p4changes, second_p4changes],
-                              p4change=p4change,
-                              p4port=None, p4user=None,
-                              p4base='//depot/myproject/',
-                              split_file=lambda x: x.split('/', 1))
-        self.t.parent = self
-
-        # The first time, it just learns the change to start at.
-        self.assert_(self.t.last_change is None)
-        self.assert_(not self.t.working)
-        return self.t.checkp4().addCallback(self._testCheck2)
-
-    def _testCheck2(self, res):
-        self.assertEquals(self.changes, [])
-        self.assertEquals(self.t.last_change, 1)
-
-        # Subsequent times, it returns Change objects for new changes.
-        return self.t.checkp4().addCallback(self._testCheck3)
-
-    def _testCheck3(self, res):
-        self.assertEquals(len(self.changes), 3)
-        self.assertEquals(self.t.last_change, 3)
-        self.assert_(not self.t.working)
-
-        # They're supposed to go oldest to newest, so this one must be first.
-        self.assertEquals(self.changes[0].asText(),
-            Change(who='slamb',
-                   files=['whatbranch'],
-                   comments=change_2_log,
-                   revision='2',
-                   when=self.makeTime("2006/04/13 21:46:23"),
-                   branch='trunk').asText())
-
-        # These two can happen in either order, since they're from the same
-        # Perforce change.
-        self.failUnlessIn(
-            Change(who='bob',
-                   files=['branch_b_file',
-                          'whatbranch'],
-                   comments=change_3_log,
-                   revision='3',
-                   when=self.makeTime("2006/04/13 21:51:39"),
-                   branch='branch_b').asText(),
-            [c.asText() for c in self.changes])
-        self.failUnlessIn(
-            Change(who='bob',
-                   files=['whatbranch'],
-                   comments=change_3_log,
-                   revision='3',
-                   when=self.makeTime("2006/04/13 21:51:39"),
-                   branch='branch_c').asText(),
-            [c.asText() for c in self.changes])
-
-    def makeTime(self, timestring):
-        datefmt = '%Y/%m/%d %H:%M:%S'
-        when = time.mktime(time.strptime(timestring, datefmt))
-        return when
-
-    def testFailedChanges(self):
-        """'p4 changes' failure is properly ignored"""
-        self.t = MockP4Source(p4changes=['Perforce client error:\n...'],
-                              p4change={},
-                              p4port=None, p4user=None)
-        self.t.parent = self
-        d = self.t.checkp4()
-        d.addCallback(self._testFailedChanges2)
-        return d
-
-    def _testFailedChanges2(self, f):
-        self.failUnlessEqual(f, None)
-        self.assert_(not self.t.working)
-
-    def testFailedDescribe(self):
-        """'p4 describe' failure is properly ignored"""
-        c = dict(p4change)
-        c[3] = 'Perforce client error:\n...'
-        self.t = MockP4Source(p4changes=[first_p4changes, second_p4changes],
-                              p4change=c, p4port=None, p4user=None)
-        self.t.parent = self
-        d = self.t.checkp4()
-        d.addCallback(self._testFailedDescribe2)
-        return d
-
-    def _testFailedDescribe2(self, res):
-        # first time finds nothing; check again.
-        return self.t.checkp4().addCallback(self._testFailedDescribe3)
-
-    def _testFailedDescribe3(self, f):
-        self.failUnlessEqual(f, None)
-        self.assert_(not self.t.working)
-        self.assertEquals(self.t.last_change, 2)
-
-    def testAlreadyWorking(self):
-        """don't launch a new poll while old is still going"""
-        self.t = P4Source()
-        self.t.working = True
-        self.assert_(self.t.last_change is None)
-        d = self.t.checkp4()
-        d.addCallback(self._testAlreadyWorking2)
-
-    def _testAlreadyWorking2(self, res):
-        self.assert_(self.t.last_change is None)
-
-    def testSplitFile(self):
-        """Make sure split file works on branch only changes"""
-        self.t = MockP4Source(p4changes=[third_p4changes],
-                              p4change=p4change,
-                              p4port=None, p4user=None,
-                              p4base='//depot/myproject/',
-                              split_file=get_simple_split)
-        self.t.parent = self
-        self.t.last_change = 50
-        d = self.t.checkp4()
-        d.addCallback(self._testSplitFile)
-
-    def _testSplitFile(self, res):
-        self.assertEquals(len(self.changes), 2)
-        self.assertEquals(self.t.last_change, 5)