U J]jA@sdZddlmZddlmZddlmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZmZddlmZmZdd lmZmZmZmZdd lmZdd lm Z Gd d d e Z!Gddde Z"GdddZ#Gddde Z$GdddeZ%Gddde Z&dS)z( Tests for L{twisted.internet.testing}. ) annotations)Callable) verifyObject)Protocol) IPv4Address) IAddress IConnector IConsumerIListeningPort IPushProducer IReactorSSL IReactorTCP IReactorUNIX ITransport) ClientFactoryFactory) MemoryReactorNonStreamingProducerRaisingMemoryReactorStringTransport)namedAny)TestCasec@seZdZdZddddZddddZdddd Zddd d Zddd d ZddddZ ddddZ ddddZ ddddZ ddddZ ddddZddddZddddZddddZddd d!Zddd"d#Zddd$d%Zddd&d'Zddd(d)Zd*S)+StringTransportTestsz@ Tests for L{twisted.internet.testing.StringTransport}. NonereturncCs t|_dSN)r transportselfr kc:\users\psant\downloads\hotstarbackend\school\venv\Lib\site-packages\twisted/internet/test/test_testing.pysetUp+szStringTransportTests.setUpcCs:|tt|j|tt|j|tt|jdS)zq L{StringTransport} instances provide L{ITransport}, L{IPushProducer}, and L{IConsumer}. N) assertTruerrrr r rr r r!test_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)r r r!test_registerProducer7s z*StringTransportTests.test_registerProducercCsLt}|j|d|t|jjtd||jj|||jjdS)zy L{StringTransport.registerProducer} raises L{RuntimeError} if a producer is already registered. TFN) r%rr& assertRaises RuntimeErrorr'r(r#r))rr(r r r!test_disallowedRegisterProducerBsz4StringTransportTests.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) r%rr&unregisterProducer assertIsNoner(r'r#r))rZ oldProducerZ newProducerr r r!test_unregisterProducerOs z,StringTransportTests.test_unregisterProducercCs|t|jjdS)z L{StringTransport.unregisterProducer} raises L{RuntimeError} if called when no producer is registered. N)r,r-rr/rr r r!test_invalidUnregisterProducer^sz3StringTransportTests.test_invalidUnregisterProducercCs||jjddS)zO L{StringTransport.producerState} is initially C{'producing'}. producingN) assertEqualr producerStaterr r r!test_initialProducerStateesz.StringTransportTests.test_initialProducerStatecCs|j||jjddS)zy L{StringTransport.pauseProducing} changes the C{producerState} of the transport to C{'paused'}. pausedN)rpauseProducingr4r5rr r r!test_pauseProducingks z(StringTransportTests.test_pauseProducingcCs(|j|j||jjddS)z} L{StringTransport.resumeProducing} changes the C{producerState} of the transport to C{'producing'}. r3N)rr8resumeProducingr4r5rr r r!test_resumeProducingss  z)StringTransportTests.test_resumeProducingcCs|j||jjddS)z{ L{StringTransport.stopProducing} changes the C{'producerState'} of the transport to C{'stopped'}. stoppedN)r stopProducingr4r5rr r r!test_stopProducing|s z'StringTransportTests.test_stopProducingcCs|j|t|jjdS)zu L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport has been stopped. N)rr=r,r-r8rr r r! test_stoppedTransportCannotPauses z5StringTransportTests.test_stoppedTransportCannotPausecCs|j|t|jjdS)zv L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport has been stopped. N)rr=r,r-r:rr r r!!test_stoppedTransportCannotResumes z6StringTransportTests.test_stoppedTransportCannotResumecCs|j|t|jjdS)zz L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport is being disconnected. N)rloseConnectionr,r-r8rr r r!&test_disconnectingTransportCannotPauses z;StringTransportTests.test_disconnectingTransportCannotPausecCs|j|t|jjdS)z{ L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport is being disconnected. N)rrAr,r-r:rr r r!'test_disconnectingTransportCannotResumes zr?r@rBrCrErIrKrMrNr r r r!r&s(     rc@sdeZdZdZddddZddddZdddd Zddd d Zddd d ZddddZ dS) ReactorTestszA Tests for L{MemoryReactor} and L{RaisingMemoryReactor}. rrcCs(t}tt|tt|tt|dS)zt L{MemoryReactor} provides all of the attributes described by the interfaces it advertises. N)rrr r r)r memoryReactorr r r!test_memoryReactorProvidess  z'ReactorTests.test_memoryReactorProvidescCs(t}tt|tt|tt|dS)z{ L{RaisingMemoryReactor} provides all of the attributes described by the interfaces it advertises. N)rrr r r)rZraisingReactorr r r!test_raisingReactorProvidess  z(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) rZ connectTCPrZ connectSSLrrZgetDestinationrr4hostportZ connectUNIXname)rrTZ connectorrHr r r!test_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) rZ listenTCPrZ listenSSLrr rFrr4rWrXZ listenUNIXrY)rrTrXrHr r r!test_listenDefaultHosts      z#ReactorTests.test_listenDefaultHostcCsPt}t}|||||||g||||gdS)z> Adding, removing, and listing readers works. N)r%rZ addReaderr4Z getReadersZ removeReader)rreaderreactorr r r! test_readerss   zReactorTests.test_readerscCsPt}t}|||||||g||||gdS)z> Adding, removing, and listing writers works. N)r%rZ addWriterr4Z getWritersZ removeWriter)rwriterr]r r r! test_writers's   zReactorTests.test_writersN) rOrPrQrRrUrVrZr[r^r`r r r r!rSs  rSc@sNeZdZdZddddZddddd d Zddd d Zd ddddZdS) TestConsumerzP A very basic test consumer for use with the NonStreamingProducerTests. rrcCsg|_d|_d|_dSr)writesr(producerStreamingrr r r!__init__=szTestConsumer.__init__r%bool)r(r)rcCs||_||_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(rcr*r r r!r&BszTestConsumer.registerProducercCsd|_d|_dS)zC Forget the producer we had previously registered. Nrfrr r r!r/LszTestConsumer.unregisterProducerbytes)datarcCs|j|dS)zz Some data was written to the consumer: stores it for later use. @param data: The data to write. N)rbappend)rrhr r r!writeSszTestConsumer.writeN)rOrPrQrRrdr&r/rjr r r r!ra8s  rac@s,eZdZdZddddZddddZdS) NonStreamingProducerTestszF Tests for the L{NonStreamingProducer} to validate behaviour. rrc Cst}t|}||d||j|||j|||jtdD] }| qJddddddd d d d g }| |j| |j| |j| |j || t|j d S)z When the L{NonStreamingProducer} has resumeProducing called 10 times, it writes the counter each time and then fails. F 0123456789N)rarr&r'r(consumerrDrcranger:r0r4rbr,r-)rrwr(_ZexpectedWritesr r r!test_producesOnly10Timesas       z2NonStreamingProducerTests.test_producesOnly10TimescCs4t}t|}||d||t|jdS)zb When the L{NonStreamingProducer} is paused, it raises a L{RuntimeError}. FN)rarr&r:r,r-r8)rrwr(r r r!test_cannotPauseProduction{s  z4NonStreamingProducerTests.test_cannotPauseProductionN)rOrPrQrRrzr{r r r r!rk\srkc@seZdZeddddZdS) _SupportsNamestrrcCsdSrr rr r r!rOsz_SupportsName.__name__N)rOrPrQpropertyr r r r!r|sr|c@seZdZdZddddddZddd d Zddd d Zddd dZddddZddddZ ddddZ ddddZ ddddZ ddddZ ddddZddddZdddd Zddd!d"Zddd#d$Zd%S)&DeprecationTestsz8 Deprecations in L{twisted.test.proto_helpers}. zCallable[..., object]r|r)testobjrcCsdd|j}||g}|t|dd|dt||||dd||t|dS)Nztwisted.internet.testing.rcategorymessage)rOZ flushWarningsr4DeprecationWarninglenZassertInr'r)rrrnew_pathwarningsr r r!helpers   zDeprecationTests.helperrcCsddlm}||j|dS)Nr)AccumulatingProtocol)twisted.test.proto_helpersrrtest_accumulatingProtocol)rrr r r!rs z*DeprecationTests.test_accumulatingProtocolcCsddlm}||j|dS)Nr)LineSendingProtocol)rrrtest_lineSendingProtocol)rrr r r!rs z)DeprecationTests.test_lineSendingProtocolcCsddlm}||j|dS)Nr)FakeDatagramTransport)rrrtest_fakeDatagramTransport)rrr r r!rs z+DeprecationTests.test_fakeDatagramTransportcCsddlm}||j|dS)Nr)r)rrrtest_stringTransport)rrr r r!rs z%DeprecationTests.test_stringTransportcCsddlm}||j|dS)Nr) StringTransportWithDisconnection)rrr%test_stringTransportWithDisconnection)rrr r r!rs  z6DeprecationTests.test_stringTransportWithDisconnectioncCsddlm}||j|dS)Nr)StringIOWithoutClosing)rrrtest_stringIOWithoutClosing)rrr r r!rs z,DeprecationTests.test_stringIOWithoutClosingcCsddlm}||j|dS)Nr)_FakeConnector)rrrtest__fakeConnector)rrr r r!rs z$DeprecationTests.test__fakeConnectorcCsddlm}||j|dS)Nr) _FakePort)rrrtest__fakePort)rrr r r!rs zDeprecationTests.test__fakePortcCsddlm}||j|dS)Nr)r)rrrtest_memoryReactor)rrr r r!rs z#DeprecationTests.test_memoryReactorcCsddlm}||j|dS)Nr)MemoryReactorClock)rrrtest_memoryReactorClock)rrr r r!rs z(DeprecationTests.test_memoryReactorClockcCsddlm}||j|dS)Nr)r)rrrtest_raisingMemoryReactor)rrr r r!rs z*DeprecationTests.test_raisingMemoryReactorcCsddlm}||j|dS)Nr)r)rrrtest_nonStreamingProducer)rrr r r!rs z*DeprecationTests.test_nonStreamingProducercCsddlm}||j|dS)Nr)waitUntilAllDisconnected)rrrtest_waitUntilAllDisconnected)rrr r r!rs z.DeprecationTests.test_waitUntilAllDisconnectedcCsddlm}||j|dS)Nr)EventLoggingObserver)rrrtest_eventLoggingObserver)rrr r r!rs z*DeprecationTests.test_eventLoggingObserverN)rOrPrQrRrrrrrrrrrrrrrrrr r r r!rs rN)'rR __future__rtypingrZzope.interface.verifyrZtyping_extensionsrZtwisted.internet.addressrZtwisted.internet.interfacesrrr r r r r rrZtwisted.internet.protocolrrZtwisted.internet.testingrrrrZtwisted.python.reflectrZtwisted.trial.unittestrrrSrarkr|rr r r r!s"     ,   )j$.