<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import <EOL> __metaclass__ = type <EOL> import errno <EOL> import socket <EOL> from functools import wraps <EOL> from zope . interface import implementer <EOL> from zope . interface . verify import verifyClass <EOL> from twisted . python . runtime import platform <EOL> from twisted . python . failure import Failure <EOL> from twisted . python import log <EOL> from twisted . trial . unittest import SkipTest , TestCase <EOL> from twisted . internet . error import ( <EOL> ConnectionLost , UserError , ConnectionRefusedError , ConnectionDone , <EOL> ConnectionAborted , DNSLookupError , NoProtocol ) <EOL> from twisted . internet . test . connectionmixins import ( <EOL> LogObserverMixin , ConnectionTestsMixin , StreamClientTestsMixin , <EOL> findFreePort , ConnectableProtocol , EndpointCreator , <EOL> runProtocolsWithReactor , Stop , BrokenContextFactory ) <EOL> from twisted . internet . test . reactormixins import ( <EOL> ReactorBuilder , needsRunningReactor , stopOnError ) <EOL> from twisted . internet . interfaces import ( <EOL> ILoggingContext , IConnector , IReactorFDSet , IReactorSocket , IReactorTCP , <EOL> IResolverSimple , ITLSTransport ) <EOL> from twisted . internet . address import IPv4Address , IPv6Address <EOL> from twisted . internet . defer import ( <EOL> Deferred , DeferredList , maybeDeferred , gatherResults , succeed , fail ) <EOL> from twisted . internet . endpoints import TCP4ServerEndpoint , TCP4ClientEndpoint <EOL> from twisted . internet . protocol import ServerFactory , ClientFactory , Protocol <EOL> from twisted . internet . interfaces import ( <EOL> IPushProducer , IPullProducer , IHalfCloseableProtocol ) <EOL> from twisted . internet . tcp import Connection , Server , _resolveIPv6 <EOL> from twisted . internet . test . test_core import ObjectModelIntegrationMixin <EOL> from twisted . test . test_tcp import MyClientFactory , MyServerFactory <EOL> from twisted . test . test_tcp import ClosingFactory , ClientStartStopFactory <EOL> try : <EOL> from OpenSSL import SSL <EOL> except ImportError : <EOL> useSSL = False <EOL> else : <EOL> from twisted . internet . ssl import ClientContextFactory <EOL> useSSL = True <EOL> try : <EOL> socket . socket ( socket . AF_INET6 , socket . SOCK_STREAM ) . close ( ) <EOL> except socket . error as e : <EOL> ipv6Skip = str ( e ) <EOL> else : <EOL> ipv6Skip = None <EOL> if platform . isWindows ( ) : <EOL> from twisted . internet . test import _win32ifaces <EOL> getLinkLocalIPv6Addresses = _win32ifaces . win32GetLinkLocalIPv6Addresses <EOL> else : <EOL> try : <EOL> from twisted . internet . test import _posixifaces <EOL> except ImportError : <EOL> getLinkLocalIPv6Addresses = lambda : [ ] <EOL> else : <EOL> getLinkLocalIPv6Addresses = _posixifaces . posixGetLinkLocalIPv6Addresses <EOL> def getLinkLocalIPv6Address ( ) : <EOL> """<STR_LIT>""" <EOL> addresses = getLinkLocalIPv6Addresses ( ) <EOL> if addresses : <EOL> return addresses [ <NUM_LIT:0> ] <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> def connect ( client , destination ) : <EOL> """<STR_LIT>""" <EOL> ( host , port ) = destination <EOL> if '<STR_LIT:%>' in host or '<STR_LIT::>' in host : <EOL> address = socket . getaddrinfo ( host , port ) [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] <EOL> else : <EOL> address = ( host , port ) <EOL> client . connect ( address ) <EOL> class FakeSocket ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , data ) : <EOL> self . data = data <EOL> self . sendBuffer = [ ] <EOL> def setblocking ( self , blocking ) : <EOL> self . blocking = blocking <EOL> def recv ( self , size ) : <EOL> return self . data <EOL> def send ( self , bytes ) : <EOL> """<STR_LIT>""" <EOL> self . sendBuffer . append ( bytes ) <EOL> return len ( bytes ) <EOL> def shutdown ( self , how ) : <EOL> """<STR_LIT>""" <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> def setsockopt ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> def fileno ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:1> <EOL> class TestFakeSocket ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_blocking ( self ) : <EOL> skt = FakeSocket ( b"<STR_LIT>" ) <EOL> skt . setblocking ( <NUM_LIT:0> ) <EOL> self . assertEqual ( skt . blocking , <NUM_LIT:0> ) <EOL> def test_recv ( self ) : <EOL> skt = FakeSocket ( b"<STR_LIT>" ) <EOL> self . assertEqual ( skt . recv ( <NUM_LIT:10> ) , b"<STR_LIT>" ) <EOL> def test_send ( self ) : <EOL> """<STR_LIT>""" <EOL> skt = FakeSocket ( b"<STR_LIT>" ) <EOL> count = skt . send ( b"<STR_LIT:foo>" ) <EOL> self . assertEqual ( count , <NUM_LIT:3> ) <EOL> self . assertEqual ( skt . sendBuffer , [ b"<STR_LIT:foo>" ] ) <EOL> class FakeProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> def dataReceived ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return ( ) <EOL> @ implementer ( IReactorFDSet ) <EOL> class _FakeFDSetReactor ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _readers = set ( ) <EOL> self . _writers = set ( ) <EOL> def addReader ( self , reader ) : <EOL> self . _readers . add ( reader ) <EOL> def removeReader ( self , reader ) : <EOL> if reader in self . _readers : <EOL> self . _readers . remove ( reader ) <EOL> def addWriter ( self , writer ) : <EOL> self . _writers . add ( writer ) <EOL> def removeWriter ( self , writer ) : <EOL> if writer in self . _writers : <EOL> self . _writers . remove ( writer ) <EOL> def removeAll ( self ) : <EOL> result = self . getReaders ( ) + self . getWriters ( ) <EOL> self . __init__ ( ) <EOL> return result <EOL> def getReaders ( self ) : <EOL> return list ( self . _readers ) <EOL> def getWriters ( self ) : <EOL> return list ( self . _writers ) <EOL> verifyClass ( IReactorFDSet , _FakeFDSetReactor ) <EOL> class TCPServerTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . reactor = _FakeFDSetReactor ( ) <EOL> class FakePort ( object ) : <EOL> _realPortNumber = <NUM_LIT:3> <EOL> self . skt = FakeSocket ( b"<STR_LIT>" ) <EOL> self . protocol = Protocol ( ) <EOL> self . server = Server ( <EOL> self . skt , self . protocol , ( "<STR_LIT>" , <NUM_LIT:0> ) , FakePort ( ) , None , self . reactor ) <EOL> def test_writeAfterDisconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> self . server . connectionLost ( <EOL> Failure ( Exception ( "<STR_LIT>" ) ) ) <EOL> self . server . write ( b"<STR_LIT>" ) <EOL> self . assertEqual ( self . skt . sendBuffer , [ ] ) <EOL> def test_writeAfterDisconnectAfterTLS ( self ) : <EOL> """<STR_LIT>""" <EOL> self . server . TLS = True <EOL> self . test_writeAfterDisconnect ( ) <EOL> def test_writeSequenceAfterDisconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> self . server . connectionLost ( <EOL> Failure ( Exception ( "<STR_LIT>" ) ) ) <EOL> self . server . writeSequence ( [ b"<STR_LIT>" ] ) <EOL> self . assertEqual ( self . skt . sendBuffer , [ ] ) <EOL> def test_writeSequenceAfterDisconnectAfterTLS ( self ) : <EOL> """<STR_LIT>""" <EOL> self . server . TLS = True <EOL> self . test_writeSequenceAfterDisconnect ( ) <EOL> class TCPConnectionTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_doReadWarningIsRaised ( self ) : <EOL> """<STR_LIT>""" <EOL> skt = FakeSocket ( b"<STR_LIT>" ) <EOL> protocol = FakeProtocol ( ) <EOL> conn = Connection ( skt , protocol ) <EOL> conn . doRead ( ) <EOL> warnings = self . flushWarnings ( [ FakeProtocol . dataReceived ] ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> warnings [ <NUM_LIT:0> ] [ "<STR_LIT:message>" ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> def test_noTLSBeforeStartTLS ( self ) : <EOL> """<STR_LIT>""" <EOL> skt = FakeSocket ( b"<STR_LIT>" ) <EOL> protocol = FakeProtocol ( ) <EOL> conn = Connection ( skt , protocol ) <EOL> self . assertFalse ( conn . TLS ) <EOL> def test_tlsAfterStartTLS ( self ) : <EOL> """<STR_LIT>""" <EOL> skt = FakeSocket ( b"<STR_LIT>" ) <EOL> protocol = FakeProtocol ( ) <EOL> conn = Connection ( skt , protocol , reactor = _FakeFDSetReactor ( ) ) <EOL> conn . _tlsClientDefault = True <EOL> conn . startTLS ( ClientContextFactory ( ) , True ) <EOL> self . assertTrue ( conn . TLS ) <EOL> if not useSSL : <EOL> test_tlsAfterStartTLS . skip = "<STR_LIT>" <EOL> class TCPCreator ( EndpointCreator ) : <EOL> """<STR_LIT>""" <EOL> interface = "<STR_LIT:127.0.0.1>" <EOL> def server ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return TCP4ServerEndpoint ( reactor , <NUM_LIT:0> , interface = self . interface ) <EOL> def client ( self , reactor , serverAddress ) : <EOL> """<STR_LIT>""" <EOL> return TCP4ClientEndpoint ( reactor , self . interface , serverAddress . port ) <EOL> class TCP6Creator ( TCPCreator ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . interface = getLinkLocalIPv6Address ( ) <EOL> @ implementer ( IResolverSimple ) <EOL> class FakeResolver ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , names ) : <EOL> self . names = names <EOL> def getHostByName ( self , name , timeout ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return succeed ( self . names [ name ] ) <EOL> except KeyError : <EOL> return fail ( DNSLookupError ( "<STR_LIT>" + name ) ) <EOL> class TCPClientTestsBase ( ReactorBuilder , ConnectionTestsMixin , <EOL> StreamClientTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = ( IReactorTCP , ) <EOL> _port = None <EOL> @ property <EOL> def port ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _port is not None : <EOL> return self . _port . getHost ( ) . port <EOL> return findFreePort ( self . interface , self . family ) [ <NUM_LIT:1> ] <EOL> @ property <EOL> def interface ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . endpoints . interface <EOL> def listen ( self , reactor , factory ) : <EOL> """<STR_LIT>""" <EOL> self . _port = reactor . listenTCP ( <NUM_LIT:0> , factory , interface = self . interface ) <EOL> return self . _port <EOL> def connect ( self , reactor , factory ) : <EOL> """<STR_LIT>""" <EOL> return reactor . connectTCP ( self . interface , self . port , factory ) <EOL> def test_buildProtocolReturnsNone ( self ) : <EOL> """<STR_LIT>""" <EOL> connectionLost = Deferred ( ) <EOL> reactor = self . buildReactor ( ) <EOL> serverFactory = MyServerFactory ( ) <EOL> serverFactory . protocolConnectionLost = connectionLost <EOL> stopOnError ( self , reactor ) <EOL> class NoneFactory ( ServerFactory ) : <EOL> def buildProtocol ( self , address ) : <EOL> return None <EOL> listening = self . endpoints . server ( reactor ) . listen ( serverFactory ) <EOL> def listened ( port ) : <EOL> clientFactory = NoneFactory ( ) <EOL> endpoint = self . endpoints . client ( reactor , port . getHost ( ) ) <EOL> return endpoint . connect ( clientFactory ) <EOL> connecting = listening . addCallback ( listened ) <EOL> def connectSucceeded ( protocol ) : <EOL> self . fail ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( protocol , ) ) <EOL> def connectFailed ( reason ) : <EOL> reason . trap ( NoProtocol ) <EOL> connecting . addCallbacks ( connectSucceeded , connectFailed ) <EOL> def connected ( ignored ) : <EOL> return connectionLost <EOL> disconnecting = connecting . addCallback ( connected ) <EOL> disconnecting . addErrback ( log . err ) <EOL> def disconnected ( ignored ) : <EOL> reactor . stop ( ) <EOL> disconnecting . addCallback ( disconnected ) <EOL> self . runReactor ( reactor ) <EOL> def test_addresses ( self ) : <EOL> """<STR_LIT>""" <EOL> host , port = findFreePort ( self . interface , self . family ) [ : <NUM_LIT:2> ] <EOL> reactor = self . buildReactor ( ) <EOL> fakeDomain = self . fakeDomainName <EOL> reactor . installResolver ( FakeResolver ( { fakeDomain : self . interface } ) ) <EOL> server = reactor . listenTCP ( <EOL> <NUM_LIT:0> , ServerFactory . forProtocol ( Protocol ) , interface = host ) <EOL> serverAddress = server . getHost ( ) <EOL> transportData = { '<STR_LIT:host>' : None , '<STR_LIT>' : None , '<STR_LIT>' : None } <EOL> class CheckAddress ( Protocol ) : <EOL> def makeConnection ( self , transport ) : <EOL> transportData [ '<STR_LIT:host>' ] = transport . getHost ( ) <EOL> transportData [ '<STR_LIT>' ] = transport . getPeer ( ) <EOL> transportData [ '<STR_LIT>' ] = transport <EOL> reactor . stop ( ) <EOL> clientFactory = Stop ( reactor ) <EOL> clientFactory . protocol = CheckAddress <EOL> def connectMe ( ) : <EOL> reactor . connectTCP ( <EOL> fakeDomain , server . getHost ( ) . port , clientFactory , <EOL> bindAddress = ( self . interface , port ) ) <EOL> needsRunningReactor ( reactor , connectMe ) <EOL> self . runReactor ( reactor ) <EOL> if clientFactory . failReason : <EOL> self . fail ( clientFactory . failReason . getTraceback ( ) ) <EOL> transportRepr = "<STR_LIT>" % ( <EOL> transportData [ '<STR_LIT>' ] . __class__ , <EOL> transportData [ '<STR_LIT>' ] . addr , <EOL> id ( transportData [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <EOL> transportData [ '<STR_LIT:host>' ] , <EOL> self . addressClass ( '<STR_LIT>' , self . interface , port ) ) <EOL> self . assertEqual ( <EOL> transportData [ '<STR_LIT>' ] , <EOL> self . addressClass ( '<STR_LIT>' , self . interface , serverAddress . port ) ) <EOL> self . assertEqual ( <EOL> repr ( transportData [ '<STR_LIT>' ] ) , transportRepr ) <EOL> def test_badContext ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> brokenFactory = BrokenContextFactory ( ) <EOL> results = [ ] <EOL> serverFactory = ServerFactory . forProtocol ( Protocol ) <EOL> port = reactor . listenTCP ( <NUM_LIT:0> , serverFactory , interface = self . interface ) <EOL> endpoint = self . endpoints . client ( reactor , port . getHost ( ) ) <EOL> clientFactory = ClientFactory ( ) <EOL> clientFactory . protocol = Protocol <EOL> connectDeferred = endpoint . connect ( clientFactory ) <EOL> def connected ( protocol ) : <EOL> if not ITLSTransport . providedBy ( protocol . transport ) : <EOL> results . append ( "<STR_LIT>" ) <EOL> else : <EOL> results . append ( self . assertRaises ( ValueError , <EOL> protocol . transport . startTLS , <EOL> brokenFactory ) ) <EOL> def connectFailed ( failure ) : <EOL> results . append ( failure ) <EOL> def whenRun ( ) : <EOL> connectDeferred . addCallback ( connected ) <EOL> connectDeferred . addErrback ( connectFailed ) <EOL> connectDeferred . addBoth ( lambda ign : reactor . stop ( ) ) <EOL> needsRunningReactor ( reactor , whenRun ) <EOL> self . runReactor ( reactor ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> , <EOL> "<STR_LIT>" % ( results , ) ) <EOL> if isinstance ( results [ <NUM_LIT:0> ] , Failure ) : <EOL> results [ <NUM_LIT:0> ] . raiseException ( ) <EOL> if results [ <NUM_LIT:0> ] == "<STR_LIT>" : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> self . assertEqual ( BrokenContextFactory . message , str ( results [ <NUM_LIT:0> ] ) ) <EOL> class TCP4ClientTestsBuilder ( TCPClientTestsBase ) : <EOL> """<STR_LIT>""" <EOL> fakeDomainName = '<STR_LIT>' <EOL> family = socket . AF_INET <EOL> addressClass = IPv4Address <EOL> endpoints = TCPCreator ( ) <EOL> class TCP6ClientTestsBuilder ( TCPClientTestsBase ) : <EOL> """<STR_LIT>""" <EOL> if ipv6Skip : <EOL> skip = ipv6Skip <EOL> family = socket . AF_INET6 <EOL> addressClass = IPv6Address <EOL> def setUp ( self ) : <EOL> self . endpoints = TCP6Creator ( ) <EOL> self . fakeDomainName = self . endpoints . interface <EOL> class TCPConnectorTestsBuilder ( ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = ( IReactorTCP , ) <EOL> def test_connectorIdentity ( self ) : <EOL> """<STR_LIT>""" <EOL> serverFactory = ClosingFactory ( ) <EOL> reactor = self . buildReactor ( ) <EOL> tcpPort = reactor . listenTCP ( <NUM_LIT:0> , serverFactory , interface = self . interface ) <EOL> serverFactory . port = tcpPort <EOL> portNumber = tcpPort . getHost ( ) . port <EOL> seenConnectors = [ ] <EOL> seenFailures = [ ] <EOL> clientFactory = ClientStartStopFactory ( ) <EOL> clientFactory . clientConnectionLost = ( <EOL> lambda connector , reason : ( seenConnectors . append ( connector ) , <EOL> seenFailures . append ( reason ) ) ) <EOL> clientFactory . startedConnecting = seenConnectors . append <EOL> connector = reactor . connectTCP ( self . interface , portNumber , <EOL> clientFactory ) <EOL> self . assertTrue ( IConnector . providedBy ( connector ) ) <EOL> dest = connector . getDestination ( ) <EOL> self . assertEqual ( dest . type , "<STR_LIT>" ) <EOL> self . assertEqual ( dest . host , self . interface ) <EOL> self . assertEqual ( dest . port , portNumber ) <EOL> clientFactory . whenStopped . addBoth ( lambda _ : reactor . stop ( ) ) <EOL> self . runReactor ( reactor ) <EOL> seenFailures [ <NUM_LIT:0> ] . trap ( ConnectionDone ) <EOL> self . assertEqual ( seenConnectors , [ connector , connector ] ) <EOL> def test_userFail ( self ) : <EOL> """<STR_LIT>""" <EOL> serverFactory = MyServerFactory ( ) <EOL> reactor = self . buildReactor ( ) <EOL> tcpPort = reactor . listenTCP ( <NUM_LIT:0> , serverFactory , interface = self . interface ) <EOL> portNumber = tcpPort . getHost ( ) . port <EOL> fatalErrors = [ ] <EOL> def startedConnecting ( connector ) : <EOL> try : <EOL> connector . stopConnecting ( ) <EOL> except Exception : <EOL> fatalErrors . append ( Failure ( ) ) <EOL> reactor . stop ( ) <EOL> clientFactory = ClientStartStopFactory ( ) <EOL> clientFactory . startedConnecting = startedConnecting <EOL> clientFactory . whenStopped . addBoth ( lambda _ : reactor . stop ( ) ) <EOL> reactor . callWhenRunning ( lambda : reactor . connectTCP ( self . interface , <EOL> portNumber , <EOL> clientFactory ) ) <EOL> self . runReactor ( reactor ) <EOL> if fatalErrors : <EOL> self . fail ( fatalErrors [ <NUM_LIT:0> ] . getTraceback ( ) ) <EOL> clientFactory . reason . trap ( UserError ) <EOL> self . assertEqual ( clientFactory . failed , <NUM_LIT:1> ) <EOL> def test_reconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> serverFactory = ClosingFactory ( ) <EOL> reactor = self . buildReactor ( ) <EOL> tcpPort = reactor . listenTCP ( <NUM_LIT:0> , serverFactory , interface = self . interface ) <EOL> serverFactory . port = tcpPort <EOL> portNumber = tcpPort . getHost ( ) . port <EOL> clientFactory = MyClientFactory ( ) <EOL> def clientConnectionLost ( connector , reason ) : <EOL> connector . connect ( ) <EOL> clientFactory . clientConnectionLost = clientConnectionLost <EOL> reactor . connectTCP ( self . interface , portNumber , clientFactory ) <EOL> protocolMadeAndClosed = [ ] <EOL> def reconnectFailed ( ignored ) : <EOL> p = clientFactory . protocol <EOL> protocolMadeAndClosed . append ( ( p . made , p . closed ) ) <EOL> reactor . stop ( ) <EOL> clientFactory . failDeferred . addCallback ( reconnectFailed ) <EOL> self . runReactor ( reactor ) <EOL> clientFactory . reason . trap ( ConnectionRefusedError ) <EOL> self . assertEqual ( protocolMadeAndClosed , [ ( <NUM_LIT:1> , <NUM_LIT:1> ) ] ) <EOL> class TCP4ConnectorTestsBuilder ( TCPConnectorTestsBuilder ) : <EOL> interface = '<STR_LIT:127.0.0.1>' <EOL> family = socket . AF_INET <EOL> addressClass = IPv4Address <EOL> class TCP6ConnectorTestsBuilder ( TCPConnectorTestsBuilder ) : <EOL> family = socket . AF_INET6 <EOL> addressClass = IPv6Address <EOL> if ipv6Skip : <EOL> skip = ipv6Skip <EOL> def setUp ( self ) : <EOL> self . interface = getLinkLocalIPv6Address ( ) <EOL> def createTestSocket ( test , addressFamily , socketType ) : <EOL> """<STR_LIT>""" <EOL> skt = socket . socket ( addressFamily , socketType ) <EOL> test . addCleanup ( skt . close ) <EOL> return skt <EOL> class StreamTransportTestsMixin ( LogObserverMixin ) : <EOL> """<STR_LIT>""" <EOL> def test_startedListeningLogMessage ( self ) : <EOL> """<STR_LIT>""" <EOL> loggedMessages = self . observe ( ) <EOL> reactor = self . buildReactor ( ) <EOL> @ implementer ( ILoggingContext ) <EOL> class SomeFactory ( ServerFactory ) : <EOL> def logPrefix ( self ) : <EOL> return "<STR_LIT>" <EOL> factory = SomeFactory ( ) <EOL> p = self . getListeningPort ( reactor , factory ) <EOL> expectedMessage = self . getExpectedStartListeningLogMessage ( <EOL> p , "<STR_LIT>" ) <EOL> self . assertEqual ( ( expectedMessage , ) , loggedMessages [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] ) <EOL> def test_connectionLostLogMsg ( self ) : <EOL> """<STR_LIT>""" <EOL> loggedMessages = [ ] <EOL> def logConnectionLostMsg ( eventDict ) : <EOL> loggedMessages . append ( log . textFromEventDict ( eventDict ) ) <EOL> reactor = self . buildReactor ( ) <EOL> p = self . getListeningPort ( reactor , ServerFactory ( ) ) <EOL> expectedMessage = self . getExpectedConnectionLostLogMsg ( p ) <EOL> log . addObserver ( logConnectionLostMsg ) <EOL> def stopReactor ( ignored ) : <EOL> log . removeObserver ( logConnectionLostMsg ) <EOL> reactor . stop ( ) <EOL> def doStopListening ( ) : <EOL> log . addObserver ( logConnectionLostMsg ) <EOL> maybeDeferred ( p . stopListening ) . addCallback ( stopReactor ) <EOL> reactor . callWhenRunning ( doStopListening ) <EOL> reactor . run ( ) <EOL> self . assertIn ( expectedMessage , loggedMessages ) <EOL> def test_allNewStyle ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> port = self . getListeningPort ( reactor , ServerFactory ( ) ) <EOL> self . assertFullyNewStyle ( port ) <EOL> class ListenTCPMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def getListeningPort ( self , reactor , factory , port = <NUM_LIT:0> , interface = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return reactor . listenTCP ( port , factory , interface = interface ) <EOL> class SocketTCPMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def getListeningPort ( self , reactor , factory , port = <NUM_LIT:0> , interface = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if IReactorSocket . providedBy ( reactor ) : <EOL> if '<STR_LIT::>' in interface : <EOL> domain = socket . AF_INET6 <EOL> address = socket . getaddrinfo ( interface , port ) [ <NUM_LIT:0> ] [ <NUM_LIT:4> ] <EOL> else : <EOL> domain = socket . AF_INET <EOL> address = ( interface , port ) <EOL> portSock = socket . socket ( domain ) <EOL> portSock . bind ( address ) <EOL> portSock . listen ( <NUM_LIT:3> ) <EOL> portSock . setblocking ( False ) <EOL> try : <EOL> return reactor . adoptStreamPort ( <EOL> portSock . fileno ( ) , portSock . family , factory ) <EOL> finally : <EOL> portSock . fileno ( ) <EOL> portSock . close ( ) <EOL> else : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> class TCPPortTestsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = ( IReactorTCP , ) <EOL> def getExpectedStartListeningLogMessage ( self , port , factory ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> factory , port . getHost ( ) . port ) <EOL> def getExpectedConnectionLostLogMsg ( self , port ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( port . getHost ( ) . port , ) <EOL> def test_portGetHostOnIPv4 ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> port = self . getListeningPort ( reactor , ServerFactory ( ) ) <EOL> address = port . getHost ( ) <EOL> self . assertIsInstance ( address , IPv4Address ) <EOL> def test_portGetHostOnIPv6 ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> host , portNumber = findFreePort ( <EOL> family = socket . AF_INET6 , interface = '<STR_LIT>' ) [ : <NUM_LIT:2> ] <EOL> port = self . getListeningPort ( <EOL> reactor , ServerFactory ( ) , portNumber , host ) <EOL> address = port . getHost ( ) <EOL> self . assertIsInstance ( address , IPv6Address ) <EOL> self . assertEqual ( '<STR_LIT>' , address . host ) <EOL> self . assertEqual ( portNumber , address . port ) <EOL> if ipv6Skip : <EOL> test_portGetHostOnIPv6 . skip = ipv6Skip <EOL> def test_portGetHostOnIPv6ScopeID ( self ) : <EOL> """<STR_LIT>""" <EOL> linkLocal = getLinkLocalIPv6Address ( ) <EOL> reactor = self . buildReactor ( ) <EOL> port = self . getListeningPort ( reactor , ServerFactory ( ) , <NUM_LIT:0> , linkLocal ) <EOL> address = port . getHost ( ) <EOL> self . assertIsInstance ( address , IPv6Address ) <EOL> self . assertEqual ( linkLocal , address . host ) <EOL> if ipv6Skip : <EOL> test_portGetHostOnIPv6ScopeID . skip = ipv6Skip <EOL> def _buildProtocolAddressTest ( self , client , interface ) : <EOL> """<STR_LIT>""" <EOL> class ObserveAddress ( ServerFactory ) : <EOL> def buildProtocol ( self , address ) : <EOL> reactor . stop ( ) <EOL> self . observedAddress = address <EOL> return Protocol ( ) <EOL> factory = ObserveAddress ( ) <EOL> reactor = self . buildReactor ( ) <EOL> port = self . getListeningPort ( reactor , factory , <NUM_LIT:0> , interface ) <EOL> client . setblocking ( False ) <EOL> try : <EOL> connect ( client , ( port . getHost ( ) . host , port . getHost ( ) . port ) ) <EOL> except socket . error as e : <EOL> errnum , message = e . args <EOL> self . assertIn ( errnum , ( errno . EINPROGRESS , errno . EWOULDBLOCK ) ) <EOL> self . runReactor ( reactor ) <EOL> return factory . observedAddress <EOL> def test_buildProtocolIPv4Address ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = '<STR_LIT:127.0.0.1>' <EOL> client = createTestSocket ( self , socket . AF_INET , socket . SOCK_STREAM ) <EOL> observedAddress = self . _buildProtocolAddressTest ( client , interface ) <EOL> self . assertEqual ( <EOL> IPv4Address ( '<STR_LIT>' , * client . getsockname ( ) ) , observedAddress ) <EOL> def test_buildProtocolIPv6Address ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = '<STR_LIT>' <EOL> client = createTestSocket ( self , socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> observedAddress = self . _buildProtocolAddressTest ( client , interface ) <EOL> self . assertEqual ( <EOL> IPv6Address ( '<STR_LIT>' , * client . getsockname ( ) [ : <NUM_LIT:2> ] ) , observedAddress ) <EOL> if ipv6Skip : <EOL> test_buildProtocolIPv6Address . skip = ipv6Skip <EOL> def test_buildProtocolIPv6AddressScopeID ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = getLinkLocalIPv6Address ( ) <EOL> client = createTestSocket ( self , socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> observedAddress = self . _buildProtocolAddressTest ( client , interface ) <EOL> self . assertEqual ( <EOL> IPv6Address ( '<STR_LIT>' , * client . getsockname ( ) [ : <NUM_LIT:2> ] ) , observedAddress ) <EOL> if ipv6Skip : <EOL> test_buildProtocolIPv6AddressScopeID . skip = ipv6Skip <EOL> def _serverGetConnectionAddressTest ( self , client , interface , which ) : <EOL> """<STR_LIT>""" <EOL> class ObserveAddress ( Protocol ) : <EOL> def makeConnection ( self , transport ) : <EOL> reactor . stop ( ) <EOL> self . factory . address = getattr ( transport , which ) ( ) <EOL> reactor = self . buildReactor ( ) <EOL> factory = ServerFactory ( ) <EOL> factory . protocol = ObserveAddress <EOL> port = self . getListeningPort ( reactor , factory , <NUM_LIT:0> , interface ) <EOL> client . setblocking ( False ) <EOL> try : <EOL> connect ( client , ( port . getHost ( ) . host , port . getHost ( ) . port ) ) <EOL> except socket . error as e : <EOL> errnum , message = e . args <EOL> self . assertIn ( errnum , ( errno . EINPROGRESS , errno . EWOULDBLOCK ) ) <EOL> self . runReactor ( reactor ) <EOL> return factory . address <EOL> def test_serverGetHostOnIPv4 ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = '<STR_LIT:127.0.0.1>' <EOL> client = createTestSocket ( self , socket . AF_INET , socket . SOCK_STREAM ) <EOL> hostAddress = self . _serverGetConnectionAddressTest ( <EOL> client , interface , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> IPv4Address ( '<STR_LIT>' , * client . getpeername ( ) ) , hostAddress ) <EOL> def test_serverGetHostOnIPv6 ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = '<STR_LIT>' <EOL> client = createTestSocket ( self , socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> hostAddress = self . _serverGetConnectionAddressTest ( <EOL> client , interface , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> IPv6Address ( '<STR_LIT>' , * client . getpeername ( ) [ : <NUM_LIT:2> ] ) , hostAddress ) <EOL> if ipv6Skip : <EOL> test_serverGetHostOnIPv6 . skip = ipv6Skip <EOL> def test_serverGetHostOnIPv6ScopeID ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = getLinkLocalIPv6Address ( ) <EOL> client = createTestSocket ( self , socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> hostAddress = self . _serverGetConnectionAddressTest ( <EOL> client , interface , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> IPv6Address ( '<STR_LIT>' , * client . getpeername ( ) [ : <NUM_LIT:2> ] ) , hostAddress ) <EOL> if ipv6Skip : <EOL> test_serverGetHostOnIPv6ScopeID . skip = ipv6Skip <EOL> def test_serverGetPeerOnIPv4 ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = '<STR_LIT:127.0.0.1>' <EOL> client = createTestSocket ( self , socket . AF_INET , socket . SOCK_STREAM ) <EOL> peerAddress = self . _serverGetConnectionAddressTest ( <EOL> client , interface , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> IPv4Address ( '<STR_LIT>' , * client . getsockname ( ) ) , peerAddress ) <EOL> def test_serverGetPeerOnIPv6 ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = '<STR_LIT>' <EOL> client = createTestSocket ( self , socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> peerAddress = self . _serverGetConnectionAddressTest ( <EOL> client , interface , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> IPv6Address ( '<STR_LIT>' , * client . getsockname ( ) [ : <NUM_LIT:2> ] ) , peerAddress ) <EOL> if ipv6Skip : <EOL> test_serverGetPeerOnIPv6 . skip = ipv6Skip <EOL> def test_serverGetPeerOnIPv6ScopeID ( self ) : <EOL> """<STR_LIT>""" <EOL> interface = getLinkLocalIPv6Address ( ) <EOL> client = createTestSocket ( self , socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> peerAddress = self . _serverGetConnectionAddressTest ( <EOL> client , interface , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> IPv6Address ( '<STR_LIT>' , * client . getsockname ( ) [ : <NUM_LIT:2> ] ) , peerAddress ) <EOL> if ipv6Skip : <EOL> test_serverGetPeerOnIPv6ScopeID . skip = ipv6Skip <EOL> class TCPPortTestsBuilder ( ReactorBuilder , ListenTCPMixin , TCPPortTestsMixin , <EOL> ObjectModelIntegrationMixin , <EOL> StreamTransportTestsMixin ) : <EOL> pass <EOL> class TCPFDPortTestsBuilder ( ReactorBuilder , SocketTCPMixin , TCPPortTestsMixin , <EOL> ObjectModelIntegrationMixin , <EOL> StreamTransportTestsMixin ) : <EOL> pass <EOL> class StopStartReadingProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> def connectionMade ( self ) : <EOL> self . data = b'<STR_LIT>' <EOL> self . pauseResumeProducing ( <NUM_LIT:3> ) <EOL> def pauseResumeProducing ( self , counter ) : <EOL> """<STR_LIT>""" <EOL> self . transport . pauseProducing ( ) <EOL> self . transport . resumeProducing ( ) <EOL> if counter : <EOL> self . factory . reactor . callLater ( <NUM_LIT:0> , <EOL> self . pauseResumeProducing , counter - <NUM_LIT:1> ) <EOL> else : <EOL> self . factory . reactor . callLater ( <NUM_LIT:0> , <EOL> self . factory . ready . callback , self ) <EOL> def dataReceived ( self , data ) : <EOL> log . msg ( '<STR_LIT>' , len ( data ) ) <EOL> self . data += data <EOL> if len ( self . data ) == <NUM_LIT:4> * <NUM_LIT> : <EOL> self . factory . stop . callback ( self . data ) <EOL> def oneTransportTest ( testMethod ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( testMethod ) <EOL> def actualTestMethod ( builder ) : <EOL> other = ConnectableProtocol ( ) <EOL> class ServerProtocol ( ConnectableProtocol ) : <EOL> def connectionMade ( self ) : <EOL> try : <EOL> testMethod ( builder , self . reactor , self . transport ) <EOL> finally : <EOL> if self . transport is not None : <EOL> self . transport . loseConnection ( ) <EOL> if other . transport is not None : <EOL> other . transport . loseConnection ( ) <EOL> serverProtocol = ServerProtocol ( ) <EOL> runProtocolsWithReactor ( builder , serverProtocol , other , TCPCreator ( ) ) <EOL> return actualTestMethod <EOL> def assertReading ( testCase , reactor , transport ) : <EOL> """<STR_LIT>""" <EOL> if IReactorFDSet . providedBy ( reactor ) : <EOL> testCase . assertIn ( transport , reactor . getReaders ( ) ) <EOL> else : <EOL> testCase . assertIn ( transport , reactor . handles ) <EOL> testCase . assertTrue ( transport . reading ) <EOL> def assertNotReading ( testCase , reactor , transport ) : <EOL> """<STR_LIT>""" <EOL> if IReactorFDSet . providedBy ( reactor ) : <EOL> testCase . assertNotIn ( transport , reactor . getReaders ( ) ) <EOL> else : <EOL> testCase . assertFalse ( transport . reading ) <EOL> class TCPConnectionTestsBuilder ( ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = ( IReactorTCP , ) <EOL> def test_stopStartReading ( self ) : <EOL> """<STR_LIT>""" <EOL> sf = ServerFactory ( ) <EOL> reactor = sf . 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> sf . protocol = StopStartReadingProtocol <EOL> sf . ready = Deferred ( ) <EOL> sf . stop = Deferred ( ) <EOL> p = reactor . listenTCP ( <NUM_LIT:0> , sf ) <EOL> port = p . getHost ( ) . port <EOL> def proceed ( protos , port ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( protos [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( protos [ <NUM_LIT:1> ] ) <EOL> protos = protos [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , protos [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] <EOL> protos [ <NUM_LIT:0> ] . transport . write ( b'<STR_LIT:x>' * ( <NUM_LIT:2> * <NUM_LIT> ) + b'<STR_LIT:y>' * ( <NUM_LIT:2> * <NUM_LIT> ) ) <EOL> return ( sf . stop . addCallback ( cleanup , protos , port ) <EOL> . addCallback ( lambda ign : reactor . stop ( ) ) ) <EOL> def cleanup ( data , protos , port ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( data , b'<STR_LIT:x>' * ( <NUM_LIT:2> * <NUM_LIT> ) + b'<STR_LIT:y>' * ( <NUM_LIT:2> * <NUM_LIT> ) , <EOL> '<STR_LIT>' ) <EOL> return DeferredList ( [ <EOL> maybeDeferred ( protos [ <NUM_LIT:0> ] . transport . loseConnection ) , <EOL> maybeDeferred ( protos [ <NUM_LIT:1> ] . transport . loseConnection ) , <EOL> maybeDeferred ( port . stopListening ) ] ) <EOL> cc = TCP4ClientEndpoint ( reactor , '<STR_LIT:127.0.0.1>' , port ) <EOL> cf = ClientFactory ( ) <EOL> cf . protocol = Protocol <EOL> d = DeferredList ( [ cc . connect ( cf ) , sf . ready ] ) . addCallback ( proceed , p ) <EOL> d . addErrback ( log . err ) <EOL> self . runReactor ( reactor ) <EOL> @ oneTransportTest <EOL> def test_resumeProducing ( self , reactor , server ) : <EOL> """<STR_LIT>""" <EOL> server . pauseProducing ( ) <EOL> assertNotReading ( self , reactor , server ) <EOL> server . resumeProducing ( ) <EOL> assertReading ( self , reactor , server ) <EOL> @ oneTransportTest <EOL> def test_resumeProducingWhileDisconnecting ( self , reactor , server ) : <EOL> """<STR_LIT>""" <EOL> server . loseConnection ( ) <EOL> server . resumeProducing ( ) <EOL> assertNotReading ( self , reactor , server ) <EOL> @ oneTransportTest <EOL> def test_resumeProducingWhileDisconnected ( self , reactor , server ) : <EOL> """<STR_LIT>""" <EOL> server . connectionLost ( Failure ( Exception ( "<STR_LIT>" ) ) ) <EOL> assertNotReading ( self , reactor , server ) <EOL> server . resumeProducing ( ) <EOL> assertNotReading ( self , reactor , server ) <EOL> def test_connectionLostAfterPausedTransport ( self ) : <EOL> """<STR_LIT>""" <EOL> class Pauser ( ConnectableProtocol ) : <EOL> def __init__ ( self ) : <EOL> self . events = [ ] <EOL> def dataReceived ( self , bytes ) : <EOL> self . events . append ( "<STR_LIT>" ) <EOL> self . transport . pauseProducing ( ) <EOL> self . reactor . callLater ( <NUM_LIT:0> , self . resume ) <EOL> def resume ( self ) : <EOL> self . events . append ( "<STR_LIT>" ) <EOL> self . transport . resumeProducing ( ) <EOL> def connectionLost ( self , reason ) : <EOL> self . events . append ( "<STR_LIT>" ) <EOL> ConnectableProtocol . connectionLost ( self , reason ) <EOL> class Client ( ConnectableProtocol ) : <EOL> def connectionMade ( self ) : <EOL> self . transport . write ( b"<STR_LIT>" ) <EOL> self . transport . loseConnection ( ) <EOL> pauser = Pauser ( ) <EOL> runProtocolsWithReactor ( self , pauser , Client ( ) , TCPCreator ( ) ) <EOL> self . assertEqual ( pauser . events , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_doubleHalfClose ( self ) : <EOL> """<STR_LIT>""" <EOL> @ implementer ( IHalfCloseableProtocol ) <EOL> class ListenerProtocol ( ConnectableProtocol ) : <EOL> def readConnectionLost ( self ) : <EOL> self . transport . loseWriteConnection ( ) <EOL> def writeConnectionLost ( self ) : <EOL> self . transport . loseConnection ( ) <EOL> class Client ( ConnectableProtocol ) : <EOL> def connectionMade ( self ) : <EOL> self . transport . loseConnection ( ) <EOL> runProtocolsWithReactor ( <EOL> self , ListenerProtocol ( ) , Client ( ) , TCPCreator ( ) ) <EOL> class WriteSequenceTestsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = ( IReactorTCP , ) <EOL> def setWriteBufferSize ( self , transport , value ) : <EOL> """<STR_LIT>""" <EOL> if getattr ( transport , "<STR_LIT>" , None ) is not None : <EOL> transport . writeBufferSize = value <EOL> else : <EOL> transport . bufferSize = value <EOL> def test_writeSequeceWithoutWrite ( self ) : <EOL> """<STR_LIT>""" <EOL> def connected ( protocols ) : <EOL> client , server , port = protocols <EOL> def dataReceived ( data ) : <EOL> log . msg ( "<STR_LIT>" % data ) <EOL> self . assertEqual ( data , b"<STR_LIT>" ) <EOL> client . transport . loseConnection ( ) <EOL> server . dataReceived = dataReceived <EOL> client . transport . writeSequence ( [ b"<STR_LIT>" , b"<STR_LIT>" , b"<STR_LIT>" ] ) <EOL> reactor = self . buildReactor ( ) <EOL> d = self . getConnectedClientAndServer ( reactor , "<STR_LIT:127.0.0.1>" , <EOL> socket . AF_INET ) <EOL> d . addCallback ( connected ) <EOL> d . addErrback ( log . err ) <EOL> self . runReactor ( reactor ) <EOL> def test_writeSequenceWithUnicodeRaisesException ( self ) : <EOL> """<STR_LIT>""" <EOL> def connected ( protocols ) : <EOL> client , server , port = protocols <EOL> exc = self . assertRaises ( <EOL> TypeError , <EOL> server . transport . writeSequence , [ u"<STR_LIT>" ] ) <EOL> self . assertEqual ( str ( exc ) , "<STR_LIT>" ) <EOL> server . transport . loseConnection ( ) <EOL> reactor = self . buildReactor ( ) <EOL> d = self . getConnectedClientAndServer ( reactor , "<STR_LIT:127.0.0.1>" , <EOL> socket . AF_INET ) <EOL> d . addCallback ( connected ) <EOL> d . addErrback ( log . err ) <EOL> self . runReactor ( reactor ) <EOL> def test_streamingProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> @ implementer ( IPushProducer ) <EOL> class SaveActionProducer ( object ) : <EOL> client = None <EOL> server = None <EOL> def __init__ ( self ) : <EOL> self . actions = [ ] <EOL> def pauseProducing ( self ) : <EOL> self . actions . append ( "<STR_LIT>" ) <EOL> def resumeProducing ( self ) : <EOL> self . actions . append ( "<STR_LIT>" ) <EOL> self . client . transport . unregisterProducer ( ) <EOL> self . server . transport . loseConnection ( ) <EOL> def stopProducing ( self ) : <EOL> self . actions . append ( "<STR_LIT>" ) <EOL> producer = SaveActionProducer ( ) <EOL> def connected ( protocols ) : <EOL> client , server = protocols [ : <NUM_LIT:2> ] <EOL> producer . client = client <EOL> producer . server = server <EOL> client . transport . registerProducer ( producer , True ) <EOL> self . assertEqual ( producer . actions , [ ] ) <EOL> self . setWriteBufferSize ( client . transport , <NUM_LIT> ) <EOL> client . transport . writeSequence ( [ b"<STR_LIT:x>" * <NUM_LIT:50> ] * <NUM_LIT:20> ) <EOL> self . assertEqual ( producer . actions , [ "<STR_LIT>" ] ) <EOL> reactor = self . buildReactor ( ) <EOL> d = self . getConnectedClientAndServer ( reactor , "<STR_LIT:127.0.0.1>" , <EOL> socket . AF_INET ) <EOL> d . addCallback ( connected ) <EOL> d . addErrback ( log . err ) <EOL> self . runReactor ( reactor ) <EOL> self . assertEqual ( producer . actions , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_nonStreamingProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> test = self <EOL> @ implementer ( IPullProducer ) <EOL> class SaveActionProducer ( object ) : <EOL> client = None <EOL> def __init__ ( self ) : <EOL> self . actions = [ ] <EOL> def resumeProducing ( self ) : <EOL> self . actions . append ( "<STR_LIT>" ) <EOL> if self . actions . count ( "<STR_LIT>" ) == <NUM_LIT:2> : <EOL> self . client . transport . stopConsuming ( ) <EOL> else : <EOL> test . setWriteBufferSize ( self . client . transport , <NUM_LIT> ) <EOL> self . client . transport . writeSequence ( [ b"<STR_LIT:x>" * <NUM_LIT:50> ] * <NUM_LIT:20> ) <EOL> def stopProducing ( self ) : <EOL> self . actions . append ( "<STR_LIT>" ) <EOL> producer = SaveActionProducer ( ) <EOL> def connected ( protocols ) : <EOL> client = protocols [ <NUM_LIT:0> ] <EOL> producer . client = client <EOL> client . transport . registerProducer ( producer , False ) <EOL> self . assertEqual ( producer . actions , [ "<STR_LIT>" ] ) <EOL> reactor = self . buildReactor ( ) <EOL> d = self . getConnectedClientAndServer ( reactor , "<STR_LIT:127.0.0.1>" , <EOL> socket . AF_INET ) <EOL> d . addCallback ( connected ) <EOL> d . addErrback ( log . err ) <EOL> self . runReactor ( reactor ) <EOL> self . assertEqual ( producer . actions , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> class TCPTransportServerAddressTestMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def getConnectedClientAndServer ( self , reactor , interface , addressFamily ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def _testServerAddress ( self , interface , addressFamily , adressClass ) : <EOL> """<STR_LIT>""" <EOL> def connected ( protocols ) : <EOL> client , server , port = protocols <EOL> try : <EOL> self . assertEqual ( <EOL> "<STR_LIT>" % <EOL> ( server . transport . sessionno , port . getHost ( ) . port ) , <EOL> str ( server . transport ) ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" % <EOL> ( server . transport . sessionno , interface ) , <EOL> server . transport . logstr ) <EOL> [ peerAddress ] = server . factory . peerAddresses <EOL> self . assertIsInstance ( peerAddress , adressClass ) <EOL> self . assertEqual ( '<STR_LIT>' , peerAddress . type ) <EOL> self . assertEqual ( interface , peerAddress . host ) <EOL> finally : <EOL> server . transport . loseConnection ( ) <EOL> reactor = self . buildReactor ( ) <EOL> d = self . getConnectedClientAndServer ( reactor , interface , addressFamily ) <EOL> d . addCallback ( connected ) <EOL> d . addErrback ( log . err ) <EOL> self . runReactor ( reactor ) <EOL> def test_serverAddressTCP4 ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _testServerAddress ( "<STR_LIT:127.0.0.1>" , socket . AF_INET , <EOL> IPv4Address ) <EOL> def test_serverAddressTCP6 ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _testServerAddress ( getLinkLocalIPv6Address ( ) , <EOL> socket . AF_INET6 , IPv6Address ) <EOL> if ipv6Skip : <EOL> test_serverAddressTCP6 . skip = ipv6Skip <EOL> class TCPTransportTestsBuilder ( TCPTransportServerAddressTestMixin , <EOL> WriteSequenceTestsMixin , ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> def getConnectedClientAndServer ( self , reactor , interface , addressFamily ) : <EOL> """<STR_LIT>""" <EOL> server = MyServerFactory ( ) <EOL> server . protocolConnectionMade = Deferred ( ) <EOL> server . protocolConnectionLost = Deferred ( ) <EOL> client = MyClientFactory ( ) <EOL> client . protocolConnectionMade = Deferred ( ) <EOL> client . protocolConnectionLost = Deferred ( ) <EOL> port = reactor . listenTCP ( <NUM_LIT:0> , server , interface = interface ) <EOL> lostDeferred = gatherResults ( [ client . protocolConnectionLost , <EOL> server . protocolConnectionLost ] ) <EOL> def stop ( result ) : <EOL> reactor . stop ( ) <EOL> return result <EOL> lostDeferred . addBoth ( stop ) <EOL> startDeferred = gatherResults ( [ client . protocolConnectionMade , <EOL> server . protocolConnectionMade ] ) <EOL> deferred = Deferred ( ) <EOL> def start ( protocols ) : <EOL> client , server = protocols <EOL> log . msg ( "<STR_LIT>" % client ) <EOL> log . msg ( "<STR_LIT>" % server ) <EOL> deferred . callback ( ( client , server , port ) ) <EOL> startDeferred . addCallback ( start ) <EOL> reactor . connectTCP ( interface , port . getHost ( ) . port , client ) <EOL> return deferred <EOL> class AdoptStreamConnectionTestsBuilder ( TCPTransportServerAddressTestMixin , <EOL> WriteSequenceTestsMixin , <EOL> ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = ( IReactorFDSet , IReactorSocket ) <EOL> def getConnectedClientAndServer ( self , reactor , interface , addressFamily ) : <EOL> """<STR_LIT>""" <EOL> firstServer = MyServerFactory ( ) <EOL> firstServer . protocolConnectionMade = Deferred ( ) <EOL> server = MyServerFactory ( ) <EOL> server . protocolConnectionMade = Deferred ( ) <EOL> server . protocolConnectionLost = Deferred ( ) <EOL> client = MyClientFactory ( ) <EOL> client . protocolConnectionMade = Deferred ( ) <EOL> client . protocolConnectionLost = Deferred ( ) <EOL> port = reactor . listenTCP ( <NUM_LIT:0> , firstServer , interface = interface ) <EOL> def firtServerConnected ( proto ) : <EOL> reactor . removeReader ( proto . transport ) <EOL> reactor . removeWriter ( proto . transport ) <EOL> reactor . adoptStreamConnection ( <EOL> proto . transport . fileno ( ) , addressFamily , server ) <EOL> firstServer . protocolConnectionMade . addCallback ( firtServerConnected ) <EOL> lostDeferred = gatherResults ( [ client . protocolConnectionLost , <EOL> server . protocolConnectionLost ] ) <EOL> def stop ( result ) : <EOL> if reactor . running : <EOL> reactor . stop ( ) <EOL> return result <EOL> lostDeferred . addBoth ( stop ) <EOL> deferred = Deferred ( ) <EOL> deferred . addErrback ( stop ) <EOL> startDeferred = gatherResults ( [ client . protocolConnectionMade , <EOL> server . protocolConnectionMade ] ) <EOL> def start ( protocols ) : <EOL> client , server = protocols <EOL> log . msg ( "<STR_LIT>" % client ) <EOL> log . msg ( "<STR_LIT>" % server ) <EOL> deferred . callback ( ( client , server , port ) ) <EOL> startDeferred . addCallback ( start ) <EOL> reactor . connectTCP ( interface , port . getHost ( ) . port , client ) <EOL> return deferred <EOL> globals ( ) . update ( TCP4ClientTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( TCP6ClientTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( TCPPortTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( TCPFDPortTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( TCPConnectionTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( TCP4ConnectorTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( TCP6ConnectorTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( TCPTransportTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( AdoptStreamConnectionTestsBuilder . makeTestCaseClasses ( ) ) <EOL> class ServerAbortsTwice ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> def dataReceived ( self , data ) : <EOL> self . transport . abortConnection ( ) <EOL> self . transport . abortConnection ( ) <EOL> class ServerAbortsThenLoses ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> def dataReceived ( self , data ) : <EOL> self . transport . abortConnection ( ) <EOL> self . transport . loseConnection ( ) <EOL> class AbortServerWritingProtocol ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> def connectionMade ( self ) : <EOL> """<STR_LIT>""" <EOL> self . transport . write ( b"<STR_LIT>" ) <EOL> class ReadAbortServerProtocol ( AbortServerWritingProtocol ) : <EOL> """<STR_LIT>""" <EOL> def dataReceived ( self , data ) : <EOL> if data . replace ( b'<STR_LIT:X>' , b'<STR_LIT>' ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> class NoReadServer ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> def connectionMade ( self ) : <EOL> self . transport . stopReading ( ) <EOL> class EventualNoReadServer ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> gotData = False <EOL> stoppedReading = False <EOL> def dataReceived ( self , data ) : <EOL> if not self . gotData : <EOL> self . gotData = True <EOL> self . transport . registerProducer ( self , False ) <EOL> self . transport . write ( b"<STR_LIT:hello>" ) <EOL> def resumeProducing ( self ) : <EOL> if self . stoppedReading : <EOL> return <EOL> self . stoppedReading = True <EOL> self . transport . stopReading ( ) <EOL> def pauseProducing ( self ) : <EOL> pass <EOL> def stopProducing ( self ) : <EOL> pass <EOL> class BaseAbortingClient ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> inReactorMethod = False <EOL> def connectionLost ( self , reason ) : <EOL> if self . inReactorMethod : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> ConnectableProtocol . connectionLost ( self , reason ) <EOL> class WritingButNotAbortingClient ( BaseAbortingClient ) : <EOL> """<STR_LIT>""" <EOL> def connectionMade ( self ) : <EOL> self . transport . write ( b"<STR_LIT:hello>" ) <EOL> class AbortingClient ( BaseAbortingClient ) : <EOL> """<STR_LIT>""" <EOL> def dataReceived ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . inReactorMethod = True <EOL> self . writeAndAbort ( ) <EOL> self . inReactorMethod = False <EOL> def writeAndAbort ( self ) : <EOL> self . transport . write ( b"<STR_LIT:X>" * <NUM_LIT> ) <EOL> self . transport . abortConnection ( ) <EOL> self . transport . write ( b"<STR_LIT:Y>" * <NUM_LIT> ) <EOL> class AbortingTwiceClient ( AbortingClient ) : <EOL> """<STR_LIT>""" <EOL> def writeAndAbort ( self ) : <EOL> AbortingClient . writeAndAbort ( self ) <EOL> self . transport . abortConnection ( ) <EOL> class AbortingThenLosingClient ( AbortingClient ) : <EOL> """<STR_LIT>""" <EOL> def writeAndAbort ( self ) : <EOL> AbortingClient . writeAndAbort ( self ) <EOL> self . transport . loseConnection ( ) <EOL> class ProducerAbortingClient ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> inReactorMethod = True <EOL> producerStopped = False <EOL> def write ( self ) : <EOL> self . transport . write ( b"<STR_LIT>" * <NUM_LIT> ) <EOL> self . inRegisterProducer = True <EOL> self . transport . registerProducer ( self , False ) <EOL> self . inRegisterProducer = False <EOL> def connectionMade ( self ) : <EOL> self . write ( ) <EOL> def resumeProducing ( self ) : <EOL> self . inReactorMethod = True <EOL> if not self . inRegisterProducer : <EOL> self . transport . abortConnection ( ) <EOL> self . inReactorMethod = False <EOL> def stopProducing ( self ) : <EOL> self . producerStopped = True <EOL> def connectionLost ( self , reason ) : <EOL> if not self . producerStopped : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if self . inReactorMethod : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> ConnectableProtocol . connectionLost ( self , reason ) <EOL> class StreamingProducerClient ( ConnectableProtocol ) : <EOL> """<STR_LIT>""" <EOL> paused = False <EOL> extraWrites = <NUM_LIT:0> <EOL> inReactorMethod = False <EOL> def connectionMade ( self ) : <EOL> self . write ( ) <EOL> def write ( self ) : <EOL> """<STR_LIT>""" <EOL> self . transport . registerProducer ( self , True ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> self . transport . write ( b"<STR_LIT>" * <NUM_LIT> ) <EOL> def resumeProducing ( self ) : <EOL> self . paused = False <EOL> def stopProducing ( self ) : <EOL> pass <EOL> def pauseProducing ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . paused : <EOL> return <EOL> self . paused = True <EOL> self . reactor . callLater ( <NUM_LIT> , self . doAbort ) <EOL> def doAbort ( self ) : <EOL> if not self . paused : <EOL> log . err ( RuntimeError ( "<STR_LIT>" ) ) <EOL> self . inReactorMethod = True <EOL> self . transport . abortConnection ( ) <EOL> self . inReactorMethod = False <EOL> def connectionLost ( self , reason ) : <EOL> self . otherProtocol . transport . startReading ( ) <EOL> ConnectableProtocol . connectionLost ( self , reason ) <EOL> class StreamingProducerClientLater ( StreamingProducerClient ) : <EOL> """<STR_LIT>""" <EOL> def connectionMade ( self ) : <EOL> self . transport . write ( b"<STR_LIT:hello>" ) <EOL> self . gotData = False <EOL> def dataReceived ( self , data ) : <EOL> if not self . gotData : <EOL> self . gotData = True <EOL> self . write ( ) <EOL> class ProducerAbortingClientLater ( ProducerAbortingClient ) : <EOL> """<STR_LIT>""" <EOL> def connectionMade ( self ) : <EOL> pass <EOL> def dataReceived ( self , data ) : <EOL> self . write ( ) <EOL> class DataReceivedRaisingClient ( AbortingClient ) : <EOL> """<STR_LIT>""" <EOL> def dataReceived ( self , data ) : <EOL> self . transport . abortConnection ( ) <EOL> raise ZeroDivisionError ( "<STR_LIT>" ) <EOL> class ResumeThrowsClient ( ProducerAbortingClient ) : <EOL> """<STR_LIT>""" <EOL> def resumeProducing ( self ) : <EOL> if not self . inRegisterProducer : <EOL> self . transport . abortConnection ( ) <EOL> raise ZeroDivisionError ( "<STR_LIT>" ) <EOL> def connectionLost ( self , reason ) : <EOL> ConnectableProtocol . connectionLost ( self , reason ) <EOL> class AbortConnectionMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> endpoints = None <EOL> def runAbortTest ( self , clientClass , serverClass , <EOL> clientConnectionLostReason = None ) : <EOL> """<STR_LIT>""" <EOL> clientExpectedExceptions = ( ConnectionAborted , ConnectionLost ) <EOL> serverExpectedExceptions = ( ConnectionLost , ConnectionDone ) <EOL> if useSSL : <EOL> clientExpectedExceptions = clientExpectedExceptions + ( SSL . Error , ) <EOL> serverExpectedExceptions = serverExpectedExceptions + ( SSL . Error , ) <EOL> client = clientClass ( ) <EOL> server = serverClass ( ) <EOL> client . otherProtocol = server <EOL> server . otherProtocol = client <EOL> reactor = runProtocolsWithReactor ( self , server , client , self . endpoints ) <EOL> self . assertEqual ( reactor . removeAll ( ) , [ ] ) <EOL> self . assertEqual ( reactor . getDelayedCalls ( ) , [ ] ) <EOL> if clientConnectionLostReason is not None : <EOL> self . assertIsInstance ( <EOL> client . disconnectReason . value , <EOL> ( clientConnectionLostReason , ) + clientExpectedExceptions ) <EOL> else : <EOL> self . assertIsInstance ( client . disconnectReason . value , <EOL> clientExpectedExceptions ) <EOL> self . assertIsInstance ( server . disconnectReason . value , serverExpectedExceptions ) <EOL> def test_dataReceivedAbort ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . runAbortTest ( AbortingClient , ReadAbortServerProtocol ) <EOL> def test_clientAbortsConnectionTwice ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . runAbortTest ( AbortingTwiceClient , ReadAbortServerProtocol ) <EOL> def test_clientAbortsConnectionThenLosesConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . runAbortTest ( AbortingThenLosingClient , <EOL> ReadAbortServerProtocol ) <EOL> def test_serverAbortsConnectionTwice ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . runAbortTest ( WritingButNotAbortingClient , ServerAbortsTwice , <EOL> clientConnectionLostReason = ConnectionLost ) <EOL> def test_serverAbortsConnectionThenLosesConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . runAbortTest ( WritingButNotAbortingClient , <EOL> ServerAbortsThenLoses , <EOL> clientConnectionLostReason = ConnectionLost ) <EOL> def test_resumeProducingAbort ( self ) : <EOL> """<STR_LIT>""" <EOL> self . runAbortTest ( ProducerAbortingClient , <EOL> ConnectableProtocol ) <EOL> def test_resumeProducingAbortLater ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . runAbortTest ( ProducerAbortingClientLater , <EOL> AbortServerWritingProtocol ) <EOL> def test_fullWriteBuffer ( self ) : <EOL> """<STR_LIT>""" <EOL> self . runAbortTest ( StreamingProducerClient , <EOL> NoReadServer ) <EOL> def test_fullWriteBufferAfterByteExchange ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . runAbortTest ( StreamingProducerClientLater , <EOL> EventualNoReadServer ) <EOL> def test_dataReceivedThrows ( self ) : <EOL> """<STR_LIT>""" <EOL> self . runAbortTest ( DataReceivedRaisingClient , <EOL> AbortServerWritingProtocol , <EOL> clientConnectionLostReason = ZeroDivisionError ) <EOL> errors = self . flushLoggedErrors ( ZeroDivisionError ) <EOL> self . assertEqual ( len ( errors ) , <NUM_LIT:1> ) <EOL> def test_resumeProducingThrows ( self ) : <EOL> """<STR_LIT>""" <EOL> self . runAbortTest ( ResumeThrowsClient , <EOL> ConnectableProtocol , <EOL> clientConnectionLostReason = ZeroDivisionError ) <EOL> errors = self . flushLoggedErrors ( ZeroDivisionError ) <EOL> self . assertEqual ( len ( errors ) , <NUM_LIT:1> ) <EOL> class AbortConnectionTestCase ( ReactorBuilder , AbortConnectionMixin ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = ( IReactorTCP , ) <EOL> endpoints = TCPCreator ( ) <EOL> globals ( ) . update ( AbortConnectionTestCase . makeTestCaseClasses ( ) ) <EOL> class SimpleUtilityTestCase ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> if ipv6Skip : <EOL> skip = ipv6Skip <EOL> def test_resolveNumericHost ( self ) : <EOL> """<STR_LIT>""" <EOL> err = self . assertRaises ( socket . gaierror , _resolveIPv6 , "<STR_LIT:localhost>" , <NUM_LIT:1> ) <EOL> self . assertEqual ( err . args [ <NUM_LIT:0> ] , socket . EAI_NONAME ) <EOL> def test_resolveNumericService ( self ) : <EOL> """<STR_LIT>""" <EOL> err = self . assertRaises ( socket . gaierror , _resolveIPv6 , "<STR_LIT>" , "<STR_LIT:http>" ) <EOL> self . assertEqual ( err . args [ <NUM_LIT:0> ] , socket . EAI_NONAME ) <EOL> if platform . isWindows ( ) : <EOL> test_resolveNumericService . skip = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_resolveIPv6 ( self ) : <EOL> """<STR_LIT>""" <EOL> result = _resolveIPv6 ( "<STR_LIT>" , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( result ) , <NUM_LIT:4> ) <EOL> self . assertIsInstance ( result [ <NUM_LIT:2> ] , int ) <EOL> self . assertIsInstance ( result [ <NUM_LIT:3> ] , int ) <EOL> self . assertEqual ( result [ : <NUM_LIT:2> ] , ( "<STR_LIT>" , <NUM_LIT:2> ) ) </s>
