@ -1,9 +1,11 @@
import sys
import time
import zlib
import unittest2
import testlib
import mitogen . core
import mitogen . master
import mitogen . parent
import mitogen . utils
@ -341,22 +343,42 @@ class NoRouteTest(testlib.RouterMixin, testlib.TestCase):
) )
class UnidirectionalTest ( testlib . RouterMixin , testlib . TestCase ) :
def test_siblings_cant_talk ( self ) :
self . router . unidirectional = True
l1 = self . router . local ( )
l2 = self . router . local ( )
def test_siblings_cant_talk ( router ) :
l1 = router . local ( )
l2 = router . local ( )
logs = testlib . LogCapturer ( )
logs . start ( )
e = self . assertRaises ( mitogen . core . CallError ,
lambda : l2 . call ( ping_context , l1 ) )
msg = self . router . unidirectional_msg % (
try :
l2 . call ( ping_context , l1 )
except mitogen . core . CallError :
e = sys . exc_info ( ) [ 1 ]
msg = mitogen . core . Router . unidirectional_msg % (
l2 . context_id ,
l1 . context_id ,
)
self . assertTrue ( msg in str ( e ) )
self . assertTrue ( ' routing mode prevents forward of ' in logs . stop ( ) )
assert msg in str ( e )
assert ' routing mode prevents forward of ' in logs . stop ( )
@mitogen.core.takes_econtext
def test_siblings_cant_talk_remote ( econtext ) :
mitogen . parent . upgrade_router ( econtext )
test_siblings_cant_talk ( econtext . router )
class UnidirectionalTest ( testlib . RouterMixin , testlib . TestCase ) :
def test_siblings_cant_talk_master ( self ) :
self . router . unidirectional = True
test_siblings_cant_talk ( self . router )
def test_siblings_cant_talk_parent ( self ) :
# ensure 'unidirectional' attribute is respected for contexts started
# by children.
self . router . unidirectional = True
parent = self . router . local ( )
parent . call ( test_siblings_cant_talk_remote )
def test_auth_id_can_talk ( self ) :
self . router . unidirectional = True