<s> """<STR_LIT>""" <EOL> from errno import EPERM <EOL> from socket import AF_INET , AF_INET6 , SOCK_STREAM , IPPROTO_TCP <EOL> from zope . interface import implements <EOL> from zope . interface . verify import verifyObject <EOL> from twisted . trial import unittest <EOL> from twisted . internet import error , interfaces , defer <EOL> from twisted . internet import endpoints <EOL> from twisted . internet . address import IPv4Address , IPv6Address , UNIXAddress <EOL> from twisted . internet . protocol import ClientFactory , Protocol <EOL> from twisted . test . proto_helpers import ( <EOL> MemoryReactor , RaisingMemoryReactor , StringTransport ) <EOL> from twisted . python . failure import Failure <EOL> from twisted . python . systemd import ListenFDs <EOL> from twisted . plugin import getPlugins <EOL> from twisted import plugins <EOL> from twisted . python . modules import getModule <EOL> from twisted . python . filepath import FilePath <EOL> from twisted . protocols import basic <EOL> from twisted . internet import protocol , reactor , stdio <EOL> from twisted . internet . stdio import PipeAddress <EOL> pemPath = getModule ( "<STR_LIT>" ) . filePath . sibling ( "<STR_LIT>" ) <EOL> casPath = getModule ( __name__ ) . filePath . sibling ( "<STR_LIT>" ) <EOL> escapedPEMPathName = endpoints . quoteStringArgument ( pemPath . path ) <EOL> escapedCAsPathName = endpoints . quoteStringArgument ( casPath . path ) <EOL> try : <EOL> from twisted . test . test_sslverify import makeCertificate <EOL> from twisted . internet . ssl import CertificateOptions , Certificate , KeyPair , PrivateCertificate <EOL> from OpenSSL . SSL import ContextType <EOL> testCertificate = Certificate . loadPEM ( pemPath . getContent ( ) ) <EOL> testPrivateCertificate = PrivateCertificate . loadPEM ( pemPath . getContent ( ) ) <EOL> skipSSL = False <EOL> except ImportError : <EOL> skipSSL = "<STR_LIT>" <EOL> class TestProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . data = [ ] <EOL> self . connectionsLost = [ ] <EOL> self . connectionMadeCalls = <NUM_LIT:0> <EOL> def logPrefix ( self ) : <EOL> return "<STR_LIT>" <EOL> def connectionMade ( self ) : <EOL> self . connectionMadeCalls += <NUM_LIT:1> <EOL> def dataReceived ( self , data ) : <EOL> self . data . append ( data ) <EOL> def connectionLost ( self , reason ) : <EOL> self . connectionsLost . append ( reason ) <EOL> class TestHalfCloseableProtocol ( TestProtocol ) : <EOL> """<STR_LIT>""" <EOL> implements ( interfaces . IHalfCloseableProtocol ) <EOL> def __init__ ( self ) : <EOL> TestProtocol . __init__ ( self ) <EOL> self . readLost = False <EOL> self . writeLost = False <EOL> def readConnectionLost ( self ) : <EOL> self . readLost = True <EOL> def writeConnectionLost ( self ) : <EOL> self . writeLost = True <EOL> class TestFileDescriptorReceiverProtocol ( TestProtocol ) : <EOL> """<STR_LIT>""" <EOL> implements ( interfaces . IFileDescriptorReceiver ) <EOL> def connectionMade ( self ) : <EOL> TestProtocol . connectionMade ( self ) <EOL> self . receivedDescriptors = [ ] <EOL> def fileDescriptorReceived ( self , descriptor ) : <EOL> self . receivedDescriptors . append ( descriptor ) <EOL> class TestFactory ( ClientFactory ) : <EOL> """<STR_LIT>""" <EOL> protocol = TestProtocol <EOL> class WrappingFactoryTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_doStart ( self ) : <EOL> """<STR_LIT>""" <EOL> factory = ClientFactory ( ) <EOL> wf = endpoints . _WrappingFactory ( factory ) <EOL> wf . doStart ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , factory . numPorts ) <EOL> def test_doStop ( self ) : <EOL> """<STR_LIT>""" <EOL> factory = ClientFactory ( ) <EOL> factory . numPorts = <NUM_LIT:3> <EOL> wf = endpoints . _WrappingFactory ( factory ) <EOL> wf . doStop ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , factory . numPorts ) <EOL> def test_failedBuildProtocol ( self ) : <EOL> """<STR_LIT>""" <EOL> class BogusFactory ( ClientFactory ) : <EOL> """<STR_LIT>""" <EOL> def buildProtocol ( self , addr ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> wf = endpoints . _WrappingFactory ( BogusFactory ( ) ) <EOL> wf . buildProtocol ( None ) <EOL> d = self . assertFailure ( wf . _onConnection , ValueError ) <EOL> d . addCallback ( lambda e : self . assertEqual ( <EOL> e . args , <EOL> ( "<STR_LIT>" , ) ) ) <EOL> return d <EOL> def test_logPrefixPassthrough ( self ) : <EOL> """<STR_LIT>""" <EOL> wf = endpoints . _WrappingFactory ( TestFactory ( ) ) <EOL> wp = wf . buildProtocol ( None ) <EOL> self . assertEqual ( wp . logPrefix ( ) , "<STR_LIT>" ) <EOL> def test_logPrefixDefault ( self ) : <EOL> """<STR_LIT>""" <EOL> class NoProtocol ( object ) : <EOL> pass <EOL> factory = TestFactory ( ) <EOL> factory . protocol = NoProtocol <EOL> wf = endpoints . _WrappingFactory ( factory ) <EOL> wp = wf . buildProtocol ( None ) <EOL> self . assertEqual ( wp . logPrefix ( ) , "<STR_LIT>" ) <EOL> def test_wrappedProtocolDataReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> wf = endpoints . _WrappingFactory ( TestFactory ( ) ) <EOL> p = wf . buildProtocol ( None ) <EOL> p . makeConnection ( None ) <EOL> p . dataReceived ( '<STR_LIT:foo>' ) <EOL> self . assertEqual ( p . _wrappedProtocol . data , [ '<STR_LIT:foo>' ] ) <EOL> p . dataReceived ( '<STR_LIT:bar>' ) <EOL> self . assertEqual ( p . _wrappedProtocol . data , [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] ) <EOL> def test_wrappedProtocolTransport ( self ) : <EOL> """<STR_LIT>""" <EOL> wf = endpoints . _WrappingFactory ( TestFactory ( ) ) <EOL> p = wf . buildProtocol ( None ) <EOL> dummyTransport = object ( ) <EOL> p . makeConnection ( dummyTransport ) <EOL> self . assertEqual ( p . transport , dummyTransport ) <EOL> self . assertEqual ( p . _wrappedProtocol . transport , dummyTransport ) <EOL> def test_wrappedProtocolConnectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> tf = TestFactory ( ) <EOL> wf = endpoints . _WrappingFactory ( tf ) <EOL> p = wf . buildProtocol ( None ) <EOL> p . connectionLost ( "<STR_LIT>" ) <EOL> self . assertEqual ( p . _wrappedProtocol . connectionsLost , [ "<STR_LIT>" ] ) <EOL> def test_clientConnectionFailed ( self ) : <EOL> """<STR_LIT>""" <EOL> wf = endpoints . _WrappingFactory ( TestFactory ( ) ) <EOL> expectedFailure = Failure ( error . ConnectError ( string = "<STR_LIT>" ) ) <EOL> wf . clientConnectionFailed ( <EOL> None , <EOL> expectedFailure ) <EOL> errors = [ ] <EOL> def gotError ( f ) : <EOL> errors . append ( f ) <EOL> wf . _onConnection . addErrback ( gotError ) <EOL> self . assertEqual ( errors , [ expectedFailure ] ) <EOL> def test_wrappingProtocolFileDescriptorReceiver ( self ) : <EOL> """<STR_LIT>""" <EOL> connectedDeferred = None <EOL> applicationProtocol = TestFileDescriptorReceiverProtocol ( ) <EOL> wrapper = endpoints . _WrappingProtocol ( <EOL> connectedDeferred , applicationProtocol ) <EOL> self . assertTrue ( interfaces . IFileDescriptorReceiver . providedBy ( wrapper ) ) <EOL> self . assertTrue ( verifyObject ( interfaces . IFileDescriptorReceiver , wrapper ) ) <EOL> def test_wrappingProtocolNotFileDescriptorReceiver ( self ) : <EOL> """<STR_LIT>""" <EOL> tp = TestProtocol ( ) <EOL> p = endpoints . _WrappingProtocol ( None , tp ) <EOL> self . assertFalse ( interfaces . IFileDescriptorReceiver . providedBy ( p ) ) <EOL> def test_wrappedProtocolFileDescriptorReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> wrappedProtocol = TestFileDescriptorReceiverProtocol ( ) <EOL> wrapper = endpoints . _WrappingProtocol ( <EOL> defer . Deferred ( ) , wrappedProtocol ) <EOL> wrapper . makeConnection ( StringTransport ( ) ) <EOL> wrapper . fileDescriptorReceived ( <NUM_LIT> ) <EOL> self . assertEqual ( wrappedProtocol . receivedDescriptors , [ <NUM_LIT> ] ) <EOL> def test_wrappingProtocolHalfCloseable ( self ) : <EOL> """<STR_LIT>""" <EOL> cd = object ( ) <EOL> hcp = TestHalfCloseableProtocol ( ) <EOL> p = endpoints . _WrappingProtocol ( cd , hcp ) <EOL> self . assertEqual ( <EOL> interfaces . IHalfCloseableProtocol . providedBy ( p ) , True ) <EOL> def test_wrappingProtocolNotHalfCloseable ( self ) : <EOL> """<STR_LIT>""" <EOL> tp = TestProtocol ( ) <EOL> p = endpoints . _WrappingProtocol ( None , tp ) <EOL> self . assertEqual ( <EOL> interfaces . IHalfCloseableProtocol . providedBy ( p ) , False ) <EOL> def test_wrappedProtocolReadConnectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> hcp = TestHalfCloseableProtocol ( ) <EOL> p = endpoints . _WrappingProtocol ( None , hcp ) <EOL> p . readConnectionLost ( ) <EOL> self . assertEqual ( hcp . readLost , True ) <EOL> def test_wrappedProtocolWriteConnectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> hcp = TestHalfCloseableProtocol ( ) <EOL> p = endpoints . _WrappingProtocol ( None , hcp ) <EOL> p . writeConnectionLost ( ) <EOL> self . assertEqual ( hcp . writeLost , True ) <EOL> class ClientEndpointTestCaseMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def retrieveConnectedFactory ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return self . expectedClients ( reactor ) [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] <EOL> def test_endpointConnectSuccess ( self ) : <EOL> """<STR_LIT>""" <EOL> proto = object ( ) <EOL> mreactor = MemoryReactor ( ) <EOL> clientFactory = object ( ) <EOL> ep , expectedArgs , ignoredDest = self . createClientEndpoint ( <EOL> mreactor , clientFactory ) <EOL> d = ep . connect ( clientFactory ) <EOL> receivedProtos = [ ] <EOL> def checkProto ( p ) : <EOL> receivedProtos . append ( p ) <EOL> d . addCallback ( checkProto ) <EOL> factory = self . retrieveConnectedFactory ( mreactor ) <EOL> factory . _onConnection . callback ( proto ) <EOL> self . assertEqual ( receivedProtos , [ proto ] ) <EOL> expectedClients = self . expectedClients ( mreactor ) <EOL> self . assertEqual ( len ( expectedClients ) , <NUM_LIT:1> ) <EOL> self . assertConnectArgs ( expectedClients [ <NUM_LIT:0> ] , expectedArgs ) <EOL> def test_endpointConnectFailure ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedError = error . ConnectError ( string = "<STR_LIT>" ) <EOL> mreactor = RaisingMemoryReactor ( connectException = expectedError ) <EOL> clientFactory = object ( ) <EOL> ep , ignoredArgs , ignoredDest = self . createClientEndpoint ( <EOL> mreactor , clientFactory ) <EOL> d = ep . connect ( clientFactory ) <EOL> receivedExceptions = [ ] <EOL> def checkFailure ( f ) : <EOL> receivedExceptions . append ( f . value ) <EOL> d . addErrback ( checkFailure ) <EOL> self . assertEqual ( receivedExceptions , [ expectedError ] ) <EOL> def test_endpointConnectingCancelled ( self ) : <EOL> """<STR_LIT>""" <EOL> mreactor = MemoryReactor ( ) <EOL> clientFactory = object ( ) <EOL> ep , ignoredArgs , address = self . createClientEndpoint ( <EOL> mreactor , clientFactory ) <EOL> d = ep . connect ( clientFactory ) <EOL> receivedFailures = [ ] <EOL> def checkFailure ( f ) : <EOL> receivedFailures . append ( f ) <EOL> d . addErrback ( checkFailure ) <EOL> d . cancel ( ) <EOL> attemptFactory = self . retrieveConnectedFactory ( mreactor ) <EOL> attemptFactory . clientConnectionFailed ( None , Failure ( error . UserError ( ) ) ) <EOL> self . assertEqual ( len ( receivedFailures ) , <NUM_LIT:1> ) <EOL> failure = receivedFailures [ <NUM_LIT:0> ] <EOL> self . assertIsInstance ( failure . value , error . ConnectingCancelledError ) <EOL> self . assertEqual ( failure . value . address , address ) <EOL> def test_endpointConnectNonDefaultArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> factory = object ( ) <EOL> mreactor = MemoryReactor ( ) <EOL> ep , expectedArgs , ignoredHost = self . createClientEndpoint ( <EOL> mreactor , factory , <EOL> ** self . connectArgs ( ) ) <EOL> ep . connect ( factory ) <EOL> expectedClients = self . expectedClients ( mreactor ) <EOL> self . assertEqual ( len ( expectedClients ) , <NUM_LIT:1> ) <EOL> self . assertConnectArgs ( expectedClients [ <NUM_LIT:0> ] , expectedArgs ) <EOL> class ServerEndpointTestCaseMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def test_endpointListenSuccess ( self ) : <EOL> """<STR_LIT>""" <EOL> mreactor = MemoryReactor ( ) <EOL> factory = object ( ) <EOL> ep , expectedArgs , expectedHost = self . createServerEndpoint ( <EOL> mreactor , factory ) <EOL> d = ep . listen ( factory ) <EOL> receivedHosts = [ ] <EOL> def checkPortAndServer ( port ) : <EOL> receivedHosts . append ( port . getHost ( ) ) <EOL> d . addCallback ( checkPortAndServer ) <EOL> self . assertEqual ( receivedHosts , [ expectedHost ] ) <EOL> self . assertEqual ( self . expectedServers ( mreactor ) , [ expectedArgs ] ) <EOL> def test_endpointListenFailure ( self ) : <EOL> """<STR_LIT>""" <EOL> factory = object ( ) <EOL> exception = error . CannotListenError ( '<STR_LIT>' , <NUM_LIT> , factory ) <EOL> mreactor = RaisingMemoryReactor ( listenException = exception ) <EOL> ep , ignoredArgs , ignoredDest = self . createServerEndpoint ( <EOL> mreactor , factory ) <EOL> d = ep . listen ( object ( ) ) <EOL> receivedExceptions = [ ] <EOL> def checkFailure ( f ) : <EOL> receivedExceptions . append ( f . value ) <EOL> d . addErrback ( checkFailure ) <EOL> self . assertEqual ( receivedExceptions , [ exception ] ) <EOL> def test_endpointListenNonDefaultArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> factory = object ( ) <EOL> mreactor = MemoryReactor ( ) <EOL> ep , expectedArgs , ignoredHost = self . createServerEndpoint ( <EOL> mreactor , factory , <EOL> ** self . listenArgs ( ) ) <EOL> ep . listen ( factory ) <EOL> expectedServers = self . expectedServers ( mreactor ) <EOL> self . assertEqual ( expectedServers , [ expectedArgs ] ) <EOL> class EndpointTestCaseMixin ( ServerEndpointTestCaseMixin , <EOL> ClientEndpointTestCaseMixin ) : <EOL> """<STR_LIT>""" <EOL> class StdioFactory ( protocol . Factory ) : <EOL> protocol = basic . LineReceiver <EOL> class StandardIOEndpointsTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . ep = endpoints . StandardIOEndpoint ( reactor ) <EOL> def test_standardIOInstance ( self ) : <EOL> """<STR_LIT>""" <EOL> self . d = self . ep . listen ( StdioFactory ( ) ) <EOL> def checkInstanceAndLoseConnection ( stdioOb ) : <EOL> self . assertIsInstance ( stdioOb , stdio . StandardIO ) <EOL> stdioOb . loseConnection ( ) <EOL> self . d . addCallback ( checkInstanceAndLoseConnection ) <EOL> return self . d <EOL> def test_reactor ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . ep . _reactor , reactor ) <EOL> def test_protocol ( self ) : <EOL> """<STR_LIT>""" <EOL> self . d = self . ep . listen ( StdioFactory ( ) ) <EOL> def checkProtocol ( stdioOb ) : <EOL> from twisted . python . runtime import platform <EOL> if platform . isWindows ( ) : <EOL> self . assertIsInstance ( stdioOb . proto , basic . LineReceiver ) <EOL> else : <EOL> self . assertIsInstance ( stdioOb . protocol , basic . LineReceiver ) <EOL> stdioOb . loseConnection ( ) <EOL> self . d . addCallback ( checkProtocol ) <EOL> return self . d <EOL> def test_address ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestAddrFactory ( protocol . Factory ) : <EOL> protocol = basic . LineReceiver <EOL> _address = None <EOL> def buildProtocol ( self , addr ) : <EOL> self . _address = addr <EOL> p = self . protocol ( ) <EOL> p . factory = self <EOL> return p <EOL> def getAddress ( self ) : <EOL> return self . _address <EOL> myFactory = TestAddrFactory ( ) <EOL> self . d = self . ep . listen ( myFactory ) <EOL> def checkAddress ( stdioOb ) : <EOL> self . assertIsInstance ( myFactory . getAddress ( ) , PipeAddress ) <EOL> stdioOb . loseConnection ( ) <EOL> self . d . addCallback ( checkAddress ) <EOL> return self . d <EOL> class TCP4EndpointsTestCase ( EndpointTestCaseMixin , unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def expectedServers ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . tcpServers <EOL> def expectedClients ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . tcpClients <EOL> def assertConnectArgs ( self , receivedArgs , expectedArgs ) : <EOL> """<STR_LIT>""" <EOL> ( host , port , ignoredFactory , timeout , bindAddress ) = receivedArgs <EOL> ( expectedHost , expectedPort , _ignoredFactory , <EOL> expectedTimeout , expectedBindAddress ) = expectedArgs <EOL> self . assertEqual ( host , expectedHost ) <EOL> self . assertEqual ( port , expectedPort ) <EOL> self . assertEqual ( timeout , expectedTimeout ) <EOL> self . assertEqual ( bindAddress , expectedBindAddress ) <EOL> def connectArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT>' : ( '<STR_LIT:localhost>' , <NUM_LIT> ) } <EOL> def listenArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:100> , '<STR_LIT>' : '<STR_LIT:127.0.0.1>' } <EOL> def createServerEndpoint ( self , reactor , factory , ** listenArgs ) : <EOL> """<STR_LIT>""" <EOL> address = IPv4Address ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:0> ) <EOL> if listenArgs is None : <EOL> listenArgs = { } <EOL> return ( endpoints . TCP4ServerEndpoint ( reactor , <EOL> address . port , <EOL> ** listenArgs ) , <EOL> ( address . port , factory , <EOL> listenArgs . get ( '<STR_LIT>' , <NUM_LIT:50> ) , <EOL> listenArgs . get ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> address ) <EOL> def createClientEndpoint ( self , reactor , clientFactory , ** connectArgs ) : <EOL> """<STR_LIT>""" <EOL> address = IPv4Address ( "<STR_LIT>" , "<STR_LIT:localhost>" , <NUM_LIT> ) <EOL> return ( endpoints . TCP4ClientEndpoint ( reactor , <EOL> address . host , <EOL> address . port , <EOL> ** connectArgs ) , <EOL> ( address . host , address . port , clientFactory , <EOL> connectArgs . get ( '<STR_LIT>' , <NUM_LIT:30> ) , <EOL> connectArgs . get ( '<STR_LIT>' , None ) ) , <EOL> address ) <EOL> class TCP6EndpointsTestCase ( EndpointTestCaseMixin , unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def expectedServers ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . tcpServers <EOL> def expectedClients ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . tcpClients <EOL> def assertConnectArgs ( self , receivedArgs , expectedArgs ) : <EOL> """<STR_LIT>""" <EOL> ( host , port , ignoredFactory , timeout , bindAddress ) = receivedArgs <EOL> ( expectedHost , expectedPort , _ignoredFactory , <EOL> expectedTimeout , expectedBindAddress ) = expectedArgs <EOL> self . assertEqual ( host , expectedHost ) <EOL> self . assertEqual ( port , expectedPort ) <EOL> self . assertEqual ( timeout , expectedTimeout ) <EOL> self . assertEqual ( bindAddress , expectedBindAddress ) <EOL> def connectArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT>' : ( '<STR_LIT:localhost>' , <NUM_LIT> ) } <EOL> def listenArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:100> , '<STR_LIT>' : '<STR_LIT>' } <EOL> def createServerEndpoint ( self , reactor , factory , ** listenArgs ) : <EOL> """<STR_LIT>""" <EOL> interface = listenArgs . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> address = IPv6Address ( "<STR_LIT>" , interface , <NUM_LIT:0> ) <EOL> if listenArgs is None : <EOL> listenArgs = { } <EOL> return ( endpoints . TCP6ServerEndpoint ( reactor , <EOL> address . port , <EOL> ** listenArgs ) , <EOL> ( address . port , factory , <EOL> listenArgs . get ( '<STR_LIT>' , <NUM_LIT:50> ) , <EOL> interface ) , <EOL> address ) <EOL> def createClientEndpoint ( self , reactor , clientFactory , ** connectArgs ) : <EOL> """<STR_LIT>""" <EOL> address = IPv6Address ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> return ( endpoints . TCP6ClientEndpoint ( reactor , <EOL> address . host , <EOL> address . port , <EOL> ** connectArgs ) , <EOL> ( address . host , address . port , clientFactory , <EOL> connectArgs . get ( '<STR_LIT>' , <NUM_LIT:30> ) , <EOL> connectArgs . get ( '<STR_LIT>' , None ) ) , <EOL> address ) <EOL> class TCP6EndpointNameResolutionTestCase ( ClientEndpointTestCaseMixin , <EOL> unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def createClientEndpoint ( self , reactor , clientFactory , ** connectArgs ) : <EOL> """<STR_LIT>""" <EOL> address = IPv6Address ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> self . ep = endpoints . TCP6ClientEndpoint ( reactor , <EOL> '<STR_LIT>' , <EOL> address . port , <EOL> ** connectArgs ) <EOL> def testNameResolution ( host ) : <EOL> self . assertEqual ( "<STR_LIT>" , host ) <EOL> data = [ ( AF_INET6 , SOCK_STREAM , IPPROTO_TCP , '<STR_LIT>' , ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> ( AF_INET6 , SOCK_STREAM , IPPROTO_TCP , '<STR_LIT>' , ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> ( AF_INET6 , SOCK_STREAM , IPPROTO_TCP , '<STR_LIT>' , ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) ] <EOL> return defer . succeed ( data ) <EOL> self . ep . _nameResolution = testNameResolution <EOL> return ( self . ep , <EOL> ( address . host , address . port , clientFactory , <EOL> connectArgs . get ( '<STR_LIT>' , <NUM_LIT:30> ) , <EOL> connectArgs . get ( '<STR_LIT>' , None ) ) , <EOL> address ) <EOL> def connectArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT>' : ( '<STR_LIT:localhost>' , <NUM_LIT> ) } <EOL> def expectedClients ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . tcpClients <EOL> def assertConnectArgs ( self , receivedArgs , expectedArgs ) : <EOL> """<STR_LIT>""" <EOL> ( host , port , ignoredFactory , timeout , bindAddress ) = receivedArgs <EOL> ( expectedHost , expectedPort , _ignoredFactory , <EOL> expectedTimeout , expectedBindAddress ) = expectedArgs <EOL> self . assertEqual ( host , expectedHost ) <EOL> self . assertEqual ( port , expectedPort ) <EOL> self . assertEqual ( timeout , expectedTimeout ) <EOL> self . assertEqual ( bindAddress , expectedBindAddress ) <EOL> def test_nameResolution ( self ) : <EOL> """<STR_LIT>""" <EOL> calls = [ ] <EOL> def fakeDeferToThread ( f , * args , ** kwargs ) : <EOL> calls . append ( ( f , args , kwargs ) ) <EOL> return defer . Deferred ( ) <EOL> endpoint = endpoints . TCP6ClientEndpoint ( reactor , '<STR_LIT>' , <EOL> <NUM_LIT> ) <EOL> fakegetaddrinfo = object ( ) <EOL> endpoint . _getaddrinfo = fakegetaddrinfo <EOL> endpoint . _deferToThread = fakeDeferToThread <EOL> endpoint . connect ( TestFactory ( ) ) <EOL> self . assertEqual ( <EOL> [ ( fakegetaddrinfo , ( "<STR_LIT>" , <NUM_LIT:0> , AF_INET6 ) , { } ) ] , calls ) <EOL> class SSL4EndpointsTestCase ( EndpointTestCaseMixin , <EOL> unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> if skipSSL : <EOL> skip = skipSSL <EOL> def expectedServers ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . sslServers <EOL> def expectedClients ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . sslClients <EOL> def assertConnectArgs ( self , receivedArgs , expectedArgs ) : <EOL> """<STR_LIT>""" <EOL> ( host , port , ignoredFactory , contextFactory , timeout , <EOL> bindAddress ) = receivedArgs <EOL> ( expectedHost , expectedPort , _ignoredFactory , expectedContextFactory , <EOL> expectedTimeout , expectedBindAddress ) = expectedArgs <EOL> self . assertEqual ( host , expectedHost ) <EOL> self . assertEqual ( port , expectedPort ) <EOL> self . assertEqual ( contextFactory , expectedContextFactory ) <EOL> self . assertEqual ( timeout , expectedTimeout ) <EOL> self . assertEqual ( bindAddress , expectedBindAddress ) <EOL> def connectArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT>' : ( '<STR_LIT:localhost>' , <NUM_LIT> ) } <EOL> def listenArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:100> , '<STR_LIT>' : '<STR_LIT:127.0.0.1>' } <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . sKey , self . sCert = makeCertificate ( <EOL> O = "<STR_LIT>" , <EOL> CN = "<STR_LIT>" ) <EOL> self . cKey , self . cCert = makeCertificate ( <EOL> O = "<STR_LIT>" , <EOL> CN = "<STR_LIT>" ) <EOL> self . serverSSLContext = CertificateOptions ( <EOL> privateKey = self . sKey , <EOL> certificate = self . sCert , <EOL> requireCertificate = False ) <EOL> self . clientSSLContext = CertificateOptions ( <EOL> requireCertificate = False ) <EOL> def createServerEndpoint ( self , reactor , factory , ** listenArgs ) : <EOL> """<STR_LIT>""" <EOL> address = IPv4Address ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:0> ) <EOL> return ( endpoints . SSL4ServerEndpoint ( reactor , <EOL> address . port , <EOL> self . serverSSLContext , <EOL> ** listenArgs ) , <EOL> ( address . port , factory , self . serverSSLContext , <EOL> listenArgs . get ( '<STR_LIT>' , <NUM_LIT:50> ) , <EOL> listenArgs . get ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> address ) <EOL> def createClientEndpoint ( self , reactor , clientFactory , ** connectArgs ) : <EOL> """<STR_LIT>""" <EOL> address = IPv4Address ( "<STR_LIT>" , "<STR_LIT:localhost>" , <NUM_LIT> ) <EOL> if connectArgs is None : <EOL> connectArgs = { } <EOL> return ( endpoints . SSL4ClientEndpoint ( reactor , <EOL> address . host , <EOL> address . port , <EOL> self . clientSSLContext , <EOL> ** connectArgs ) , <EOL> ( address . host , address . port , clientFactory , <EOL> self . clientSSLContext , <EOL> connectArgs . get ( '<STR_LIT>' , <NUM_LIT:30> ) , <EOL> connectArgs . get ( '<STR_LIT>' , None ) ) , <EOL> address ) <EOL> class UNIXEndpointsTestCase ( EndpointTestCaseMixin , <EOL> unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def retrieveConnectedFactory ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return self . expectedClients ( reactor ) [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> def expectedServers ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . unixServers <EOL> def expectedClients ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . unixClients <EOL> def assertConnectArgs ( self , receivedArgs , expectedArgs ) : <EOL> """<STR_LIT>""" <EOL> ( path , ignoredFactory , timeout , checkPID ) = receivedArgs <EOL> ( expectedPath , _ignoredFactory , expectedTimeout , <EOL> expectedCheckPID ) = expectedArgs <EOL> self . assertEqual ( path , expectedPath ) <EOL> self . assertEqual ( timeout , expectedTimeout ) <EOL> self . assertEqual ( checkPID , expectedCheckPID ) <EOL> def connectArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:1> } <EOL> def listenArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : <NUM_LIT:100> , '<STR_LIT>' : <NUM_LIT:0> <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:1> } <EOL> def createServerEndpoint ( self , reactor , factory , ** listenArgs ) : <EOL> """<STR_LIT>""" <EOL> address = UNIXAddress ( self . mktemp ( ) ) <EOL> return ( endpoints . UNIXServerEndpoint ( reactor , address . name , <EOL> ** listenArgs ) , <EOL> ( address . name , factory , <EOL> listenArgs . get ( '<STR_LIT>' , <NUM_LIT:50> ) , <EOL> listenArgs . get ( '<STR_LIT>' , <NUM_LIT:0> <NUM_LIT> ) , <EOL> listenArgs . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) , <EOL> address ) <EOL> def createClientEndpoint ( self , reactor , clientFactory , ** connectArgs ) : <EOL> """<STR_LIT>""" <EOL> address = UNIXAddress ( self . mktemp ( ) ) <EOL> return ( endpoints . UNIXClientEndpoint ( reactor , address . name , <EOL> ** connectArgs ) , <EOL> ( address . name , clientFactory , <EOL> connectArgs . get ( '<STR_LIT>' , <NUM_LIT:30> ) , <EOL> connectArgs . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) , <EOL> address ) <EOL> class ParserTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> f = "<STR_LIT>" <EOL> def parse ( self , * a , ** kw ) : <EOL> """<STR_LIT>""" <EOL> return endpoints . _parseServer ( * a , ** kw ) <EOL> def test_simpleTCP ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . parse ( '<STR_LIT>' , self . f ) , <EOL> ( '<STR_LIT>' , ( <NUM_LIT> , self . f ) , { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:50> } ) ) <EOL> def test_interfaceTCP ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . parse ( '<STR_LIT>' , self . f ) , <EOL> ( '<STR_LIT>' , ( <NUM_LIT> , self . f ) , { '<STR_LIT>' : '<STR_LIT:127.0.0.1>' , '<STR_LIT>' : <NUM_LIT:50> } ) ) <EOL> def test_backlogTCP ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . parse ( '<STR_LIT>' , self . f ) , <EOL> ( '<STR_LIT>' , ( <NUM_LIT> , self . f ) , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:6> } ) ) <EOL> def test_simpleUNIX ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . parse ( '<STR_LIT>' , self . f ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , self . f ) , <EOL> { '<STR_LIT>' : <NUM_LIT:0> <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:50> , '<STR_LIT>' : True } ) ) <EOL> def test_modeUNIX ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . parse ( '<STR_LIT>' , self . f ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , self . f ) , <EOL> { '<STR_LIT>' : <NUM_LIT:0> <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:50> , '<STR_LIT>' : True } ) ) <EOL> def test_wantPIDUNIX ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . parse ( '<STR_LIT>' , self . f ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , self . f ) , <EOL> { '<STR_LIT>' : <NUM_LIT:0> <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:50> , '<STR_LIT>' : False } ) ) <EOL> def test_escape ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . parse ( r'<STR_LIT>' , self . f ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , self . f ) , <EOL> { '<STR_LIT>' : <NUM_LIT:0> <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:50> , '<STR_LIT>' : True } ) ) <EOL> def test_quoteStringArgument ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( endpoints . quoteStringArgument ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" ) <EOL> def test_impliedEscape ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . parse ( r'<STR_LIT>' , self . f ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , self . f ) , <EOL> { '<STR_LIT>' : <NUM_LIT:0> <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:50> , '<STR_LIT>' : True } ) ) <EOL> def test_nonstandardDefault ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> self . parse ( '<STR_LIT:filename>' , self . f , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT:filename>' , self . f ) , <EOL> { '<STR_LIT>' : <NUM_LIT:0> <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:50> , '<STR_LIT>' : True } ) ) <EOL> def test_unknownType ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( ValueError , self . parse , "<STR_LIT>" , self . f ) <EOL> class ServerStringTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_tcp ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> server = endpoints . serverFromString ( <EOL> reactor , "<STR_LIT>" ) <EOL> self . assertIsInstance ( server , endpoints . TCP4ServerEndpoint ) <EOL> self . assertIdentical ( server . _reactor , reactor ) <EOL> self . assertEqual ( server . _port , <NUM_LIT> ) <EOL> self . assertEqual ( server . _backlog , <NUM_LIT:12> ) <EOL> self . assertEqual ( server . _interface , "<STR_LIT>" ) <EOL> def test_ssl ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> server = endpoints . serverFromString ( <EOL> reactor , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( escapedPEMPathName , <EOL> escapedPEMPathName ) ) <EOL> self . assertIsInstance ( server , endpoints . SSL4ServerEndpoint ) <EOL> self . assertIdentical ( server . _reactor , reactor ) <EOL> self . assertEqual ( server . _port , <NUM_LIT> ) <EOL> self . assertEqual ( server . _backlog , <NUM_LIT:12> ) <EOL> self . assertEqual ( server . _interface , "<STR_LIT>" ) <EOL> ctx = server . _sslContextFactory . getContext ( ) <EOL> self . assertIsInstance ( ctx , ContextType ) <EOL> if skipSSL : <EOL> test_ssl . skip = skipSSL <EOL> def test_unix ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> endpoint = endpoints . serverFromString ( <EOL> reactor , <EOL> "<STR_LIT>" ) <EOL> self . assertIsInstance ( endpoint , endpoints . UNIXServerEndpoint ) <EOL> self . assertIdentical ( endpoint . _reactor , reactor ) <EOL> self . assertEqual ( endpoint . _address , "<STR_LIT>" ) <EOL> self . assertEqual ( endpoint . _backlog , <NUM_LIT:7> ) <EOL> self . assertEqual ( endpoint . _mode , <NUM_LIT:0> <NUM_LIT> ) <EOL> self . assertEqual ( endpoint . _wantPID , True ) <EOL> def test_implicitDefaultNotAllowed ( self ) : <EOL> """<STR_LIT>""" <EOL> value = self . assertRaises ( <EOL> ValueError , endpoints . serverFromString , None , "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> str ( value ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_unknownType ( self ) : <EOL> """<STR_LIT>""" <EOL> value = self . assertRaises ( <EOL> ValueError , endpoints . serverFromString , None , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> str ( value ) , <EOL> "<STR_LIT>" ) <EOL> def test_typeFromPlugin ( self ) : <EOL> """<STR_LIT>""" <EOL> addFakePlugin ( self ) <EOL> notAReactor = object ( ) <EOL> fakeEndpoint = endpoints . serverFromString ( <EOL> notAReactor , "<STR_LIT>" ) <EOL> from twisted . plugins . fakeendpoint import fake <EOL> self . assertIdentical ( fakeEndpoint . parser , fake ) <EOL> self . assertEqual ( fakeEndpoint . args , ( notAReactor , '<STR_LIT:hello>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( fakeEndpoint . kwargs , dict ( yes = '<STR_LIT>' , up = '<STR_LIT>' ) ) <EOL> def addFakePlugin ( testCase , dropinSource = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> import sys <EOL> savedModules = sys . modules . copy ( ) <EOL> savedPluginPath = plugins . __path__ <EOL> def cleanup ( ) : <EOL> sys . modules . clear ( ) <EOL> sys . modules . update ( savedModules ) <EOL> plugins . __path__ [ : ] = savedPluginPath <EOL> testCase . addCleanup ( cleanup ) <EOL> fp = FilePath ( testCase . mktemp ( ) ) <EOL> fp . createDirectory ( ) <EOL> getModule ( __name__ ) . filePath . sibling ( dropinSource ) . copyTo ( <EOL> fp . child ( dropinSource ) ) <EOL> plugins . __path__ . append ( fp . path ) <EOL> class ClientStringTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_tcp ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" ) <EOL> self . assertIsInstance ( client , endpoints . TCP4ClientEndpoint ) <EOL> self . assertIdentical ( client . _reactor , reactor ) <EOL> self . assertEqual ( client . _host , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _port , <NUM_LIT> ) <EOL> self . assertEqual ( client . _timeout , <NUM_LIT:7> ) <EOL> self . assertEqual ( client . _bindAddress , "<STR_LIT>" ) <EOL> def test_tcpPositionalArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" ) <EOL> self . assertIsInstance ( client , endpoints . TCP4ClientEndpoint ) <EOL> self . assertIdentical ( client . _reactor , reactor ) <EOL> self . assertEqual ( client . _host , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _port , <NUM_LIT> ) <EOL> self . assertEqual ( client . _timeout , <NUM_LIT:7> ) <EOL> self . assertEqual ( client . _bindAddress , "<STR_LIT>" ) <EOL> def test_tcpHostPositionalArg ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( client . _host , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _port , <NUM_LIT> ) <EOL> def test_tcpPortPositionalArg ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( client . _host , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _port , <NUM_LIT> ) <EOL> def test_tcpDefaults ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( client . _timeout , <NUM_LIT:30> ) <EOL> self . assertEqual ( client . _bindAddress , None ) <EOL> def test_unix ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" ) <EOL> self . assertIsInstance ( client , endpoints . UNIXClientEndpoint ) <EOL> self . assertIdentical ( client . _reactor , reactor ) <EOL> self . assertEqual ( client . _path , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _timeout , <NUM_LIT:9> ) <EOL> self . assertEqual ( client . _checkPID , True ) <EOL> def test_unixDefaults ( self ) : <EOL> """<STR_LIT>""" <EOL> client = endpoints . clientFromString ( object ( ) , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _timeout , <NUM_LIT:30> ) <EOL> self . assertEqual ( client . _checkPID , False ) <EOL> def test_unixPathPositionalArg ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" ) <EOL> self . assertIsInstance ( client , endpoints . UNIXClientEndpoint ) <EOL> self . assertIdentical ( client . _reactor , reactor ) <EOL> self . assertEqual ( client . _path , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _timeout , <NUM_LIT:9> ) <EOL> self . assertEqual ( client . _checkPID , True ) <EOL> def test_typeFromPlugin ( self ) : <EOL> """<STR_LIT>""" <EOL> addFakePlugin ( self ) <EOL> notAReactor = object ( ) <EOL> clientEndpoint = endpoints . clientFromString ( <EOL> notAReactor , "<STR_LIT>" ) <EOL> from twisted . plugins . fakeendpoint import fakeClient <EOL> self . assertIdentical ( clientEndpoint . parser , fakeClient ) <EOL> self . assertEqual ( clientEndpoint . args , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( clientEndpoint . kwargs , dict ( cee = '<STR_LIT>' , num = '<STR_LIT:1>' ) ) <EOL> def test_unknownType ( self ) : <EOL> """<STR_LIT>""" <EOL> value = self . assertRaises ( <EOL> ValueError , endpoints . clientFromString , None , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> str ( value ) , <EOL> "<STR_LIT>" ) <EOL> class SSLClientStringTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> if skipSSL : <EOL> skip = skipSSL <EOL> def test_ssl ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( escapedPEMPathName , <EOL> escapedPEMPathName , <EOL> escapedCAsPathName ) ) <EOL> self . assertIsInstance ( client , endpoints . SSL4ClientEndpoint ) <EOL> self . assertIdentical ( client . _reactor , reactor ) <EOL> self . assertEqual ( client . _host , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _port , <NUM_LIT> ) <EOL> self . assertEqual ( client . _timeout , <NUM_LIT:3> ) <EOL> self . assertEqual ( client . _bindAddress , "<STR_LIT>" ) <EOL> certOptions = client . _sslContextFactory <EOL> self . assertIsInstance ( certOptions , CertificateOptions ) <EOL> ctx = certOptions . getContext ( ) <EOL> self . assertIsInstance ( ctx , ContextType ) <EOL> self . assertEqual ( Certificate ( certOptions . certificate ) , <EOL> testCertificate ) <EOL> privateCert = PrivateCertificate ( certOptions . certificate ) <EOL> privateCert . _setPrivateKey ( KeyPair ( certOptions . privateKey ) ) <EOL> self . assertEqual ( privateCert , testPrivateCertificate ) <EOL> expectedCerts = [ <EOL> Certificate . loadPEM ( x . getContent ( ) ) for x in <EOL> [ casPath . child ( "<STR_LIT>" ) , casPath . child ( "<STR_LIT>" ) ] <EOL> if x . basename ( ) . lower ( ) . endswith ( '<STR_LIT>' ) <EOL> ] <EOL> self . assertEqual ( sorted ( ( Certificate ( x ) for x in certOptions . caCerts ) , <EOL> key = lambda cert : cert . digest ( ) ) , <EOL> sorted ( expectedCerts , <EOL> key = lambda cert : cert . digest ( ) ) ) <EOL> def test_sslPositionalArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( escapedPEMPathName , <EOL> escapedPEMPathName , <EOL> escapedCAsPathName ) ) <EOL> self . assertIsInstance ( client , endpoints . SSL4ClientEndpoint ) <EOL> self . assertIdentical ( client . _reactor , reactor ) <EOL> self . assertEqual ( client . _host , "<STR_LIT>" ) <EOL> self . assertEqual ( client . _port , <NUM_LIT> ) <EOL> self . assertEqual ( client . _timeout , <NUM_LIT:3> ) <EOL> self . assertEqual ( client . _bindAddress , "<STR_LIT>" ) <EOL> def test_unreadableCertificate ( self ) : <EOL> """<STR_LIT>""" <EOL> class UnreadableFilePath ( FilePath ) : <EOL> def getContent ( self ) : <EOL> data = FilePath . getContent ( self ) <EOL> if data == casPath . child ( "<STR_LIT>" ) . getContent ( ) : <EOL> raise IOError ( EPERM ) <EOL> else : <EOL> return data <EOL> casPathClone = casPath . child ( "<STR_LIT>" ) . parent ( ) <EOL> casPathClone . clonePath = UnreadableFilePath <EOL> self . assertEqual ( <EOL> [ Certificate ( x ) for x in endpoints . _loadCAsFromDir ( casPathClone ) ] , <EOL> [ Certificate . loadPEM ( casPath . child ( "<STR_LIT>" ) . getContent ( ) ) ] ) <EOL> def test_sslSimple ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> client = endpoints . clientFromString ( <EOL> reactor , "<STR_LIT>" ) <EOL> certOptions = client . _sslContextFactory <EOL> self . assertIsInstance ( certOptions , CertificateOptions ) <EOL> self . assertEqual ( certOptions . verify , False ) <EOL> ctx = certOptions . getContext ( ) <EOL> self . assertIsInstance ( ctx , ContextType ) <EOL> class AdoptedStreamServerEndpointTestCase ( ServerEndpointTestCaseMixin , <EOL> unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _createStubbedAdoptedEndpoint ( self , reactor , fileno , addressFamily ) : <EOL> """<STR_LIT>""" <EOL> e = endpoints . AdoptedStreamServerEndpoint ( reactor , fileno , addressFamily ) <EOL> e . _close = lambda fd : None <EOL> e . _setNonBlocking = lambda fd : None <EOL> return e <EOL> def createServerEndpoint ( self , reactor , factory ) : <EOL> """<STR_LIT>""" <EOL> fileno = <NUM_LIT:12> <EOL> addressFamily = AF_INET <EOL> endpoint = self . _createStubbedAdoptedEndpoint ( <EOL> reactor , fileno , addressFamily ) <EOL> address = IPv4Address ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> return ( endpoint , ( fileno , addressFamily , factory ) , address ) <EOL> def expectedServers ( self , reactor ) : <EOL> """<STR_LIT>""" <EOL> return reactor . adoptedPorts <EOL> def listenArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { } <EOL> def test_singleUse ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = MemoryReactor ( ) <EOL> endpoint = self . _createStubbedAdoptedEndpoint ( reactor , <NUM_LIT> , AF_INET ) <EOL> endpoint . listen ( object ( ) ) <EOL> d = self . assertFailure ( endpoint . listen ( object ( ) ) , error . AlreadyListened ) <EOL> def listenFailed ( ignored ) : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( reactor . adoptedPorts ) ) <EOL> d . addCallback ( listenFailed ) <EOL> return d <EOL> def test_descriptionNonBlocking ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = MemoryReactor ( ) <EOL> endpoint = self . _createStubbedAdoptedEndpoint ( reactor , <NUM_LIT> , AF_INET ) <EOL> events = [ ] <EOL> def setNonBlocking ( fileno ) : <EOL> events . append ( ( "<STR_LIT>" , fileno ) ) <EOL> endpoint . _setNonBlocking = setNonBlocking <EOL> d = endpoint . listen ( object ( ) ) <EOL> def listened ( ignored ) : <EOL> self . assertEqual ( [ ( "<STR_LIT>" , <NUM_LIT> ) ] , events ) <EOL> d . addCallback ( listened ) <EOL> return d <EOL> def test_descriptorClosed ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = MemoryReactor ( ) <EOL> endpoint = self . _createStubbedAdoptedEndpoint ( reactor , <NUM_LIT> , AF_INET ) <EOL> events = [ ] <EOL> def close ( fileno ) : <EOL> events . append ( ( "<STR_LIT>" , fileno , len ( reactor . adoptedPorts ) ) ) <EOL> endpoint . _close = close <EOL> d = endpoint . listen ( object ( ) ) <EOL> def listened ( ignored ) : <EOL> self . assertEqual ( [ ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:1> ) ] , events ) <EOL> d . addCallback ( listened ) <EOL> return d <EOL> class SystemdEndpointPluginTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> _parserClass = endpoints . _SystemdParser <EOL> def test_pluginDiscovery ( self ) : <EOL> """<STR_LIT>""" <EOL> parsers = list ( getPlugins ( <EOL> interfaces . IStreamServerEndpointStringParser ) ) <EOL> for p in parsers : <EOL> if isinstance ( p , self . _parserClass ) : <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( parsers , ) ) <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> parser = self . _parserClass ( ) <EOL> self . assertTrue ( verifyObject ( <EOL> interfaces . IStreamServerEndpointStringParser , parser ) ) <EOL> def _parseStreamServerTest ( self , addressFamily , addressFamilyString ) : <EOL> """<STR_LIT>""" <EOL> reactor = object ( ) <EOL> descriptors = [ <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> , <NUM_LIT:9> ] <EOL> index = <NUM_LIT:3> <EOL> parser = self . _parserClass ( ) <EOL> parser . _sddaemon = ListenFDs ( descriptors ) <EOL> server = parser . parseStreamServer ( <EOL> reactor , domain = addressFamilyString , index = str ( index ) ) <EOL> self . assertIdentical ( server . reactor , reactor ) <EOL> self . assertEqual ( server . addressFamily , addressFamily ) <EOL> self . assertEqual ( server . fileno , descriptors [ index ] ) <EOL> def test_parseStreamServerINET ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _parseStreamServerTest ( AF_INET , "<STR_LIT>" ) <EOL> def test_parseStreamServerINET6 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _parseStreamServerTest ( AF_INET6 , "<STR_LIT>" ) <EOL> def test_parseStreamServerUNIX ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> from socket import AF_UNIX <EOL> except ImportError : <EOL> raise unittest . SkipTest ( "<STR_LIT>" ) <EOL> else : <EOL> self . _parseStreamServerTest ( AF_UNIX , "<STR_LIT>" ) <EOL> class TCP6ServerEndpointPluginTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> _parserClass = endpoints . _TCP6ServerParser <EOL> def test_pluginDiscovery ( self ) : <EOL> """<STR_LIT>""" <EOL> parsers = list ( getPlugins ( <EOL> interfaces . IStreamServerEndpointStringParser ) ) <EOL> for p in parsers : <EOL> if isinstance ( p , self . _parserClass ) : <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( parsers , ) ) <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> parser = self . _parserClass ( ) <EOL> self . assertTrue ( verifyObject ( <EOL> interfaces . IStreamServerEndpointStringParser , parser ) ) <EOL> def test_stringDescription ( self ) : <EOL> """<STR_LIT>""" <EOL> ep = endpoints . serverFromString ( MemoryReactor ( ) , <EOL> "<STR_LIT>" ) <EOL> self . assertIsInstance ( ep , endpoints . TCP6ServerEndpoint ) <EOL> self . assertIsInstance ( ep . _reactor , MemoryReactor ) <EOL> self . assertEqual ( ep . _port , <NUM_LIT> ) <EOL> self . assertEqual ( ep . _backlog , <NUM_LIT:12> ) <EOL> self . assertEqual ( ep . _interface , '<STR_LIT>' ) <EOL> class StandardIOEndpointPluginTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> _parserClass = endpoints . _StandardIOParser <EOL> def test_pluginDiscovery ( self ) : <EOL> """<STR_LIT>""" <EOL> parsers = list ( getPlugins ( <EOL> interfaces . IStreamServerEndpointStringParser ) ) <EOL> for p in parsers : <EOL> if isinstance ( p , self . _parserClass ) : <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( parsers , ) ) <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> parser = self . _parserClass ( ) <EOL> self . assertTrue ( verifyObject ( <EOL> interfaces . IStreamServerEndpointStringParser , parser ) ) <EOL> def test_stringDescription ( self ) : <EOL> """<STR_LIT>""" <EOL> ep = endpoints . serverFromString ( MemoryReactor ( ) , "<STR_LIT>" ) <EOL> self . assertIsInstance ( ep , endpoints . StandardIOEndpoint ) <EOL> self . assertIsInstance ( ep . _reactor , MemoryReactor ) </s>
