<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import <EOL> import socket <EOL> from gc import collect <EOL> from weakref import ref <EOL> from zope . interface . verify import verifyObject <EOL> from twisted . python import context , log <EOL> from twisted . python . failure import Failure <EOL> from twisted . python . runtime import platform <EOL> from twisted . python . log import ILogContext , msg , err <EOL> from twisted . internet . defer import Deferred , gatherResults <EOL> from twisted . internet . interfaces import IConnector , IReactorFDSet <EOL> from twisted . internet . protocol import ClientFactory , Protocol , ServerFactory <EOL> from twisted . trial . unittest import SkipTest <EOL> from twisted . internet . test . reactormixins import needsRunningReactor <EOL> from twisted . test . test_tcp import ClosingProtocol <EOL> def findFreePort ( interface = '<STR_LIT:127.0.0.1>' , family = socket . AF_INET , <EOL> type = socket . SOCK_STREAM ) : <EOL> """<STR_LIT>""" <EOL> addr = socket . getaddrinfo ( interface , <NUM_LIT:0> ) [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] <EOL> probe = socket . socket ( family , type ) <EOL> try : <EOL> probe . bind ( addr ) <EOL> return probe . getsockname ( ) <EOL> finally : <EOL> probe . close ( ) <EOL> class ConnectableProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> disconnectReason = None <EOL> def _setAttributes ( self , reactor , done ) : <EOL> """<STR_LIT>""" <EOL> self . reactor = reactor <EOL> self . _done = done <EOL> def connectionLost ( self , reason ) : <EOL> self . disconnectReason = reason <EOL> self . _done . callback ( None ) <EOL> del self . _done <EOL> class EndpointCreator : <EOL> """<STR_LIT>""" <EOL> def server ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def client ( self , reactor , serverAddress ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> class _SingleProtocolFactory ( ClientFactory ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , protocol ) : <EOL> self . _protocol = protocol <EOL> def buildProtocol ( self , addr ) : <EOL> return self . _protocol <EOL> def runProtocolsWithReactor ( reactorBuilder , serverProtocol , clientProtocol , <EOL> endpointCreator ) : <EOL> """<STR_LIT>""" <EOL> reactor = reactorBuilder . buildReactor ( ) <EOL> serverProtocol . _setAttributes ( reactor , Deferred ( ) ) <EOL> clientProtocol . _setAttributes ( reactor , Deferred ( ) ) <EOL> serverFactory = _SingleProtocolFactory ( serverProtocol ) <EOL> clientFactory = _SingleProtocolFactory ( clientProtocol ) <EOL> serverEndpoint = endpointCreator . server ( reactor ) <EOL> d = serverEndpoint . listen ( serverFactory ) <EOL> def gotPort ( p ) : <EOL> clientEndpoint = endpointCreator . client ( <EOL> reactor , p . getHost ( ) ) <EOL> return clientEndpoint . connect ( clientFactory ) <EOL> d . addCallback ( gotPort ) <EOL> def failed ( result ) : <EOL> log . err ( result , "<STR_LIT>" ) <EOL> disconnected = gatherResults ( [ serverProtocol . _done , clientProtocol . _done ] ) <EOL> d . addCallback ( lambda _ : disconnected ) <EOL> d . addErrback ( failed ) <EOL> d . addCallback ( lambda _ : needsRunningReactor ( reactor , reactor . stop ) ) <EOL> reactorBuilder . runReactor ( reactor ) <EOL> return reactor <EOL> def _getWriters ( reactor ) : <EOL> """<STR_LIT>""" <EOL> if IReactorFDSet . providedBy ( reactor ) : <EOL> return reactor . getWriters ( ) <EOL> elif '<STR_LIT>' in reactor . __class__ . __name__ : <EOL> return reactor . handles <EOL> else : <EOL> raise Exception ( "<STR_LIT>" % ( reactor , ) ) <EOL> class _AcceptOneClient ( ServerFactory ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reactor , result ) : <EOL> self . reactor = reactor <EOL> self . result = result <EOL> def buildProtocol ( self , addr ) : <EOL> protocol = ServerFactory . buildProtocol ( self , addr ) <EOL> self . reactor . callLater ( <NUM_LIT:0> , self . result . callback , protocol ) <EOL> return protocol <EOL> class _SimplePullProducer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , consumer ) : <EOL> self . consumer = consumer <EOL> def stopProducing ( self ) : <EOL> pass <EOL> def resumeProducing ( self ) : <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . consumer . write ( b'<STR_LIT:x>' ) <EOL> class Stop ( ClientFactory ) : <EOL> """<STR_LIT>""" <EOL> failReason = None <EOL> def __init__ ( self , reactor ) : <EOL> self . reactor = reactor <EOL> def clientConnectionFailed ( self , connector , reason ) : <EOL> self . failReason = reason <EOL> msg ( "<STR_LIT>" % ( reason . getErrorMessage ( ) , ) ) <EOL> self . reactor . stop ( ) <EOL> class ClosingLaterProtocol ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , onConnectionLost ) : <EOL> self . lostConnectionReason = None <EOL> self . onConnectionLost = onConnectionLost <EOL> def connectionMade ( self ) : <EOL> msg ( "<STR_LIT>" ) <EOL> def dataReceived ( self , bytes ) : <EOL> msg ( "<STR_LIT>" % ( bytes , ) ) <EOL> self . transport . loseConnection ( ) <EOL> def connectionLost ( self , reason ) : <EOL> msg ( "<STR_LIT>" ) <EOL> self . lostConnectionReason = reason <EOL> self . onConnectionLost . callback ( self ) <EOL> class ConnectionTestsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> endpoints = None <EOL> def test_logPrefix ( self ) : <EOL> """<STR_LIT>""" <EOL> class CustomLogPrefixProtocol ( ConnectableProtocol ) : <EOL> def __init__ ( self , prefix ) : <EOL> self . _prefix = prefix <EOL> self . system = None <EOL> def connectionMade ( self ) : <EOL> self . transport . write ( b"<STR_LIT:a>" ) <EOL> def logPrefix ( self ) : <EOL> return self . _prefix <EOL> def dataReceived ( self , bytes ) : <EOL> self . system = context . get ( ILogContext ) [ "<STR_LIT>" ] <EOL> self . transport . write ( b"<STR_LIT:b>" ) <EOL> if b"<STR_LIT:b>" in bytes : <EOL> self . transport . loseConnection ( ) <EOL> client = CustomLogPrefixProtocol ( "<STR_LIT>" ) <EOL> server = CustomLogPrefixProtocol ( "<STR_LIT>" ) <EOL> runProtocolsWithReactor ( self , server , client , self . endpoints ) <EOL> self . assertIn ( "<STR_LIT>" , client . system ) <EOL> self . assertIn ( "<STR_LIT>" , server . system ) <EOL> def test_writeAfterDisconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> finished = [ ] <EOL> serverConnectionLostDeferred = Deferred ( ) <EOL> protocol = lambda : ClosingLaterProtocol ( serverConnectionLostDeferred ) <EOL> portDeferred = self . endpoints . server ( reactor ) . listen ( <EOL> ServerFactory . forProtocol ( protocol ) ) <EOL> def listening ( port ) : <EOL> msg ( "<STR_LIT>" % ( port . getHost ( ) , ) ) <EOL> endpoint = self . endpoints . client ( reactor , port . getHost ( ) ) <EOL> lostConnectionDeferred = Deferred ( ) <EOL> protocol = lambda : ClosingLaterProtocol ( lostConnectionDeferred ) <EOL> client = endpoint . connect ( ClientFactory . forProtocol ( protocol ) ) <EOL> def write ( proto ) : <EOL> msg ( "<STR_LIT>" % ( proto , ) ) <EOL> proto . transport . write ( b'<STR_LIT:x>' ) <EOL> client . addCallbacks ( write , lostConnectionDeferred . errback ) <EOL> def disconnected ( proto ) : <EOL> msg ( "<STR_LIT>" % ( proto , ) ) <EOL> proto . transport . write ( b"<STR_LIT>" ) <EOL> proto . transport . writeSequence ( [ b"<STR_LIT>" , b"<STR_LIT>" ] ) <EOL> finished . append ( True ) <EOL> lostConnectionDeferred . addCallback ( disconnected ) <EOL> serverConnectionLostDeferred . addCallback ( disconnected ) <EOL> return gatherResults ( [ lostConnectionDeferred , <EOL> serverConnectionLostDeferred ] ) <EOL> def onListen ( ) : <EOL> portDeferred . addCallback ( listening ) <EOL> portDeferred . addErrback ( err ) <EOL> portDeferred . addCallback ( lambda ignored : reactor . stop ( ) ) <EOL> needsRunningReactor ( reactor , onListen ) <EOL> self . runReactor ( reactor ) <EOL> self . assertEqual ( finished , [ True , True ] ) <EOL> def test_protocolGarbageAfterLostConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> lostConnectionDeferred = Deferred ( ) <EOL> clientProtocol = ClosingLaterProtocol ( lostConnectionDeferred ) <EOL> clientRef = ref ( clientProtocol ) <EOL> reactor = self . buildReactor ( ) <EOL> portDeferred = self . endpoints . server ( reactor ) . listen ( <EOL> ServerFactory . forProtocol ( Protocol ) ) <EOL> def listening ( port ) : <EOL> msg ( "<STR_LIT>" % ( port . getHost ( ) , ) ) <EOL> endpoint = self . endpoints . client ( reactor , port . getHost ( ) ) <EOL> client = endpoint . connect ( <EOL> ClientFactory . forProtocol ( lambda : clientProtocol ) ) <EOL> def disconnect ( proto ) : <EOL> msg ( "<STR_LIT>" % ( proto , ) ) <EOL> proto . transport . loseConnection ( ) <EOL> client . addCallback ( disconnect ) <EOL> client . addErrback ( lostConnectionDeferred . errback ) <EOL> return lostConnectionDeferred <EOL> def onListening ( ) : <EOL> portDeferred . addCallback ( listening ) <EOL> portDeferred . addErrback ( err ) <EOL> portDeferred . addBoth ( lambda ignored : reactor . stop ( ) ) <EOL> needsRunningReactor ( reactor , onListening ) <EOL> self . runReactor ( reactor ) <EOL> clientProtocol = None <EOL> collect ( ) <EOL> self . assertIs ( None , clientRef ( ) ) <EOL> class LogObserverMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def observe ( self ) : <EOL> loggedMessages = [ ] <EOL> log . addObserver ( loggedMessages . append ) <EOL> self . addCleanup ( log . removeObserver , loggedMessages . append ) <EOL> return loggedMessages <EOL> class BrokenContextFactory ( object ) : <EOL> """<STR_LIT>""" <EOL> message = "<STR_LIT>" <EOL> def getContext ( self ) : <EOL> raise ValueError ( self . message ) <EOL> class StreamClientTestsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> connector = self . connect ( reactor , ClientFactory ( ) ) <EOL> self . assertTrue ( verifyObject ( IConnector , connector ) ) <EOL> def test_clientConnectionFailedStopsReactor ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> needsRunningReactor ( <EOL> reactor , lambda : self . connect ( reactor , Stop ( reactor ) ) ) <EOL> self . runReactor ( reactor ) <EOL> def test_connectEvent ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> self . listen ( reactor , ServerFactory . forProtocol ( Protocol ) ) <EOL> connected = [ ] <EOL> class CheckConnection ( Protocol ) : <EOL> def connectionMade ( self ) : <EOL> connected . append ( self ) <EOL> reactor . stop ( ) <EOL> clientFactory = Stop ( reactor ) <EOL> clientFactory . protocol = CheckConnection <EOL> needsRunningReactor ( <EOL> reactor , lambda : self . connect ( reactor , clientFactory ) ) <EOL> reactor . run ( ) <EOL> self . assertTrue ( connected ) <EOL> def test_unregisterProducerAfterDisconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> self . listen ( reactor , ServerFactory . forProtocol ( ClosingProtocol ) ) <EOL> finished = Deferred ( ) <EOL> finished . addErrback ( log . err ) <EOL> finished . addCallback ( lambda ign : reactor . stop ( ) ) <EOL> writing = [ ] <EOL> class ClientProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> def connectionMade ( self ) : <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . transport . registerProducer ( <EOL> _SimplePullProducer ( self . transport ) , False ) <EOL> self . transport . loseConnection ( ) <EOL> def connectionLost ( self , reason ) : <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . unregister ( ) <EOL> writing . append ( self . transport in _getWriters ( reactor ) ) <EOL> finished . callback ( None ) <EOL> def unregister ( self ) : <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . transport . unregisterProducer ( ) <EOL> clientFactory = ClientFactory ( ) <EOL> clientFactory . protocol = ClientProtocol <EOL> self . connect ( reactor , clientFactory ) <EOL> self . runReactor ( reactor ) <EOL> self . assertFalse ( writing [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> def test_disconnectWhileProducing ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> skippedReactors = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> reactorClassName = reactor . __class__ . __name__ <EOL> if reactorClassName in skippedReactors and platform . isWindows ( ) : <EOL> raise SkipTest ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class Producer : <EOL> def resumeProducing ( self ) : <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . listen ( reactor , ServerFactory . forProtocol ( Protocol ) ) <EOL> finished = Deferred ( ) <EOL> finished . addErrback ( log . err ) <EOL> finished . addCallback ( lambda ign : reactor . stop ( ) ) <EOL> class ClientProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> def connectionMade ( self ) : <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . transport . registerProducer ( Producer ( ) , False ) <EOL> self . transport . loseConnection ( ) <EOL> reactor . callLater ( <NUM_LIT:0> , reactor . callLater , <NUM_LIT:0> , self . unregister ) <EOL> def unregister ( self ) : <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . transport . unregisterProducer ( ) <EOL> reactor . callLater ( <EOL> <NUM_LIT:1.0> , finished . errback , <EOL> Failure ( Exception ( "<STR_LIT>" ) ) ) <EOL> def connectionLost ( self , reason ) : <EOL> log . msg ( "<STR_LIT>" ) <EOL> finished . callback ( None ) <EOL> clientFactory = ClientFactory ( ) <EOL> clientFactory . protocol = ClientProtocol <EOL> self . connect ( reactor , clientFactory ) <EOL> self . runReactor ( reactor ) </s>
