Rooba
/
mitogen
mirror of https://github.com/mitogen-hq/mitogen.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

issue #477: stop using fork in service_test.

This commit is contained in:
David Wilson 2019-01-23 12:44:08 +00:00
parent f1a712fad3
commit 9d5b3d9e02
1 changed files with 10 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
tests/service_test.py
View File

@ -38,21 +38,21 @@ def call_service_in(context, service_name, method_name):
class ActivationTest(testlib.RouterMixin, testlib.TestCase):
def test_parent_can_activate(self):
l1 = self.router.fork()
l1 = self.router.local()
counter, id_ = l1.call_service(MyService, 'get_id')
self.assertEquals(1, counter)
self.assertTrue(isinstance(id_, int))
def test_sibling_cannot_activate_framework(self):
l1 = self.router.fork()
l2 = self.router.fork()
l1 = self.router.local()
l2 = self.router.local()
exc = self.assertRaises(mitogen.core.CallError,
lambda: l2.call(call_service_in, l1, MyService2.name(), 'get_id'))
self.assertTrue(mitogen.core.Router.refused_msg in exc.args[0])
def test_sibling_cannot_activate_service(self):
l1 = self.router.fork()
l2 = self.router.fork()
l1 = self.router.local()
l2 = self.router.local()
l1.call_service(MyService, 'get_id') # force framework activation
capture = testlib.LogCapturer()
capture.start()
@ -65,7 +65,7 @@ class ActivationTest(testlib.RouterMixin, testlib.TestCase):
self.assertTrue(msg in exc.args[0])
def test_activates_only_once(self):
l1 = self.router.fork()
l1 = self.router.local()
counter, id_ = l1.call_service(MyService, 'get_id')
counter2, id_2 = l1.call_service(MyService, 'get_id')
self.assertEquals(1, counter)
@ -75,16 +75,16 @@ class ActivationTest(testlib.RouterMixin, testlib.TestCase):
class PermissionTest(testlib.RouterMixin, testlib.TestCase):
def test_sibling_unprivileged_ok(self):
l1 = self.router.fork()
l1 = self.router.local()
l1.call_service(MyService, 'get_id')
l2 = self.router.fork()
l2 = self.router.local()
self.assertEquals('unprivileged!',
l2.call(call_service_in, l1, MyService.name(), 'unprivileged_op'))
def test_sibling_privileged_bad(self):
l1 = self.router.fork()
l1 = self.router.local()
l1.call_service(MyService, 'get_id')
l2 = self.router.fork()
l2 = self.router.local()
capture = testlib.LogCapturer()
capture.start()
try: