"Fossies" - the Fresh Open Source Software Archive

Member "buildbot-2.3.1/buildbot/test/unit/test_pbmanager.py" (23 May 2019, 5360 Bytes) of package /linux/misc/buildbot-2.3.1.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "test_pbmanager.py": 2.0.1_vs_2.1.0.

    1 # This file is part of Buildbot.  Buildbot is free software: you can
    2 # redistribute it and/or modify it under the terms of the GNU General Public
    3 # License as published by the Free Software Foundation, version 2.
    4 #
    5 # This program is distributed in the hope that it will be useful, but WITHOUT
    6 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
    7 # FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
    8 # details.
    9 #
   10 # You should have received a copy of the GNU General Public License along with
   11 # this program; if not, write to the Free Software Foundation, Inc., 51
   12 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
   13 #
   14 # Copyright Buildbot Team Members
   15 """
   16 Test clean shutdown functionality of the master
   17 """
   18 
   19 import mock
   20 
   21 from twisted.cred import credentials
   22 from twisted.internet import defer
   23 from twisted.spread import pb
   24 from twisted.trial import unittest
   25 
   26 from buildbot import pbmanager
   27 
   28 
   29 class FakeMaster:
   30     initLock = defer.DeferredLock()
   31 
   32     def addService(self, svc):
   33         pass
   34 
   35     @property
   36     def master(self):
   37         return self
   38 
   39 
   40 class TestPBManager(unittest.TestCase):
   41 
   42     def setUp(self):
   43         self.pbm = pbmanager.PBManager()
   44         self.pbm.setServiceParent(FakeMaster())
   45         self.pbm.startService()
   46         self.connections = []
   47 
   48     def tearDown(self):
   49         return self.pbm.stopService()
   50 
   51     def perspectiveFactory(self, mind, username):
   52         persp = mock.Mock()
   53         persp.is_my_persp = True
   54         persp.attached = lambda mind: defer.succeed(None)
   55         self.connections.append(username)
   56         return defer.succeed(persp)
   57 
   58     def test_repr(self):
   59         reg = self.pbm.register(
   60             'tcp:0:interface=127.0.0.1', "x", "y", self.perspectiveFactory)
   61         self.assertEqual(repr(self.pbm.dispatchers['tcp:0:interface=127.0.0.1']),
   62                          '<pbmanager.Dispatcher for x on tcp:0:interface=127.0.0.1>')
   63         self.assertEqual(
   64             repr(reg), '<pbmanager.Registration for x on tcp:0:interface=127.0.0.1>')
   65 
   66     @defer.inlineCallbacks
   67     def test_register_unregister(self):
   68         portstr = "tcp:0:interface=127.0.0.1"
   69         reg = self.pbm.register(
   70             portstr, "boris", "pass", self.perspectiveFactory)
   71 
   72         # make sure things look right
   73         self.assertIn(portstr, self.pbm.dispatchers)
   74         disp = self.pbm.dispatchers[portstr]
   75         self.assertIn('boris', disp.users)
   76 
   77         # we can't actually connect to it, as that requires finding the
   78         # dynamically allocated port number which is buried out of reach;
   79         # however, we can try the requestAvatar and requestAvatarId methods.
   80 
   81         username = yield disp.requestAvatarId(credentials.UsernamePassword(b'boris', b'pass'))
   82 
   83         self.assertEqual(username, b'boris')
   84         avatar = yield disp.requestAvatar(b'boris', mock.Mock(), pb.IPerspective)
   85 
   86         (iface, persp, detach_fn) = avatar
   87         self.assertTrue(persp.is_my_persp)
   88         self.assertIn('boris', self.connections)
   89 
   90         yield reg.unregister()
   91 
   92     @defer.inlineCallbacks
   93     def test_double_register_unregister(self):
   94         portstr = "tcp:0:interface=127.0.0.1"
   95         reg1 = self.pbm.register(portstr, "boris", "pass", None)
   96         reg2 = self.pbm.register(portstr, "ivona", "pass", None)
   97 
   98         # make sure things look right
   99         self.assertEqual(len(self.pbm.dispatchers), 1)
  100         self.assertIn(portstr, self.pbm.dispatchers)
  101         disp = self.pbm.dispatchers[portstr]
  102         self.assertIn('boris', disp.users)
  103         self.assertIn('ivona', disp.users)
  104 
  105         yield reg1.unregister()
  106 
  107         self.assertEqual(len(self.pbm.dispatchers), 1)
  108         self.assertIn(portstr, self.pbm.dispatchers)
  109         disp = self.pbm.dispatchers[portstr]
  110         self.assertNotIn('boris', disp.users)
  111         self.assertIn('ivona', disp.users)
  112         yield reg2.unregister()
  113 
  114         self.assertEqual(len(self.pbm.dispatchers), 0)
  115 
  116     @defer.inlineCallbacks
  117     def test_requestAvatarId_noinitLock(self):
  118         portstr = "tcp:0:interface=127.0.0.1"
  119         reg = self.pbm.register(
  120             portstr, "boris", "pass", self.perspectiveFactory)
  121 
  122         disp = self.pbm.dispatchers[portstr]
  123 
  124         d = disp.requestAvatarId(credentials.UsernamePassword(b'boris', b'pass'))
  125         self.assertTrue(d.called,
  126             "requestAvatarId should have been called since the lock is free")
  127 
  128         yield reg.unregister()
  129 
  130     @defer.inlineCallbacks
  131     def test_requestAvatarId_initLock(self):
  132         portstr = "tcp:0:interface=127.0.0.1"
  133         reg = self.pbm.register(
  134             portstr, "boris", "pass", self.perspectiveFactory)
  135 
  136         disp = self.pbm.dispatchers[portstr]
  137 
  138         try:
  139             # simulate a reconfig/restart in progress
  140             yield self.pbm.master.initLock.acquire()
  141             # try to authenticate while the lock is locked
  142             d = disp.requestAvatarId(credentials.UsernamePassword(b'boris', b'pass'))
  143             self.assertFalse(d.called,
  144                 "requestAvatarId should block until the lock is released")
  145         finally:
  146             # release the lock, it should allow for auth to proceed
  147             yield self.pbm.master.initLock.release()
  148 
  149         self.assertTrue(d.called,
  150             "requestAvatarId should have been called after the lock was released")
  151         yield reg.unregister()