aboutsummaryrefslogtreecommitdiffstats
path: root/lib/python2.7/site-packages/buildbot-0.8.8-py2.7.egg/buildbot/test/unit/test_pbmanager.py
diff options
context:
space:
mode:
Diffstat (limited to 'lib/python2.7/site-packages/buildbot-0.8.8-py2.7.egg/buildbot/test/unit/test_pbmanager.py')
-rw-r--r--lib/python2.7/site-packages/buildbot-0.8.8-py2.7.egg/buildbot/test/unit/test_pbmanager.py98
1 files changed, 0 insertions, 98 deletions
diff --git a/lib/python2.7/site-packages/buildbot-0.8.8-py2.7.egg/buildbot/test/unit/test_pbmanager.py b/lib/python2.7/site-packages/buildbot-0.8.8-py2.7.egg/buildbot/test/unit/test_pbmanager.py
deleted file mode 100644
index 032c8ff9..00000000
--- a/lib/python2.7/site-packages/buildbot-0.8.8-py2.7.egg/buildbot/test/unit/test_pbmanager.py
+++ /dev/null
@@ -1,98 +0,0 @@
-# This file is part of Buildbot. Buildbot is free software: you can
-# redistribute it and/or modify it under the terms of the GNU General Public
-# License as published by the Free Software Foundation, version 2.
-#
-# This program is distributed in the hope that it will be useful, but WITHOUT
-# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
-# details.
-#
-# You should have received a copy of the GNU General Public License along with
-# this program; if not, write to the Free Software Foundation, Inc., 51
-# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
-#
-# Copyright Buildbot Team Members
-
-# Test clean shutdown functionality of the master
-import mock
-from twisted.trial import unittest
-from twisted.internet import defer
-from twisted.spread import pb
-from twisted.cred import credentials
-from buildbot import pbmanager
-
-class TestPBManager(unittest.TestCase):
-
- def setUp(self):
- self.pbm = pbmanager.PBManager()
- self.pbm.startService()
- self.connections = []
-
- def tearDown(self):
- return self.pbm.stopService()
-
- def perspectiveFactory(self, mind, username):
- persp = mock.Mock()
- persp.is_my_persp = True
- persp.attached = lambda mind : defer.succeed(None)
- self.connections.append(username)
- return defer.succeed(persp)
-
- def test_repr(self):
- reg = self.pbm.register('tcp:0:interface=127.0.0.1', "x", "y", self.perspectiveFactory)
- self.assertEqual(`self.pbm.dispatchers['tcp:0:interface=127.0.0.1']`,
- '<pbmanager.Dispatcher for x on tcp:0:interface=127.0.0.1>')
- self.assertEqual(`reg`, '<pbmanager.Registration for x on tcp:0:interface=127.0.0.1>')
-
- def test_register_unregister(self):
- portstr = "tcp:0:interface=127.0.0.1"
- reg = self.pbm.register(portstr, "boris", "pass", self.perspectiveFactory)
-
- # make sure things look right
- self.assertIn(portstr, self.pbm.dispatchers)
- disp = self.pbm.dispatchers[portstr]
- self.assertIn('boris', disp.users)
-
- # we can't actually connect to it, as that requires finding the
- # dynamically allocated port number which is buried out of reach;
- # however, we can try the requestAvatar and requestAvatarId methods.
-
- d = disp.requestAvatarId(credentials.UsernamePassword('boris', 'pass'))
- def check_avatarid(username):
- self.assertEqual(username, 'boris')
- d.addCallback(check_avatarid)
- d.addCallback(lambda _ :
- disp.requestAvatar('boris', mock.Mock(), pb.IPerspective))
- def check_avatar((iface, persp, detach_fn)):
- self.failUnless(persp.is_my_persp)
- self.assertIn('boris', self.connections)
- d.addCallback(check_avatar)
-
- d.addCallback(lambda _ : reg.unregister())
- return d
-
- def test_double_register_unregister(self):
- portstr = "tcp:0:interface=127.0.0.1"
- reg1 = self.pbm.register(portstr, "boris", "pass", None)
- reg2 = self.pbm.register(portstr, "ivona", "pass", None)
-
- # make sure things look right
- self.assertEqual(len(self.pbm.dispatchers), 1)
- self.assertIn(portstr, self.pbm.dispatchers)
- disp = self.pbm.dispatchers[portstr]
- self.assertIn('boris', disp.users)
- self.assertIn('ivona', disp.users)
-
- d = reg1.unregister()
- def check_boris_gone(_):
- self.assertEqual(len(self.pbm.dispatchers), 1)
- self.assertIn(portstr, self.pbm.dispatchers)
- disp = self.pbm.dispatchers[portstr]
- self.assertNotIn('boris', disp.users)
- self.assertIn('ivona', disp.users)
- d.addCallback(check_boris_gone)
- d.addCallback(lambda _ : reg2.unregister())
- def check_dispatcher_gone(_):
- self.assertEqual(len(self.pbm.dispatchers), 0)
- d.addCallback(check_dispatcher_gone)
- return d