o bo?@sdZddlmZddlmZddlmZmZmZm Z m Z m Z m Z m Z mZddlmZmZddlmZmZmZmZddlmZddlmZGd d d eZGd d d eZGd ddZGdddeZGdddeZdS)z( Tests for L{twisted.internet.testing}. ) verifyObject) IPv4Address) IAddress IConnector IConsumerIListeningPort IPushProducer IReactorSSL IReactorTCP IReactorUNIX ITransport) ClientFactoryFactory) MemoryReactorNonStreamingProducerRaisingMemoryReactorStringTransport)namedAny)TestCasec@seZdZdZddZddZddZdd Zd d Zd d Z ddZ ddZ ddZ ddZ ddZddZddZddZddZd d!Zd"d#Zd$d%Zd&d'Zd(S))StringTransportTestsz@ Tests for L{twisted.internet.testing.StringTransport}. cCs t|_dSN)r transportselfrD/usr/lib/python3/dist-packages/twisted/internet/test/test_testing.pysetUp&s zStringTransportTests.setUpcCs:|tt|j|tt|j|tt|jdS)zq L{StringTransport} instances provide L{ITransport}, L{IPushProducer}, and L{IConsumer}. N) assertTruerr rrrrrrrtest_interfaces)sz$StringTransportTests.test_interfacescCs>t}t}|j||||jj|||jj|dS)zz L{StringTransport.registerProducer} records the arguments supplied to it as instance attributes. N)objectrregisterProducerassertIsproducer streamingrr"r#rrrtest_registerProducer2s z*StringTransportTests.test_registerProducercCsLt}|j|d|t|jjtd||jj|||jjdS)zy L{StringTransport.registerProducer} raises L{RuntimeError} if a producer is already registered. TFN) rrr assertRaises RuntimeErrorr!r"rr#)rr"rrrtest_disallowedRegisterProducer=sz4StringTransportTests.test_disallowedRegisterProducercCsbt}t}|j|d|j||jj|j|d||jj|||jjdS)z L{StringTransport.unregisterProducer} causes the transport to forget about the registered producer and makes it possible to register a new one. FTN) rrr unregisterProducer assertIsNoner"r!rr#)r oldProducer newProducerrrrtest_unregisterProducerJs z,StringTransportTests.test_unregisterProducercCs|t|jjdS)z L{StringTransport.unregisterProducer} raises L{RuntimeError} if called when no producer is registered. N)r&r'rr)rrrrtest_invalidUnregisterProducerYsz3StringTransportTests.test_invalidUnregisterProducercCs||jjddS)zO L{StringTransport.producerState} is initially C{'producing'}. producingN) assertEqualr producerStaterrrrtest_initialProducerState`sz.StringTransportTests.test_initialProducerStatecC|j||jjddS)zy L{StringTransport.pauseProducing} changes the C{producerState} of the transport to C{'paused'}. pausedN)rpauseProducingr0r1rrrrtest_pauseProducingf z(StringTransportTests.test_pauseProducingcCs(|j|j||jjddS)z} L{StringTransport.resumeProducing} changes the C{producerState} of the transport to C{'producing'}. r/N)rr5resumeProducingr0r1rrrrtest_resumeProducingns  z)StringTransportTests.test_resumeProducingcCr3)z{ L{StringTransport.stopProducing} changes the C{'producerState'} of the transport to C{'stopped'}. stoppedN)r stopProducingr0r1rrrrtest_stopProducingwr7z'StringTransportTests.test_stopProducingcC|j|t|jjdS)zu L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport has been stopped. N)rr;r&r'r5rrrr test_stoppedTransportCannotPauser7z5StringTransportTests.test_stoppedTransportCannotPausecCr=)zv L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport has been stopped. N)rr;r&r'r8rrrr!test_stoppedTransportCannotResumer7z6StringTransportTests.test_stoppedTransportCannotResumecCr=)zz L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport is being disconnected. N)rloseConnectionr&r'r5rrrr&test_disconnectingTransportCannotPauser7z;StringTransportTests.test_disconnectingTransportCannotPausecCr=)z{ L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport is being disconnected. N)rr@r&r'r8rrrr'test_disconnectingTransportCannotResumer7zr?rArBrErIrLrOrQrrrrr!s*      rc@s@eZdZdZddZddZddZdd Zd d Zd d Z dS) ReactorTestszA Tests for L{MemoryReactor} and L{RaisingMemoryReactor}. cC(t}tt|tt|tt|dS)zt L{MemoryReactor} provides all of the attributes described by the interfaces it advertises. N)rrr r r )r memoryReactorrrrtest_memoryReactorProvides  z'ReactorTests.test_memoryReactorProvidescCrW)z{ L{RaisingMemoryReactor} provides all of the attributes described by the interfaces it advertises. N)rrr r r )rraisingReactorrrrtest_raisingReactorProvidesrZz(ReactorTests.test_raisingReactorProvidescCst}|ddt|ddtdfD]}tt||}tt|||j d||j dq| dt}tt||}tt|||j ddS)a 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. ztest.example.comi Ns /fake/path) r connectTCPr connectSSLrrgetDestinationrr0hostport connectUNIXname)rrX connectorrHrrrtest_connectDestinations    z$ReactorTests.test_connectDestinationcCst}|dt|dtdfD]}tt||}tt|||j d||j dq| dt}tt||}tt|||j ddS)a 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 Nz0.0.0.0s/path/to/socket) r listenTCPr listenSSLrrrFrr0r`ra listenUNIXrc)rrXrarHrrrtest_listenDefaultHosts     z#ReactorTests.test_listenDefaultHostcCPt}t}|||||||g||||gdS)z> Adding, removing, and listing readers works. N)rr addReaderr0 getReaders removeReader)rreaderreactorrrr test_readers   zReactorTests.test_readerscCrj)z> Adding, removing, and listing writers works. N)rr addWriterr0 getWriters removeWriter)rwriterrorrr test_writers"rqzReactorTests.test_writersN) rRrSrTrUrYr\rerirprvrrrrrVs   rVc@s0eZdZdZddZddZddZdd Zd S) TestConsumerzP A very basic test consumer for use with the NonStreamingProducerTests. cCsg|_d|_d|_dSr)writesr"producerStreamingrrrr__init__8s zTestConsumer.__init__cCs||_||_dS)z Registers a single producer with this consumer. Just keeps track of it. @param producer: The producer to register. @param streaming: Whether the producer is a streaming one or not. Nr"ryr$rrrr =s zTestConsumer.registerProducercCsd|_d|_dS)zC Forget the producer we had previously registered. Nr{rrrrr)Gs zTestConsumer.unregisterProducercCs|j|dS)zz Some data was written to the consumer: stores it for later use. @param data: The data to write. N)rxappend)rdatarrrwriteNszTestConsumer.writeN)rRrSrTrUrzr r)r~rrrrrw3s  rwc@s eZdZdZddZddZdS)NonStreamingProducerTestszF Tests for the L{NonStreamingProducer} to validate behaviour. cCst}t|}||d||j|||j|||jtdD]}| q%gd}| |j| |j| |j| |j || t|j dS)z When the L{NonStreamingProducer} has resumeProducing called 10 times, it writes the counter each time and then fails. F ) 0123456789N)rwrr r!r"consumerrCryranger8r*r0rxr&r')rrr"_expectedWritesrrrtest_producesOnly10Times\s       z2NonStreamingProducerTests.test_producesOnly10TimescCs4t}t|}||d||t|jdS)zb When the L{NonStreamingProducer} is paused, it raises a L{RuntimeError}. FN)rwrr r8r&r'r5)rrr"rrrtest_cannotPauseProductionvs  z4NonStreamingProducerTests.test_cannotPauseProductionN)rRrSrTrUrrrrrrrWs rc@seZdZdZddZddZddZdd Zd d Zd d Z ddZ ddZ ddZ ddZ ddZddZddZddZddZd S)!DeprecationTestsz8 Deprecations in L{twisted.test.proto_helpers}. cCsdd|j}||g}|t|dd|dt||||dd||t|dS)Nztwisted.internet.testing.rcategorymessage)rR flushWarningsr0DeprecationWarninglenassertInr!r)rtestobjnew_pathwarningsrrrhelpers  zDeprecationTests.helpercCddlm}||j|dS)Nr)AccumulatingProtocol)twisted.test.proto_helpersrrtest_accumulatingProtocol)rrrrrr z*DeprecationTests.test_accumulatingProtocolcCr)Nr)LineSendingProtocol)rrrtest_lineSendingProtocol)rrrrrrrz)DeprecationTests.test_lineSendingProtocolcCr)Nr)FakeDatagramTransport)rrrtest_fakeDatagramTransport)rrrrrrrz+DeprecationTests.test_fakeDatagramTransportcCr)Nr)r)rrrtest_stringTransport)rrrrrrrz%DeprecationTests.test_stringTransportcCr)Nr) StringTransportWithDisconnection)rrr%test_stringTransportWithDisconnection)rrrrrrs z6DeprecationTests.test_stringTransportWithDisconnectioncCr)Nr)StringIOWithoutClosing)rrrtest_stringIOWithoutClosing)rrrrrrrz,DeprecationTests.test_stringIOWithoutClosingcCr)Nr)_FakeConnector)rrrtest__fakeConnector)rrrrrrrz$DeprecationTests.test__fakeConnectorcCr)Nr) _FakePort)rrrtest__fakePort)rrrrrrrzDeprecationTests.test__fakePortcCr)Nr)r)rrrtest_memoryReactor)rrrrrrrz#DeprecationTests.test_memoryReactorcCr)Nr)MemoryReactorClock)rrrtest_memoryReactorClock)rrrrrrrz(DeprecationTests.test_memoryReactorClockcCr)Nr)r)rrrtest_raisingMemoryReactor)rrrrrrrz*DeprecationTests.test_raisingMemoryReactorcCr)Nr)r)rrrtest_nonStreamingProducer)rrrrrrrz*DeprecationTests.test_nonStreamingProducercCr)Nr)waitUntilAllDisconnected)rrrtest_waitUntilAllDisconnected)rrrrrrrz.DeprecationTests.test_waitUntilAllDisconnectedcCr)Nr)EventLoggingObserver)rrrtest_eventLoggingObserver)rrrrrrrz*DeprecationTests.test_eventLoggingObserverN)rRrSrTrUrrrrrrrrrrrrrrrrrrrrs" rN) rUzope.interface.verifyrtwisted.internet.addressrtwisted.internet.interfacesrrrrrr r r r twisted.internet.protocolr rtwisted.internet.testingrrrrtwisted.python.reflectrtwisted.trial.unittestrrrVrwrrrrrrs  ,   )j$.