<s> import unittest <EOL> class TestPartylineOperator ( unittest . TestCase ) : <EOL> def _makeOne ( self , partyline ) : <EOL> from wsgi_party import PartylineOperator <EOL> return PartylineOperator ( partyline ) <EOL> def test_connect ( self ) : <EOL> partyline = DummyPartyline ( connect_response = '<STR_LIT>' ) <EOL> inst = self . _makeOne ( partyline ) <EOL> result = inst . connect ( '<STR_LIT:name>' , '<STR_LIT>' ) <EOL> self . assertEqual ( partyline . connections , [ ( '<STR_LIT:name>' , '<STR_LIT>' ) ] ) <EOL> self . assertEqual ( inst . handlers , set ( [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( result , '<STR_LIT>' ) <EOL> def test_ask_around ( self ) : <EOL> partyline = DummyPartyline ( ask_response = [ '<STR_LIT:abc>' ] ) <EOL> inst = self . _makeOne ( partyline ) <EOL> result = inst . ask_around ( '<STR_LIT:name>' , '<STR_LIT>' ) <EOL> self . assertEqual ( partyline . asked , [ ( '<STR_LIT:name>' , '<STR_LIT>' , inst ) ] ) <EOL> self . assertEqual ( result , [ '<STR_LIT:abc>' ] ) <EOL> class TestWSGIParty ( unittest . TestCase ) : <EOL> def _makeOne ( self , app , invites = ( ) , ignore_missing_services = False ) : <EOL> from wsgi_party import WSGIParty <EOL> return WSGIParty ( app , invites , ignore_missing_services ) <EOL> def test_ctor_calls_send_invitations ( self ) : <EOL> app = DummyWSGIApp ( ) <EOL> self . _makeOne ( app , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( len ( app . environs ) , <NUM_LIT:2> ) <EOL> def test___call__ ( self ) : <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> environ = { } <EOL> def start_response ( status , headers = ( ) ) : <EOL> environ [ '<STR_LIT>' ] = True <EOL> inst ( environ , start_response ) <EOL> self . assertEqual ( app . environs , [ environ ] ) <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] , True ) <EOL> def test_send_invitations ( self ) : <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> inst . send_invitations ( ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( len ( app . environs ) , <NUM_LIT:2> ) <EOL> environ1 , environ2 = app . environs <EOL> self . assertEqual ( environ1 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( environ1 [ inst . partyline_key ] . __class__ , <EOL> inst . operator_class ) <EOL> self . assertEqual ( environ2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( environ2 [ inst . partyline_key ] . __class__ , <EOL> inst . operator_class ) <EOL> def test_connect_to_nonexisting ( self ) : <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> inst . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( inst . handlers [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ) <EOL> def test_connect_to_existing ( self ) : <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> inst . handlers [ '<STR_LIT>' ] = [ '<STR_LIT:abc>' ] <EOL> inst . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( inst . handlers [ '<STR_LIT>' ] , [ '<STR_LIT:abc>' , '<STR_LIT>' ] ) <EOL> def test_ask_around_other_operator ( self ) : <EOL> operator = DummyOperator ( ) <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> L = [ ] <EOL> def handler1 ( payload ) : <EOL> L . append ( payload ) <EOL> return '<STR_LIT:result>' <EOL> def handler2 ( payload ) : <EOL> L . append ( payload ) <EOL> return '<STR_LIT>' <EOL> inst . handlers [ '<STR_LIT>' ] = [ handler1 , handler2 ] <EOL> result = inst . ask_around ( '<STR_LIT>' , '<STR_LIT>' , operator = operator ) <EOL> self . assertEqual ( L , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( result , [ '<STR_LIT:result>' , '<STR_LIT>' ] ) <EOL> def test_ask_around_other_operator_handler_raises_HighAndDry ( self ) : <EOL> from wsgi_party import HighAndDry <EOL> operator = DummyOperator ( ) <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> L = [ ] <EOL> def handler1 ( payload ) : <EOL> L . append ( payload ) <EOL> return '<STR_LIT:result>' <EOL> def handler2 ( payload ) : <EOL> L . append ( payload ) <EOL> raise HighAndDry ( ) <EOL> inst . handlers [ '<STR_LIT>' ] = [ handler1 , handler2 ] <EOL> result = inst . ask_around ( '<STR_LIT>' , '<STR_LIT>' , operator = operator ) <EOL> self . assertEqual ( L , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( result , [ '<STR_LIT:result>' ] ) <EOL> def test_ask_around_same_operator ( self ) : <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> L = [ ] <EOL> def handler1 ( payload ) : <EOL> L . append ( payload ) <EOL> return '<STR_LIT:result>' <EOL> def handler2 ( payload ) : <EOL> L . append ( payload ) <EOL> return '<STR_LIT>' <EOL> operator = DummyOperator ( ( handler1 , handler2 ) ) <EOL> inst . handlers [ '<STR_LIT>' ] = [ handler1 , handler2 ] <EOL> result = inst . ask_around ( '<STR_LIT>' , '<STR_LIT>' , operator = operator ) <EOL> self . assertEqual ( L , [ ] ) <EOL> self . assertEqual ( result , [ ] ) <EOL> def test_ask_around_no_operator ( self ) : <EOL> from wsgi_party import HighAndDry <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> L = [ ] <EOL> def handler1 ( payload ) : <EOL> L . append ( payload ) <EOL> return '<STR_LIT:result>' <EOL> def handler2 ( payload ) : <EOL> L . append ( payload ) <EOL> raise HighAndDry ( ) <EOL> inst . handlers [ '<STR_LIT>' ] = [ handler1 , handler2 ] <EOL> result = inst . ask_around ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( L , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( result , [ '<STR_LIT:result>' ] ) <EOL> def test_ask_around_no_handler ( self ) : <EOL> from wsgi_party import NoSuchServiceName <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app ) <EOL> self . assertRaises ( NoSuchServiceName , inst . ask_around , '<STR_LIT>' , None ) <EOL> def test_ask_around_no_handler_ignored ( self ) : <EOL> from wsgi_party import NoSuchServiceName <EOL> app = DummyWSGIApp ( ) <EOL> inst = self . _makeOne ( app , ignore_missing_services = True ) <EOL> try : <EOL> self . assertEqual ( inst . ask_around ( '<STR_LIT>' , None ) , [ ] ) <EOL> except NoSuchServiceName : <EOL> self . fail ( '<STR_LIT>' ) <EOL> class DummyOperator ( object ) : <EOL> def __init__ ( self , handlers = ( ) ) : <EOL> self . handlers = handlers <EOL> class DummyPartyline ( object ) : <EOL> def __init__ ( self , connect_response = None , ask_response = None ) : <EOL> self . connections = [ ] <EOL> self . asked = [ ] <EOL> self . ask_response = ask_response <EOL> self . connect_response = connect_response <EOL> def connect ( self , name , handler ) : <EOL> self . connections . append ( ( name , handler ) ) <EOL> return self . connect_response <EOL> def ask_around ( self , service_name , payload , operator = None ) : <EOL> self . asked . append ( ( service_name , payload , operator ) ) <EOL> return self . ask_response <EOL> class DummyWSGIApp ( object ) : <EOL> def __init__ ( self , response = ( ) ) : <EOL> self . response = response <EOL> self . environs = [ ] <EOL> def __call__ ( self , environ , start_response ) : <EOL> self . environs . append ( environ ) <EOL> start_response ( '<STR_LIT>' , [ ] ) <EOL> return self . response </s>
