[XMc@sdZddlmZddlmZddlmZddlmZm Z ddlm Z ddl m Z m Z ddlmZmZdd lmZmZdd lmZdd lmZdd lmZdd lmZedjjdZeejjdZ e j!ej"Z#e j!e j"Z$yvddl%m&Z&ddl'm(Z(m)Z)m*Z*m+Z+ddl,m-Z-e)j.ej/Z0e+j.ej/Z1e2Z3Wne4k rdZ3nXdefdYZ5de5fdYZ6defdYZ7dej8fdYZ9de:fdYZ;de;ej8fd YZ<d!e;ej8fd"YZ=d#e;ej8fd$YZ>d%ej8fd&YZ?d'ej8fd(YZ@d)d*ZAd+ej8fd,YZBd-ej8fd.YZCd/S(0s Test the C{I...Endpoint} implementations that wrap the L{IReactorTCP}, L{IReactorSSL}, and L{IReactorUNIX} interfaces found in L{twisted.internet.endpoints}. i(tEPERM(t implements(tunittest(terrort interfaces(t endpoints(t IPv4Addresst UNIXAddress(t ClientFactorytProtocol(t MemoryReactortRaisingMemoryReactor(tFailure(tplugins(t getModule(tFilePaths twisted.tests server.pemtfake_CAs(tmakeCertificate(tCertificateOptionst CertificatetKeyPairtPrivateCertificate(t ContextTypes.OpenSSL is required to construct SSL Endpointst TestProtocolcBs2eZdZdZdZdZdZRS(sx Protocol whose only function is to callback deferreds on the factory when it is connected or disconnected. cCsg|_g|_d|_dS(Ni(tdatatconnectionsLosttconnectionMadeCalls(tself((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt__init__0s  cCs|jd7_dS(Ni(R(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytconnectionMade5scCs|jj|dS(N(Rtappend(RR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt dataReceived9scCs|jj|dS(N(RR(Rtreason((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytconnectionLost=s(t__name__t __module__t__doc__RRRR!(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR*s    tTestHalfCloseableProtocolcBs6eZdZeejdZdZdZRS(s A Protocol that implements L{IHalfCloseableProtocol} and records that its C{readConnectionLost} and {writeConnectionLost} methods. cCs#tj|t|_t|_dS(N(RRtFalsetreadLostt writeLost(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRIs  cCs t|_dS(N(tTrueR'(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytreadConnectionLostOscCs t|_dS(N(R)R((R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytwriteConnectionLostSs( R"R#R$RRtIHalfCloseableProtocolRR*R+(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR%Bs    t TestFactorycBseZdZeZRS(s Simple factory to be used both when connecting and listening. It contains two deferreds which are called back when my protocol connects and disconnects. (R"R#R$Rtprotocol(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR-XstWrappingFactoryTestscBs_eZdZdZdZdZdZdZdZdZ dZ d Z RS( sS Test the behaviour of our ugly implementation detail C{_WrappingFactory}. csgdtfdY}tj|d}|jdj|jt}|jfd|S(s An exception raised in C{buildProtocol} of our wrappedFactory results in our C{onConnection} errback being fired. t BogusFactorycBseZdZdZRS(sZ A one off factory whose C{buildProtocol} raises an C{Exception}. cSstddS(NsMy protocol is poorly defined.(t ValueError(Rtaddr((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt buildProtocolss(R"R#R$R3(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR0nscsj|jdS(NsMy protocol is poorly defined.(sMy protocol is poorly defined.(t assertEqualstargs(te(R(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt|sN( RRt_WrappingFactorytNoneR3t assertFailuret _onConnectionR1t addCallback(RR0twftd((RsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_failedBuildProtocolhs   cCstjtd}|jd}|jd|jd|j|jj dg|jd|j|jj ddgdS(s The wrapped C{Protocol}'s C{dataReceived} will get called when our C{_WrappingProtocol}'s C{dataReceived} gets called. tfootbarN( RR8R-R9R3tmakeConnectionRR4t_wrappedProtocolR(RR=tp((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt test_wrappedProtocolDataReceiveds   cCsgtjtd}|jd}t}|j||j|j||j|j j|dS(so Our transport is properly hooked up to the wrappedProtocol when a connection is made. N( RR8R-R9R3tobjectRBR4t transportRC(RR=RDtdummyTransport((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_wrappedProtocolTransports   cCsTt}tj|d}|jd}|jd|j|jjdgdS(s Our wrappedProtocol's connectionLost method is called when L{_WrappingProtocol.connectionLost} is called. tfailN( R-RR8R9R3R!R4RCR(RttfR=RD((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt"test_wrappedProtocolConnectionLosts   csytjtd}ttjdd}|jd|gfd}|jj ||j |gdS(s Calls to L{_WrappingFactory.clientConnectionLost} should errback the L{_WrappingFactory._onConnection} L{Deferred} tstringRJcsj|dS(N(R(tf(terrors(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytgotErrorsN( RR8R-R9R Rt ConnectErrortclientConnectionFailedR;t addErrbackR4(RR=texpectedFailureRP((ROsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_clientConnectionFailedscCsDt}t}tj||}|jtjj|tdS(sx Our L{_WrappingProtocol} should be an L{IHalfCloseableProtocol} if the C{wrappedProtocol} is. N( RFR%Rt_WrappingProtocolR4RR,t providedByR)(RtcdthcpRD((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt"test_wrappingProtocolHalfCloseables   cCs;t}tjd|}|jtjj|tdS(s Our L{_WrappingProtocol} should not provide L{IHalfCloseableProtocol} if the C{WrappedProtocol} doesn't. N( RRRVR9R4RR,RWR&(RttpRD((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt%test_wrappingProtocolNotHalfCloseables cCs<t}tjd|}|j|j|jtdS(s~ L{_WrappingProtocol.readConnectionLost} should proxy to the wrapped protocol's C{readConnectionLost} N(R%RRVR9R*R4R'R)(RRYRD((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt&test_wrappedProtocolReadConnectionLosts  cCs<t}tjd|}|j|j|jtdS(s L{_WrappingProtocol.writeConnectionLost} should proxy to the wrapped protocol's C{writeConnectionLost} N(R%RRVR9R+R4R(R)(RRYRD((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt'test_wrappedProtocolWriteConnectionLosts  ( R"R#R$R?RERIRLRURZR\R]R^(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR/cs      tEndpointTestCaseMixincBsVeZdZdZdZdZdZdZdZdZ dZ RS( sJ Generic test methods to be mixed into all endpoint test classes. cCs|j|ddS(s Retrieve a single factory that has connected using the given reactor. (This behavior is valid for TCP and SSL but needs to be overridden for UNIX.) @param reactor: a L{MemoryReactor} ii(texpectedClients(Rtreactor((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytretrieveConnectedFactorysc st}t}t}|j||\}}}|j|}gfd}|j||j|} | jj||j|g|j |} |jt | d|j | d|dS(s} A client endpoint can connect and returns a deferred who gets called back with a protocol instance. csj|dS(N(R(RD(treceivedProtos(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt checkProtosiiN( RFR tcreateClientEndpointtconnectR<RbR;tcallbackR4R`tlentassertConnectArgs( Rtprototmreactort clientFactorytept expectedArgst ignoredDestR>RdtfactoryR`((RcsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_endpointConnectSuccesss    c stjdd}td|}t}|j||\}}}|j|}gfd}|j||j|gdS(st If an endpoint tries to connect to a non-listening port it gets a C{ConnectError} failure. RMsConnection FailedtconnectExceptioncsj|jdS(N(Rtvalue(RN(treceivedExceptions(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt checkFailure7sN(RRQR RFReRfRSR4( Rt expectedErrorRkRlRmt ignoredArgsRoR>Ru((RtsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_endpointConnectFailure%s  c st}t}|j||\}}}|j|}gfd}|j||j|jtdd}|j|j t j |j|j j |dS(s Calling L{Deferred.cancel} on the L{Deferred} returned from L{IStreamClientEndpoint.connect} is errbacked with an expected L{ConnectingCancelledError} exception. csj|dS(N(R(RN(treceivedFailures(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRuPsiiN( R RFReRfRStcancelR4RhtassertIsInstanceRsRtConnectingCancelledErrortaddress( RRkRlRmRwR}R>Rutfailure((RysH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt test_endpointConnectingCancelled?s     cst}t}|j||\}}}|j|}gfd}|j||j|g|j|j||gdS(ss An endpoint can listen and returns a deferred that gets called back with a port instance. csj|jdS(N(RtgetHost(tport(t receivedHosts(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytcheckPortAndServerosN(R RFtcreateServerEndpointtlistenR<R4texpectedServers(RRkRpRmRnt expectedHostR>R((RsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_endpointListenSuccess_s   c st}tjdd|}td|}|j||\}}}|jt}gfd}|j||j|gdS(s When an endpoint tries to listen on an already listening port, a C{CannotListenError} failure is errbacked. tiPtlistenExceptioncsj|jdS(N(RRs(RN(Rt(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRusN(RFRtCannotListenErrorR RRRSR4( RRpt exceptionRkRmRwRoR>Ru((RtsH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_endpointListenFailurexs  cCst}t}|j|||j\}}}|j||j|}|jt|d|j|d|dS(sn The endpoint should pass it's connectArgs parameter to the reactor's listen methods. iiN( RFR Ret connectArgsRfR`R4RhRi(RRpRkRmRnt ignoredHostR`((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt"test_endpointConnectNonDefaultArgss   cCsit}t}|j|||j\}}}|j||j|}|j||gdS(sm The endpoint should pass it's listenArgs parameter to the reactor's listen methods. N(RFR Rt listenArgsRRR4(RRpRkRmRnRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt!test_endpointListenNonDefaultArgss   ( R"R#R$RbRqRxRRRRR(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR_s    tTCP4EndpointsTestCasecBsMeZdZdZdZdZdZdZdZdZ RS(s" Tests for TCP Endpoints. cCs|jS(sD @return: List of calls to L{IReactorTCP.listenTCP} (t tcpServers(RRa((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRscCs|jS(sE @return: List of calls to L{IReactorTCP.connectTCP} (t tcpClients(RRa((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR`sc Csn|\}}}}}|\}} } } } |j|||j|| |j|| |j|| dS(s] Compare host, port, timeout, and bindAddress in C{receivedArgs} to C{expectedArgs}. We ignore the factory because we don't only care what protocol comes out of the C{IStreamClientEndpoint.connect} call. @param receivedArgs: C{tuple} of (C{host}, C{port}, C{factory}, C{timeout}, C{bindAddress}) that was passed to L{IReactorTCP.connectTCP}. @param expectedArgs: C{tuple} of (C{host}, C{port}, C{factory}, C{timeout}, C{bindAddress}) that we expect to have been passed to L{IReactorTCP.connectTCP}. N(R4( Rt receivedArgsRnthostRtignoredFactoryttimeoutt bindAddressRt expectedPortt_ignoredFactorytexpectedTimeouttexpectedBindAddress((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRis cCsidd6dd6S(sK @return: C{dict} of keyword arguments to pass to connect. i Rt localhostiR(Ri((R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRscCsidd6dd6S(sI @return: C{dict} of keyword arguments to pass to listen idtbacklogs 127.0.0.1t interface((R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRscKsmtddd}|dkr'i}ntj||j||j||jdd|jddf|fS( s Create an L{TCP4ServerEndpoint} and return the values needed to verify its behaviour. @param reactor: A fake L{IReactorTCP} that L{TCP4ServerEndpoint} can call L{IReactorTCP.listenTCP} on. @param factory: The thing that we expect to be passed to our L{IStreamServerEndpoint.listen} implementation. @param listenArgs: Optional dictionary of arguments to L{IReactorTCP.listenTCP}. tTCPs0.0.0.0iRi2RRN(RR9RtTCP4ServerEndpointRtget(RRaRpRR}((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs     cKsdtddd}tj||j|j||j|j||jdd|jddf|fS(s Create an L{TCP4ClientEndpoint} and return the values needed to verify its behavior. @param reactor: A fake L{IReactorTCP} that L{TCP4ClientEndpoint} can call L{IReactorTCP.connectTCP} on. @param clientFactory: The thing that we expect to be passed to our L{IStreamClientEndpoint.connect} implementation. @param connectArgs: Optional dictionary of arguments to L{IReactorTCP.connectTCP} RRiPRiRN(RRtTCP4ClientEndpointRRRR9(RRaRlRR}((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRes  ( R"R#R$RR`RiRRRRe(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs      tSSL4EndpointsTestCasecBseeZdZereZndZdZdZdZdZ dZ dZ dZ RS( s" Tests for SSL Endpoints. cCs|jS(sD @return: List of calls to L{IReactorSSL.listenSSL} (t sslServers(RRa((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR/scCs|jS(sE @return: List of calls to L{IReactorSSL.connectSSL} (t sslClients(RRa((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR`6scCs|\}}}}}}|\} } } } } }|j|| |j|| |j|| |j|| |j||dS(s Compare host, port, contextFactory, timeout, and bindAddress in C{receivedArgs} to C{expectedArgs}. We ignore the factory because we don't only care what protocol comes out of the C{IStreamClientEndpoint.connect} call. @param receivedArgs: C{tuple} of (C{host}, C{port}, C{factory}, C{contextFactory}, C{timeout}, C{bindAddress}) that was passed to L{IReactorSSL.connectSSL}. @param expectedArgs: C{tuple} of (C{host}, C{port}, C{factory}, C{contextFactory}, C{timeout}, C{bindAddress}) that we expect to have been passed to L{IReactorSSL.connectSSL}. N(R4(RRRnRRRtcontextFactoryRRRRRtexpectedContextFactoryRR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRi=scCsidd6dd6S(sK @return: C{dict} of keyword arguments to pass to connect. i RRiR(s localhosti((R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRXscCsidd6dd6S(sI @return: C{dict} of keyword arguments to pass to listen idRs 127.0.0.1R((R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR_scCs|tdddd\|_|_tdddd\|_|_td|jd|jd t|_td t|_d S( sF Set up client and server SSL contexts for use later. tOsServer Test CertificatetCNtserversClient Test Certificatetclientt privateKeyt certificatetrequireCertificateN( RtsKeytsCerttcKeytcCertRR&tserverSSLContexttclientSSLContext(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytsetUpfs   cKsdtddd}tj||j|j||j||j|jdd|jddf|fS(s Create an L{SSL4ServerEndpoint} and return the tools to verify its behaviour. @param factory: The thing that we expect to be passed to our L{IStreamServerEndpoint.listen} implementation. @param reactor: A fake L{IReactorSSL} that L{SSL4ServerEndpoint} can call L{IReactorSSL.listenSSL} on. @param listenArgs: Optional dictionary of arguments to L{IReactorSSL.listenSSL}. Rs0.0.0.0iRi2RR(RRtSSL4ServerEndpointRRR(RRaRpRR}((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRxs  c Kstddd}|dkr'i}ntj||j|j|j||j|j||j|jdd|jddf|fS(s Create an L{SSL4ClientEndpoint} and return the values needed to verify its behaviour. @param reactor: A fake L{IReactorSSL} that L{SSL4ClientEndpoint} can call L{IReactorSSL.connectSSL} on. @param clientFactory: The thing that we expect to be passed to our L{IStreamClientEndpoint.connect} implementation. @param connectArgs: Optional dictionary of arguments to L{IReactorSSL.connectSSL} RRiPRiRN(RR9RtSSL4ClientEndpointRRRR(RRaRlRR}((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRes    ( R"R#R$tskipSSLtskipRR`RiRRRRRe(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR's        tUNIXEndpointsTestCasecBsVeZdZdZdZdZdZdZdZdZ dZ RS( s) Tests for UnixSocket Endpoints. cCs|j|ddS(s Override L{EndpointTestCaseMixin.retrieveConnectedFactory} to account for different index of 'factory' in C{connectUNIX} args. ii(R`(RRa((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRbscCs|jS(sF @return: List of calls to L{IReactorUNIX.listenUNIX} (t unixServers(RRa((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRscCs|jS(sG @return: List of calls to L{IReactorUNIX.connectUNIX} (t unixClients(RRa((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR`sc CsX|\}}}}|\}}} } |j|||j|| |j|| dS(s Compare path, timeout, checkPID in C{receivedArgs} to C{expectedArgs}. We ignore the factory because we don't only care what protocol comes out of the C{IStreamClientEndpoint.connect} call. @param receivedArgs: C{tuple} of (C{path}, C{timeout}, C{checkPID}) that was passed to L{IReactorUNIX.connectUNIX}. @param expectedArgs: C{tuple} of (C{path}, C{timeout}, C{checkPID}) that we expect to have been passed to L{IReactorUNIX.connectUNIX}. N(R4( RRRntpathRRtcheckPIDt expectedPathRRtexpectedCheckPID((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRis cCsidd6dd6S(sK @return: C{dict} of keyword arguments to pass to connect. i RiR((R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRscCsidd6dd6dd6S(sI @return: C{dict} of keyword arguments to pass to listen idRitmodeitwantPID((R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRscKsgt|j}tj||j||j||jdd|jdd|jddf|fS(s Create an L{UNIXServerEndpoint} and return the tools to verify its behaviour. @param reactor: A fake L{IReactorUNIX} that L{UNIXServerEndpoint} can call L{IReactorUNIX.listenUNIX} on. @param factory: The thing that we expect to be passed to our L{IStreamServerEndpoint.listen} implementation. @param listenArgs: Optional dictionary of arguments to L{IReactorUNIX.listenUNIX}. Ri2RiRi(RtmktempRtUNIXServerEndpointtnameR(RRaRpRR}((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs  cKsXt|j}tj||j||j||jdd|jddf|fS(s Create an L{UNIXClientEndpoint} and return the values needed to verify its behaviour. @param reactor: A fake L{IReactorUNIX} that L{UNIXClientEndpoint} can call L{IReactorUNIX.connectUNIX} on. @param clientFactory: The thing that we expect to be passed to our L{IStreamClientEndpoint.connect} implementation. @param connectArgs: Optional dictionary of arguments to L{IReactorUNIX.connectUNIX} RiRi(RRRtUNIXClientEndpointRR(RRaRlRR}((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRes  ( R"R#R$RbRR`RiRRRRe(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs       tParserTestCasecBseZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZRS(sK Tests for L{endpoints._parseServer}, the low-level parsing logic. tFactorycOstj||S(sT Provide a hook for test_strports to substitute the deprecated API. (Rt _parseServer(Rtatkw((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytparse"scCsC|j|jd|jdd|jfidd6dd6fdS( sN Simple strings with a 'tcp:' prefix should be parsed as TCP. stcp:80RiPRRi2RN(R4RRN(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_simpleTCP)scCsC|j|jd|jdd|jfidd6dd6fdS( sU TCP port descriptions parse their 'interface' argument as a string. stcp:80:interface=127.0.0.1RiPs 127.0.0.1Ri2RN(R4RRN(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_interfaceTCP1scCsC|j|jd|jdd|jfidd6dd6fdS( sU TCP port descriptions parse their 'backlog' argument as an integer. stcp:80:backlog=6RiPRRiRN(R4RRN(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_backlogTCP:scCsJ|j|jd|jdd|jfidd6dd6td6fd S( s L{endpoints._parseServer} returns a C{'UNIX'} port description with defaults for C{'mode'}, C{'backlog'}, and C{'wantPID'} when passed a string with the C{'unix:'} prefix and no other parameter values. sunix:/var/run/fingertUNIXs/var/run/fingeriRi2RRN(R4RRNR)(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_simpleUNIXCscCsJ|j|jd|jdd|jfidd6dd6td6fd S( sK C{mode} can be set by including C{"mode="}. sunix:/var/run/finger:mode=0660Rs/var/run/fingeriRi2RRN(R4RRNR)(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt test_modeUNIXOscCsJ|j|jd|jdd|jfidd6dd6td6fd S( sM C{wantPID} can be set to false by included C{"lockfile=0"}. sunix:/var/run/finger:lockfile=0Rs/var/run/fingeriRi2RRN(R4RRNR&(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_wantPIDUNIXYscCsJ|j|jd|jdd|jfidd6dd6td6fd S( sf Backslash can be used to escape colons and backslashes in port descriptions. sunix:foo\:bar\=baz\:qux\\Rsfoo:bar=baz:qux\iRi2RRN(R4RRNR)(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt test_escapecscCs|jtjdddS(s L{endpoints.quoteStringArgument} should quote backslashes and colons for interpolation into L{endpoints.serverFromString} and L{endpoints.clientFactory} arguments. ssome : stuff \ssome \: stuff \\N(R4RtquoteStringArgument(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_quoteStringArgumentnscCsJ|j|jd|jdd|jfidd6dd6td6fd S( s In strports descriptions, '=' in a parameter value does not need to be quoted; it will simply be parsed as part of the value. sunix:address=foo=barRsfoo=bariRi2RRN(R4RRNR)(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_impliedEscapexscCsM|j|jd|jddd|jfidd6dd6td6fd S( s For compatibility with the old L{twisted.application.strports.parse}, the third 'mode' argument may be specified to L{endpoints.parse} to indicate a default other than TCP. tfilenametunixRiRi2RRN(R4RRNR)(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_nonstandardDefaultscCs |jt|jd|jdS(se L{strports.parse} raises C{ValueError} when given an unknown endpoint type. sbogus-type:nothingN(t assertRaisesR1RRN(R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_unknownTypes(R"R#R$RNRRRRRRRRRRRR(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs   tServerStringTestscBsVeZdZdZdZer0ee_ndZdZdZ dZ RS(sC Tests for L{twisted.internet.endpoints.serverFromString}. cCs~t}tj|d}|j|tj|j|j||j|jd|j|j d|j|j ddS(s When passed a TCP strports description, L{endpoints.serverFromString} returns a L{TCP4ServerEndpoint} instance initialized with the values from the string. s&tcp:1234:backlog=12:interface=10.0.0.1ii s10.0.0.1N( RFRtserverFromStringR{RtassertIdenticalt_reactorR4t_portt_backlogt _interface(RRaR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_tcps  cCst}tj|dttf}|j|tj|j|j||j|j d|j|j d|j|j d|j j }|j|tdS(s When passed an SSL strports description, L{endpoints.serverFromString} returns a L{SSL4ServerEndpoint} instance initialized with the values from the string. s?ssl:1234:backlog=12:privateKey=%s:certKey=%s:interface=10.0.0.1ii s10.0.0.1N(RFRRtescapedPEMPathNameR{RRRR4RRRt_sslContextFactoryt getContextR(RRaRtctx((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_ssls  cCst}tj|d}|j|tj|j|j||j|jd|j|j d|j|j d|j|j t dS(s When passed a UNIX strports description, L{endpoint.serverFromString} returns a L{UNIXServerEndpoint} instance initialized with the values from the string. s0unix:/var/foo/bar:backlog=7:mode=0123:lockfile=1s /var/foo/bariiSN( RFRRR{RRRR4t_addressRt_modet_wantPIDR)(RRatendpoint((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyt test_unixs  cCs5|jttjdd}|jt|ddS(s The older service-based API (L{twisted.internet.strports.service}) allowed an implicit default of 'tcp' so that TCP ports could be specified as a simple integer, but we've since decided that's a bad idea, and the new API does not accept an implicit default argument; you have to say 'tcp:' now. If you try passing an old implicit port number to the new API, you'll get a C{ValueError}. t4321sqUnqualified strport description passed to 'service'.Use qualified endpoint descriptions; for example, 'tcp:4321'.N(RR1RRR9R4tstr(RRs((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_implicitDefaultNotAlloweds  cCs5|jttjdd}|jt|ddS(sq L{endpoints.serverFromString} raises C{ValueError} when given an unknown endpoint type. sftl:andromeda/carcosa/hali/2387sUnknown endpoint type: 'ftl'N(RR1RRR9R4R(RRs((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs    cCst|t}tj|d}ddlm}|j|j||j|j |ddf|j|j t dddd d S( s L{endpoints.serverFromString} looks up plugins of type L{IStreamServerEndpoint} and constructs endpoints from them. sfake:hello:world:yes=no:up=downi(tfakethellotworldtyestnotuptdownN( t addFakePluginRFRRttwisted.plugins.fakeendpointRRtparserR4R5tkwargstdict(Rt notAReactort fakeEndpointR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_typeFromPlugins   ( R"R#R$RRRRRRRR(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs      sfakeendpoint.pycsddljjtjfd}|j|t|j}|jt t j j |j |j|tjj|jdS(s For the duration of C{testCase}, add a fake plugin to twisted.plugins which contains some sample endpoint parsers. iNcs+jjjjtj(dS(N(tmodulestcleartupdateR t__path__((tsyst savedModulestsavedPluginPath(sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytcleanups (RRtcopyR Rt addCleanupRRtcreateDirectoryRR"tfilePathtsiblingtcopyTotchildRR(ttestCaset dropinSourceR tfp((RR R sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR s    tClientStringTestscBsDeZdZdZdZdZdZdZdZRS(sC Tests for L{twisted.internet.endpoints.clientFromString}. cCst}tj|d}|j|tj|j|j||j|jd|j|j d|j|j d|j|j ddS(s When passed a TCP strports description, L{endpointClient} returns a L{TCP4ClientEndpoint} instance initialized with the values from the string. s=tcp:host=example.com:port=1234:timeout=7:bindAddress=10.0.0.2s example.comiis10.0.0.2N( RFRtclientFromStringR{RRRR4t_hostRt_timeoutt _bindAddress(RRaR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR#s  cCsEt}tj|d}|j|jd|j|jddS(s{ A TCP strports description may omit I{timeout} or I{bindAddress} to allow the default to be used. stcp:host=example.com:port=1234iN(RFRRR4RRR9(RRaR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_tcpDefaults5s   cCs~t}tj|d}|j|tj|j|j||j|jd|j|j d|j|j t dS(s When passed a UNIX strports description, L{endpointClient} returns a L{UNIXClientEndpoint} instance initialized with the values from the string. s+unix:path=/var/foo/bar:lockfile=1:timeout=9s /var/foo/bari N( RFRRR{RRRR4t_pathRt _checkPIDR)(RRaR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRBs  cCs?tjtd}|j|jd|j|jtdS(sz A UNIX strports description may omit I{lockfile} or I{timeout} to allow the defaults to be used. sunix:path=/var/foo/bariN(RRRFR4RRR&(RR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_unixDefaultsSscCst|t}tj|d}ddlm}|j|j||j|j d |j|j t dddd d S( s L{endpoints.clientFromString} looks up plugins of type L{IStreamClientEndpoint} and constructs endpoints from them. scfake:alpha:beta:cee=dee:num=1i(t fakeClienttalphatbetatceetdeetnumt1N(R R!( RRFRRRRRRR4R5RR(RRtclientEndpointR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR]s   cCs5|jttjdd}|jt|ddS(sq L{endpoints.serverFromString} raises C{ValueError} when given an unknown endpoint type. sftl:andromeda/carcosa/hali/2387sUnknown endpoint type: 'ftl'N(RR1RRR9R4R(RRs((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRls    ( R"R#R$RRRRRR(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs   tSSLClientStringTestscBs8eZdZereZndZdZdZRS(sU Tests for L{twisted.internet.endpoints.clientFromString} which require SSL. cCst}tj|dtttf}|j|tj|j|j||j |j d|j |j d|j |j d|j |j d|j}|j|t|j}|j|t|j t|jtt|j}|jt|j|j |tgtjdtjdgD]6}|jjjdr<tj|j ^q<}|j g|j!D]}t|^q|d S( s When passed an SSL strports description, L{clientFromString} returns a L{SSL4ClientEndpoint} instance initialized with the values from the string. sdssl:host=example.net:port=4321:privateKey=%s:certKey=%s:bindAddress=10.0.0.3:timeout=3:caCertsDir=%ss example.netiis10.0.0.3s thing1.pems thing2.pems.pemN("RFRRRtescapedCAsPathNameR{RRRR4RRRRRRRRRRttestCertificateRt_setPrivateKeyRRttestPrivateCertificatetcasPathRtbasenametlowertendswithtloadPEMt getContenttcaCerts(RRaRt certOptionsRt privateCerttxt expectedCerts((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyRs6    "6%cCsdtfdY}tjdj}||_|jgtj|D]}t|^qJtj tjdj gdS(s If a certificate in the directory is unreadable, L{endpoints._loadCAsFromDir} will ignore that certificate. tUnreadableFilePathcBseZdZRS(cSsAtj|}|tjdjkr9ttn|SdS(Ns thing2.pem(RR1R,RtIOErrorR(RR((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR1s(R"R#R1(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR7stignoreds thing1.pemN( RR,Rtparentt clonePathR4Rt_loadCAsFromDirRR0R1(RR7t casPathCloneR5((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_unreadableCertificates   %cCsgt}tj|d}|j}|j|t|j|jt|j }|j|t dS(s When passed an SSL strports description without any extra parameters, L{clientFromString} returns a simple non-verifying endpoint that will speak SSL. s%ssl:host=simple.example.org:port=4321N( RFRRRR{RR4tverifyR&RR(RRaRR3R((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyttest_sslSimples    (R"R#R$RRRR>R@(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pyR'{s   & N(DR$terrnoRtzope.interfaceRt twisted.trialRttwisted.internetRRRttwisted.internet.addressRRttwisted.internet.protocolRR ttwisted.test.proto_helpersR R ttwisted.python.failureR ttwistedR ttwisted.python.modulesRttwisted.python.filepathRRRtpemPathR"R,RRRR(ttwisted.test.test_sslverifyRttwisted.internet.sslRRRRt OpenSSL.SSLRR0R1R)R+R&Rt ImportErrorRR%R-tTestCaseR/RFR_RRRRRRRR'(((sH/usr/lib/python2.7/dist-packages/twisted/internet/test/test_endpoints.pytsR"    ll}q ]