[XMc@sdZddlmZddlmZmZmZmZmZm Z m Z m Z m Z ddl mZddlmZddlmZmZmZddlmZmZdefd YZd efd YZd S( s* Tests for L{twisted.test.proto_helpers}. i(t verifyObject( t ITransportt IPushProducert IConsumert IReactorTCPt IReactorSSLt IReactorUNIXtIAddresstIListeningPortt IConnector(t IPv4Address(tTestCase(tStringTransportt MemoryReactortRaisingMemoryReactor(t ClientFactorytFactorytStringTransportTestscBseZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZdZdZdZdZRS(sB Tests for L{twisted.test.proto_helpers.StringTransport}. cCst|_dS(N(R t transport(tself((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pytsetUpscCsO|jtt|j|jtt|j|jtt|jdS(sq L{StringTransport} instances provide L{ITransport}, L{IPushProducer}, and L{IConsumer}. N(t assertTrueRRRRR(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_interfacesscCsUt}t}|jj|||j|jj||j|jj|dS(sz L{StringTransport.registerProducer} records the arguments supplied to it as instance attributes. N(tobjectRtregisterProducertassertIdenticaltproducert streaming(RRR((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_registerProducer&s   cCsht}|jj|t|jt|jjtt|j|jj||j |jj dS(sy L{StringTransport.registerProducer} raises L{RuntimeError} if a producer is already registered. N( RRRtTruet assertRaisest RuntimeErrortFalseRRRR(RR((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_disallowedRegisterProducer2s  cCst}t}|jj|t|jj|j|jjd|jj|t|j|jj||j |jj dS(s L{StringTransport.unregisterProducer} causes the transport to forget about the registered producer and makes it possible to register a new one. N( RRRR tunregisterProducerRRtNoneRRR(Rt oldProducert newProducer((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_unregisterProducer?s   cCs|jt|jjdS(s L{StringTransport.unregisterProducer} raises L{RuntimeError} if called when no producer is registered. N(RRRR"(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_invalidUnregisterProducerOscCs|j|jjddS(sO L{StringTransport.producerState} is initially C{'producing'}. t producingN(t assertEqualRt producerState(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_initialProducerStateWscCs'|jj|j|jjddS(sy L{StringTransport.pauseProducing} changes the C{producerState} of the transport to C{'paused'}. tpausedN(RtpauseProducingR)R*(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_pauseProducing^s cCs4|jj|jj|j|jjddS(s} L{StringTransport.resumeProducing} changes the C{producerState} of the transport to C{'producing'}. R(N(RR-tresumeProducingR)R*(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_resumeProducinggs  cCs'|jj|j|jjddS(s{ L{StringTransport.stopProducing} changes the C{'producerState'} of the transport to C{'stopped'}. tstoppedN(Rt stopProducingR)R*(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_stopProducingqs cCs'|jj|jt|jjdS(su L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport has been stopped. N(RR2RRR-(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyt test_stoppedTransportCannotPausezs cCs'|jj|jt|jjdS(sv L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport has been stopped. N(RR2RRR/(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyt!test_stoppedTransportCannotResumes cCs'|jj|jt|jjdS(sz L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport is being disconnected. N(RtloseConnectionRRR-(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyt&test_disconnectingTransportCannotPauses cCs'|jj|jt|jjdS(s{ L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport is being disconnected. N(RR6RRR/(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyt'test_disconnectingTransportCannotResumes cCs7|j|jj|jj|j|jjdS(sv L{StringTransport.loseConnection} toggles the C{disconnecting} instance variable to C{True}. N(t assertFalseRt disconnectingR6R(R((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyt$test_loseConnectionSetsDisconnectings cCs)t}|jt|j|dS(s If a host address is passed to L{StringTransport.__init__}, that value is returned from L{StringTransport.getHost}. N(RRR tgetHost(Rtaddress((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_specifiedHostAddresss cCs,t}|jtd|j|dS(s If a peer address is passed to L{StringTransport.__init__}, that value is returned from L{StringTransport.getPeer}. t peerAddressN(RRR tgetPeer(RR=((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_specifiedPeerAddresss cCs#tj}|j|tdS(s If no host address is passed to L{StringTransport.__init__}, an L{IPv4Address} is returned from L{StringTransport.getHost}. N(R R<tassertIsInstanceR (RR=((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_defaultHostAddressscCs#tj}|j|tdS(s If no peer address is passed to L{StringTransport.__init__}, an L{IPv4Address} is returned from L{StringTransport.getPeer}. N(R R@RBR (RR=((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_defaultPeerAddresss(t__name__t __module__t__doc__RRRR!R&R'R+R.R0R3R4R5R7R8R;R>RARCRD(((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyRs(     t ReactorTestscBs2eZdZdZdZdZdZRS(sA Tests for L{MemoryReactor} and L{RaisingMemoryReactor}. cCs4t}tt|tt|tt|dS(st L{MemoryReactor} provides all of the attributes described by the interfaces it advertises. N(R RRRR(Rt memoryReactor((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_memoryReactorProvidess   cCs4t}tt|tt|tt|dS(s{ L{RaisingMemoryReactor} provides all of the attributes described by the interfaces it advertises. N(RRRRR(RtraisingReactor((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_raisingReactorProvidess   cCst}x|jddt|jddtdgD]R}tt||j}tt||j |j d|j |j dq=W|j dt}tt||j}tt||j |j ddS(s L{MemoryReactor.connectTCP}, L{MemoryReactor.connectSSL}, and L{MemoryReactor.connectUNIX} will return an L{IConnector} whose C{getDestination} method returns an L{IAddress} with attributes which reflect the values passed. stest.example.comi s /fake/pathN(R t connectTCPRt connectSSLR#RR tgetDestinationRt assertEqualsthosttportt connectUNIXtname(RRIt connectorR=((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_connectDestinations          cCst}x|jdt|jdtdgD]R}tt||j}tt||j |j d|j |j dq7W|j dt}tt||j}tt||j |j ddS(s L{MemoryReactor.listenTCP}, L{MemoryReactor.listenSSL} and L{MemoryReactor.listenUNIX} will return an L{IListeningPort} whose C{getHost} method returns an L{IAddress}; C{listenTCP} and C{listenSSL} will have a default host of C{'0.0.0.0'}, and a port that reflects the value passed, and C{listenUNIX} will have a name that reflects the path passed. i2 s0.0.0.0s/path/to/socketN(R t listenTCPRt listenSSLR#RRR<RRPRQRRt listenUNIXRT(RRIRRR=((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyttest_listenDefaultHosts       (RERFRGRJRLRVRZ(((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyRHs  N(RGtzope.interface.verifyRttwisted.internet.interfacesRRRRRRRRR ttwisted.internet.addressR ttwisted.trial.unittestR ttwisted.test.proto_helpersR R Rttwisted.internet.protocolRRRRH(((sE/usr/lib/python2.7/dist-packages/twisted/test/test_stringtransport.pyts@