<s> import socket <EOL> import mock <EOL> from twisted . trial import unittest <EOL> from twisted . internet import defer , reactor <EOL> from twisted . application import service <EOL> from twisted . python import reflect , log <EOL> from twisted . spread import pb , flavors <EOL> from twisted . cred import credentials , error <EOL> from piped import processing , dependencies , util , exceptions <EOL> from piped . providers import spread_provider , pipeline_provider <EOL> class FakeProcessor ( object ) : <EOL> def __init__ ( self ) : <EOL> self . batons = defer . DeferredQueue ( ) <EOL> self . i = <NUM_LIT:0> <EOL> def __call__ ( self , baton ) : <EOL> self . batons . put ( baton ) <EOL> self . i += <NUM_LIT:1> <EOL> return [ self . i ] <EOL> class PBTestBase ( unittest . TestCase ) : <EOL> timeout = <NUM_LIT:2> <EOL> def setUp ( self ) : <EOL> self . runtime_environment = processing . RuntimeEnvironment ( ) <EOL> self . application = self . runtime_environment . application = service . MultiService ( ) <EOL> def replace_twisted_log_err ( self , err = lambda failure = None : None ) : <EOL> original = log . err <EOL> log . err = err <EOL> def restore ( ) : <EOL> log . err = original <EOL> self . addCleanup ( restore ) <EOL> def get_free_port ( self ) : <EOL> sock = socket . socket ( ) <EOL> sock . bind ( ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> free_port = sock . getsockname ( ) [ <NUM_LIT:1> ] <EOL> sock . close ( ) <EOL> return free_port <EOL> def create_pb_server ( self , ** kwargs ) : <EOL> self . port = self . get_free_port ( ) <EOL> listen = '<STR_LIT>' % self . port <EOL> kwargs [ '<STR_LIT>' ] = listen <EOL> kwargs . setdefault ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> kwargs . setdefault ( '<STR_LIT>' , False ) <EOL> self . server = spread_provider . PipedPBService ( ** kwargs ) <EOL> self . server . setServiceParent ( self . application ) <EOL> self . server . configure ( self . runtime_environment ) <EOL> class PBServerTest ( PBTestBase ) : <EOL> def setUp ( self ) : <EOL> super ( PBServerTest , self ) . setUp ( ) <EOL> self . processor = FakeProcessor ( ) <EOL> self . processor_dependency = dependencies . InstanceDependency ( self . processor ) <EOL> self . processor_dependency . is_ready = True <EOL> self . replace_twisted_log_err ( ) <EOL> self . runtime_environment . configure ( ) <EOL> @ defer . inlineCallbacks <EOL> def test_simple_server ( self ) : <EOL> self . create_pb_server ( ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> self . server . processor_dependency = self . processor_dependency <EOL> client = pb . PBClientFactory ( ) <EOL> reactor . connectTCP ( '<STR_LIT:localhost>' , self . port , client ) <EOL> root = yield client . getRootObject ( ) <EOL> self . addCleanup ( root . broker . transport . loseConnection ) <EOL> adding = root . callRemote ( '<STR_LIT>' , <NUM_LIT> , b = <NUM_LIT> ) <EOL> baton = yield self . processor . batons . get ( ) <EOL> self . assertEquals ( baton [ '<STR_LIT:message>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( baton [ '<STR_LIT:args>' ] , ( <NUM_LIT> , ) ) <EOL> self . assertEquals ( baton [ '<STR_LIT>' ] , dict ( b = <NUM_LIT> ) ) <EOL> baton [ '<STR_LIT>' ] . callback ( <NUM_LIT> + <NUM_LIT> ) <EOL> result = yield adding <EOL> self . assertEquals ( result , <NUM_LIT> + <NUM_LIT> ) <EOL> @ defer . inlineCallbacks <EOL> def test_waiting_for ( self ) : <EOL> processor_dependency = dependencies . ResourceDependency ( provider = '<STR_LIT>' ) <EOL> self . create_pb_server ( ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> self . server . processor_dependency = processor_dependency <EOL> client = pb . PBClientFactory ( ) <EOL> reactor . connectTCP ( '<STR_LIT:localhost>' , self . port , client ) <EOL> root = yield client . getRootObject ( ) <EOL> self . addCleanup ( root . broker . transport . loseConnection ) <EOL> try : <EOL> yield root . callRemote ( '<STR_LIT>' , <NUM_LIT> , b = <NUM_LIT> ) <EOL> except spread_provider . RemoteError as e : <EOL> self . assertEquals ( e . remoteType , reflect . fullyQualifiedName ( exceptions . UnprovidedResourceError ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> self . server . wait_for_processor = True <EOL> adding = root . callRemote ( '<STR_LIT>' , <NUM_LIT> , b = <NUM_LIT> ) <EOL> processor_dependency . on_resource_ready ( self . processor ) <EOL> processor_dependency . fire_on_ready ( ) <EOL> baton = yield self . processor . batons . get ( ) <EOL> baton [ '<STR_LIT>' ] . callback ( <NUM_LIT> + <NUM_LIT> ) <EOL> yield adding <EOL> @ defer . inlineCallbacks <EOL> def test_default_callbacks ( self ) : <EOL> self . create_pb_server ( ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> self . server . processor_dependency = self . processor_dependency <EOL> client = pb . PBClientFactory ( ) <EOL> reactor . connectTCP ( '<STR_LIT:localhost>' , self . port , client ) <EOL> root = yield client . getRootObject ( ) <EOL> self . addCleanup ( root . broker . transport . loseConnection ) <EOL> adding = root . callRemote ( '<STR_LIT>' , <NUM_LIT> , b = <NUM_LIT> ) <EOL> baton = yield self . processor . batons . get ( ) <EOL> self . assertEquals ( baton [ '<STR_LIT:message>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( baton [ '<STR_LIT:args>' ] , ( <NUM_LIT> , ) ) <EOL> self . assertEquals ( baton [ '<STR_LIT>' ] , dict ( b = <NUM_LIT> ) ) <EOL> del baton <EOL> try : <EOL> yield adding <EOL> except spread_provider . RemoteError as e : <EOL> self . assertEquals ( e . remoteType , reflect . fullyQualifiedName ( exceptions . MissingCallback ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> self . server . default_callback = '<STR_LIT>' <EOL> adding = root . callRemote ( '<STR_LIT>' , <NUM_LIT> , b = <NUM_LIT> ) <EOL> yield self . processor . batons . get ( ) <EOL> result = yield adding <EOL> self . assertEquals ( result , self . server . default_callback ) <EOL> @ defer . inlineCallbacks <EOL> def test_default_callbacks_after_failure ( self ) : <EOL> self . create_pb_server ( ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> self . server . processor_dependency = self . processor_dependency <EOL> client = pb . PBClientFactory ( ) <EOL> reactor . connectTCP ( '<STR_LIT:localhost>' , self . port , client ) <EOL> root = yield client . getRootObject ( ) <EOL> self . addCleanup ( root . broker . transport . loseConnection ) <EOL> adding = root . callRemote ( '<STR_LIT>' , <NUM_LIT> , b = <NUM_LIT> ) <EOL> baton = yield self . processor . batons . get ( ) <EOL> try : <EOL> raise Exception ( ) <EOL> except Exception as e : <EOL> from twisted . python import failure <EOL> f = failure . Failure ( ) <EOL> del baton <EOL> del f <EOL> locals ( ) <EOL> try : <EOL> yield adding <EOL> except spread_provider . RemoteError as e : <EOL> self . assertEquals ( e . remoteType , reflect . fullyQualifiedName ( exceptions . MissingCallback ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> @ defer . inlineCallbacks <EOL> def test_require_login ( self ) : <EOL> self . runtime_environment . configure ( ) <EOL> self . create_pb_server ( checker = dict ( <EOL> name = '<STR_LIT>' , <EOL> arguments = dict ( test_username = '<STR_LIT>' ) <EOL> ) ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> self . server . processor_dependency = self . processor_dependency <EOL> client = pb . PBClientFactory ( ) <EOL> reactor . connectTCP ( '<STR_LIT:localhost>' , self . port , client ) <EOL> root = yield client . getRootObject ( ) <EOL> self . addCleanup ( root . broker . transport . loseConnection ) <EOL> try : <EOL> yield root . callRemote ( '<STR_LIT>' , <NUM_LIT> , b = <NUM_LIT> ) <EOL> except spread_provider . RemoteError as e : <EOL> self . assertEquals ( e . remoteType , reflect . fullyQualifiedName ( flavors . NoSuchMethod ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> bad_credentials = list ( ) <EOL> bad_credentials . append ( credentials . UsernamePassword ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> bad_credentials . append ( credentials . UsernamePassword ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> bad_credentials . append ( credentials . UsernamePassword ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> for bad_credential in bad_credentials : <EOL> try : <EOL> yield client . login ( bad_credential ) <EOL> except spread_provider . RemoteError as e : <EOL> self . assertEquals ( e . remoteType , reflect . fullyQualifiedName ( error . UnauthorizedLogin ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> perspective = yield client . login ( credentials . UsernamePassword ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> adding = perspective . callRemote ( '<STR_LIT>' , <NUM_LIT> , b = <NUM_LIT> ) <EOL> baton = yield self . processor . batons . get ( ) <EOL> self . assertEquals ( baton [ '<STR_LIT:message>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( baton [ '<STR_LIT:args>' ] , ( <NUM_LIT> , ) ) <EOL> self . assertEquals ( baton [ '<STR_LIT>' ] , dict ( b = <NUM_LIT> ) ) <EOL> baton [ '<STR_LIT>' ] . callback ( <NUM_LIT> + <NUM_LIT> ) <EOL> result = yield adding <EOL> self . assertEquals ( result , <NUM_LIT> + <NUM_LIT> ) <EOL> class PBClientTest ( PBTestBase ) : <EOL> timeout = <NUM_LIT:2> <EOL> @ defer . inlineCallbacks <EOL> def test_reconnecting ( self ) : <EOL> self . runtime_environment . configure ( ) <EOL> cm = self . runtime_environment . configuration_manager <EOL> dm = self . runtime_environment . dependency_manager <EOL> self . create_pb_server ( ) <EOL> cm . set ( '<STR_LIT>' , '<STR_LIT>' % self . port ) <EOL> cm . set ( '<STR_LIT>' , [ { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ] ) <EOL> pp = pipeline_provider . PipelineProvider ( ) <EOL> pp . configure ( self . runtime_environment ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> client_provider = spread_provider . PBClientProvider ( ) <EOL> client_provider . configure ( self . runtime_environment ) <EOL> client_factory_dependency = dm . add_dependency ( self , dict ( provider = '<STR_LIT>' ) ) <EOL> client_root_dependency = dm . add_dependency ( self , dict ( provider = '<STR_LIT>' ) ) <EOL> dm . resolve_initial_states ( ) <EOL> factory = yield client_factory_dependency . wait_for_resource ( ) <EOL> self . assertIsInstance ( factory , spread_provider . PipedPBClientFactory ) <EOL> factory . max_delay = <NUM_LIT:0> <EOL> root = yield client_root_dependency . wait_for_resource ( ) <EOL> self . assertIsInstance ( root , pb . RemoteReference ) <EOL> yield self . server . stopService ( ) <EOL> try : <EOL> yield root . callRemote ( '<STR_LIT>' ) <EOL> except pb . PBConnectionLost as e : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> yield util . wait ( <NUM_LIT:0> ) <EOL> self . assertFalse ( client_factory_dependency . is_ready ) <EOL> self . assertFalse ( client_root_dependency . is_ready ) <EOL> yield self . server . startService ( ) <EOL> new_factory = yield client_factory_dependency . wait_for_resource ( ) <EOL> new_root = yield client_root_dependency . wait_for_resource ( ) <EOL> self . assertEquals ( new_factory , factory ) <EOL> self . assertNotEquals ( new_root , root ) <EOL> result = yield new_root . callRemote ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , '<STR_LIT:foo>' ) <EOL> @ defer . inlineCallbacks <EOL> def test_valid_login ( self ) : <EOL> self . runtime_environment . configure ( ) <EOL> cm = self . runtime_environment . configuration_manager <EOL> dm = self . runtime_environment . dependency_manager <EOL> self . create_pb_server ( checker = dict ( <EOL> name = '<STR_LIT>' , <EOL> arguments = dict ( test_username = '<STR_LIT>' ) <EOL> ) ) <EOL> cm . set ( '<STR_LIT>' , dict ( <EOL> endpoint = '<STR_LIT>' % self . port , <EOL> username = '<STR_LIT>' , <EOL> password = '<STR_LIT>' <EOL> ) ) <EOL> cm . set ( '<STR_LIT>' , [ { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ] ) <EOL> pp = pipeline_provider . PipelineProvider ( ) <EOL> pp . configure ( self . runtime_environment ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> client_provider = spread_provider . PBClientProvider ( ) <EOL> client_provider . configure ( self . runtime_environment ) <EOL> client_root_dependency = dm . add_dependency ( self , dict ( provider = '<STR_LIT>' ) ) <EOL> dm . resolve_initial_states ( ) <EOL> root = yield client_root_dependency . wait_for_resource ( ) <EOL> result = yield root . callRemote ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , '<STR_LIT>' ) <EOL> @ defer . inlineCallbacks <EOL> def test_invalid_login ( self ) : <EOL> self . runtime_environment . configure ( ) <EOL> cm = self . runtime_environment . configuration_manager <EOL> dm = self . runtime_environment . dependency_manager <EOL> self . create_pb_server ( checker = dict ( <EOL> name = '<STR_LIT>' , <EOL> arguments = dict ( test_username = '<STR_LIT>' ) <EOL> ) ) <EOL> cm . set ( '<STR_LIT>' , dict ( <EOL> endpoint = '<STR_LIT>' % self . port , <EOL> username = '<STR_LIT>' , <EOL> password = '<STR_LIT>' <EOL> ) ) <EOL> cm . set ( '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> pp = pipeline_provider . PipelineProvider ( ) <EOL> pp . configure ( self . runtime_environment ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> client_provider = spread_provider . PBClientProvider ( ) <EOL> client_provider . configure ( self . runtime_environment ) <EOL> self . replace_twisted_log_err ( ) <EOL> with mock . patch . object ( spread_provider , '<STR_LIT>' ) as mocked_logger : <EOL> client_wrong_password_factory_dependency = dm . add_dependency ( self , dict ( provider = '<STR_LIT>' ) ) <EOL> dm . resolve_initial_states ( ) <EOL> wrong_pw_factory = yield client_wrong_password_factory_dependency . wait_for_resource ( ) <EOL> ( args , kwargs ) = yield wrong_pw_factory . on_failed_getting_root_object . wait_until_fired ( ) <EOL> reason = args [ <NUM_LIT:0> ] <EOL> self . assertEquals ( reason . type , '<STR_LIT>' ) <EOL> yield util . wait ( <NUM_LIT:0> ) <EOL> @ defer . inlineCallbacks <EOL> def test_manual_login ( self ) : <EOL> self . runtime_environment . configure ( ) <EOL> cm = self . runtime_environment . configuration_manager <EOL> dm = self . runtime_environment . dependency_manager <EOL> self . create_pb_server ( checker = dict ( <EOL> name = '<STR_LIT>' , <EOL> arguments = dict ( test_username = '<STR_LIT>' ) <EOL> ) ) <EOL> cm . set ( '<STR_LIT>' , dict ( <EOL> endpoint = '<STR_LIT>' % self . port , <EOL> ) ) <EOL> cm . set ( '<STR_LIT>' , [ { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ] ) <EOL> pp = pipeline_provider . PipelineProvider ( ) <EOL> pp . configure ( self . runtime_environment ) <EOL> self . application . startService ( ) <EOL> self . addCleanup ( self . application . stopService ) <EOL> client_provider = spread_provider . PBClientProvider ( ) <EOL> client_provider . configure ( self . runtime_environment ) <EOL> client_no_login_root_dependency = dm . add_dependency ( self , dict ( provider = '<STR_LIT>' ) ) <EOL> dm . resolve_initial_states ( ) <EOL> no_login_root = yield client_no_login_root_dependency . wait_for_resource ( ) <EOL> challenge , challenger = yield no_login_root . callRemote ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> perspective = yield challenger . callRemote ( '<STR_LIT>' , pb . respond ( challenge , '<STR_LIT>' ) , mind = None ) <EOL> result = yield perspective . callRemote ( '<STR_LIT>' ) <EOL> self . assertEquals ( result , '<STR_LIT>' ) <EOL> class PBServerProviderTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . runtime_environment = processing . RuntimeEnvironment ( ) <EOL> self . application = self . runtime_environment . application = service . MultiService ( ) <EOL> self . runtime_environment . configure ( ) <EOL> def test_server_is_provided ( self ) : <EOL> cm = self . runtime_environment . configuration_manager <EOL> cm . set ( '<STR_LIT>' , dict ( <EOL> processor = '<STR_LIT>' , <EOL> listen = '<STR_LIT>' <EOL> ) ) <EOL> provider = spread_provider . PBServerProvider ( ) <EOL> provider . configure ( self . runtime_environment ) <EOL> dep = dependencies . ResourceDependency ( provider = '<STR_LIT>' ) <EOL> provider . add_consumer ( dep ) <EOL> server = dep . get_resource ( ) <EOL> self . assertEquals ( server . processor_dependency . provider , '<STR_LIT>' ) </s>
