o Òó·iþAã@sdZddlmZddlmZddlmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZmZddlmZmZdd lmZmZmZmZdd lmZdd lm Z Gd d „d e ƒZ!Gdd„de ƒZ"Gdd„dƒZ#Gdd„de ƒZ$Gdd„deƒZ%Gdd„de ƒZ&dS)z( Tests for L{twisted.internet.testing}. é)Ú annotations)ÚCallable)Ú verifyObject)ÚProtocol)Ú IPv4Address) ÚIAddressÚ IConnectorÚ IConsumerÚIListeningPortÚ IPushProducerÚ IReactorSSLÚ IReactorTCPÚ IReactorUNIXÚ ITransport)Ú ClientFactoryÚFactory)Ú MemoryReactorÚNonStreamingProducerÚRaisingMemoryReactorÚStringTransport)ÚnamedAny)ÚTestCasec@sÎeZdZdZd+dd„Zd+dd„Zd+dd „Zd+d d „Zd+d d „Zd+dd„Z d+dd„Z d+dd„Z d+dd„Z d+dd„Z d+dd„Zd+dd„Zd+dd„Zd+dd„Zd+d d!„Zd+d"d#„Zd+d$d%„Zd+d&d'„Zd+d(d)„Zd*S),ÚStringTransportTestsz@ Tests for L{twisted.internet.testing.StringTransport}. ÚreturnÚNonecCs tƒ|_dS©N)rÚ transport©Úself©rúhC:\Users\psant\Downloads\PAF backend\school\venv\Lib\site-packages\twisted/internet/test/test_testing.pyÚsetUp+s zStringTransportTests.setUpcCs:| tt|jƒ¡| tt|jƒ¡| tt|jƒ¡dS)zq L{StringTransport} instances provide L{ITransport}, L{IPushProducer}, and L{IConsumer}. N)Ú assertTruerrrr r rrrr Ú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)ÚobjectrÚregisterProducerÚassertIsÚproducerÚ streaming©rr'r(rrr Útest_registerProducer7s z*StringTransportTests.test_registerProducercCsLtƒ}|j |d¡| t|jjtƒd¡| |jj|¡| |jj¡dS)zy L{StringTransport.registerProducer} raises L{RuntimeError} if a producer is already registered. TFN) r$rr%Ú assertRaisesÚ RuntimeErrorr&r'r"r()rr'rrr Útest_disallowedRegisterProducerBsÿz4StringTransportTests.test_disallowedRegisterProducercCsbtƒ}tƒ}|j |d¡|j ¡| |jj¡|j |d¡| |jj|¡| |jj¡dS)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 newProducerrrr Útest_unregisterProducerOs z,StringTransportTests.test_unregisterProducercCs| t|jj¡dS)z€ L{StringTransport.unregisterProducer} raises L{RuntimeError} if called when no producer is registered. N)r+r,rr.rrrr Útest_invalidUnregisterProducer^sz3StringTransportTests.test_invalidUnregisterProducercCs| |jjd¡dS)zO L{StringTransport.producerState} is initially C{'producing'}. Ú producingN)Ú assertEqualrÚ producerStaterrrr Útest_initialProducerStateesz.StringTransportTests.test_initialProducerStatecCó|j ¡| |jjd¡dS)zy L{StringTransport.pauseProducing} changes the C{producerState} of the transport to C{'paused'}. ÚpausedN)rÚpauseProducingr3r4rrrr Útest_pauseProducingkó z(StringTransportTests.test_pauseProducingcCs(|j ¡|j ¡| |jjd¡dS)z} L{StringTransport.resumeProducing} changes the C{producerState} of the transport to C{'producing'}. r2N)rr8ÚresumeProducingr3r4rrrr Útest_resumeProducingss  z)StringTransportTests.test_resumeProducingcCr6)z{ L{StringTransport.stopProducing} changes the C{'producerState'} of the transport to C{'stopped'}. ÚstoppedN)rÚ stopProducingr3r4rrrr Útest_stopProducing|r:z'StringTransportTests.test_stopProducingcCó|j ¡| t|jj¡dS)zu L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport has been stopped. N)rr>r+r,r8rrrr Ú test_stoppedTransportCannotPause„r:z5StringTransportTests.test_stoppedTransportCannotPausecCr@)zv L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport has been stopped. N)rr>r+r,r;rrrr Ú!test_stoppedTransportCannotResumeŒr:z6StringTransportTests.test_stoppedTransportCannotResumecCr@)zz L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport is being disconnected. N)rÚloseConnectionr+r,r8rrrr Ú&test_disconnectingTransportCannotPause”r:z;StringTransportTests.test_disconnectingTransportCannotPausecCr@)z{ L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport is being disconnected. N)rrCr+r,r;rrrr Ú'test_disconnectingTransportCannotResumeœr:z Adding, removing, and listing readers works. N)r$rZ addReaderr3Z getReadersZ removeReader)rÚreaderÚreactorrrr Ú test_readersó   zReactorTests.test_readerscCrc)z> Adding, removing, and listing writers works. N)r$rZ addWriterr3Z getWritersZ removeWriter)rÚwriterrerrr Ú test_writers'rgzReactorTests.test_writersNrS) rTrUrVrWr[r]rarbrfrirrrr rXÎs   rXc@s8eZdZdZddd„Zdd d „Zdd d „Zddd„ZdS)Ú TestConsumerzP A very basic test consumer for use with the NonStreamingProducerTests. rrcCsg|_d|_d|_dSr)Úwritesr'ÚproducerStreamingrrrr Ú__init__=s zTestConsumer.__init__r'r$r(ÚboolcCs||_||_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. N©r'rlr)rrr r%Bs zTestConsumer.registerProducercCsd|_d|_dS)zC Forget the producer we had previously registered. Nrorrrr r.Ls zTestConsumer.unregisterProducerÚdataÚbytescCs|j |¡dS)zz Some data was written to the consumer: stores it for later use. @param data: The data to write. N)rkÚappend)rrprrr ÚwriteSszTestConsumer.writeNrS)r'r$r(rnrr)rprqrr)rTrUrVrWrmr%r.rsrrrr rj8s    rjc@s$eZdZdZd dd„Zd dd„ZdS) ÚNonStreamingProducerTestszF Tests for the L{NonStreamingProducer} to validate behaviour. rrcCs¤tƒ}t|ƒ}| |d¡| |j|¡| |j|¡| |j¡tdƒD]}|  ¡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é ) ó0ó1ó2ó3ó4ó5ó6ó7ó8ó9N)rjrr%r&r'ÚconsumerrFrlÚranger;r/r3rkr+r,)rr€r'Ú_ZexpectedWritesrrr Útest_producesOnly10Timesas       z2NonStreamingProducerTests.test_producesOnly10TimescCs4tƒ}t|ƒ}| |d¡| ¡| t|j¡dS)zb When the L{NonStreamingProducer} is paused, it raises a L{RuntimeError}. FN)rjrr%r;r+r,r8)rr€r'rrr Útest_cannotPauseProduction{s  z4NonStreamingProducerTests.test_cannotPauseProductionNrS)rTrUrVrWrƒr„rrrr rt\s rtc@seZdZeddd„ƒZdS)Ú _SupportsNamerÚstrcCsdSrrrrrr rT‹sz_SupportsName.__name__N)rr†)rTrUrVÚpropertyrrrr r…Šsr…c@s¦eZdZdZd'dd „Zd(d d „Zd(d d „Zd(dd„Zd(dd„Zd(dd„Z d(dd„Z d(dd„Z d(dd„Z d(dd„Z d(dd„Zd(dd„Zd(d d!„Zd(d"d#„Zd(d$d%„Zd&S))ÚDeprecationTestsz8 Deprecations in L{twisted.test.proto_helpers}. ÚtestúCallable[..., object]Úobjr…rrcCsdd|j›}| |g¡}| t|dd¡| dt|ƒ¡| ||dd¡| |t|ƒ¡dS)Nztwisted.internet.testing.rÚcategoryéÚmessage)rTZ flushWarningsr3ÚDeprecationWarningÚlenZassertInr&r)rr‰r‹Únew_pathÚwarningsrrr Úhelper•s  zDeprecationTests.helpercCóddlm}| |j|¡dS)Nr)ÚAccumulatingProtocol)Útwisted.test.proto_helpersr•r“Útest_accumulatingProtocol)rr•rrr r—ó z*DeprecationTests.test_accumulatingProtocolcCr”)Nr)ÚLineSendingProtocol)r–r™r“Útest_lineSendingProtocol)rr™rrr rš¢r˜z)DeprecationTests.test_lineSendingProtocolcCr”)Nr)ÚFakeDatagramTransport)r–r›r“Útest_fakeDatagramTransport)rr›rrr rœ§r˜z+DeprecationTests.test_fakeDatagramTransportcCr”)Nr)r)r–rr“Útest_stringTransport)rrrrr r¬r˜z%DeprecationTests.test_stringTransportcCr”)Nr)Ú StringTransportWithDisconnection)r–ržr“Ú%test_stringTransportWithDisconnection)rržrrr rŸ±s ÿz6DeprecationTests.test_stringTransportWithDisconnectioncCr”)Nr)ÚStringIOWithoutClosing)r–r r“Útest_stringIOWithoutClosing)rr rrr r¡¸r˜z,DeprecationTests.test_stringIOWithoutClosingcCr”)Nr)Ú_FakeConnector)r–r¢r“Útest__fakeConnector)rr¢rrr r£½r˜z$DeprecationTests.test__fakeConnectorcCr”)Nr)Ú _FakePort)r–r¤r“Útest__fakePort)rr¤rrr r¥Âr˜zDeprecationTests.test__fakePortcCr”)Nr)r)r–rr“Útest_memoryReactor)rrrrr r¦Çr˜z#DeprecationTests.test_memoryReactorcCr”)Nr)ÚMemoryReactorClock)r–r§r“Útest_memoryReactorClock)rr§rrr r¨Ìr˜z(DeprecationTests.test_memoryReactorClockcCr”)Nr)r)r–rr“Útest_raisingMemoryReactor)rrrrr r©Ñr˜z*DeprecationTests.test_raisingMemoryReactorcCr”)Nr)r)r–rr“Útest_nonStreamingProducer)rrrrr rªÖr˜z*DeprecationTests.test_nonStreamingProducercCr”)Nr)ÚwaitUntilAllDisconnected)r–r«r“Útest_waitUntilAllDisconnected)rr«rrr r¬Ûr˜z.DeprecationTests.test_waitUntilAllDisconnectedcCr”)Nr)ÚEventLoggingObserver)r–r­r“Útest_eventLoggingObserver)rr­rrr r®àr˜z*DeprecationTests.test_eventLoggingObserverN)r‰rŠr‹r…rrrS)rTrUrVrWr“r—ršrœrrŸr¡r£r¥r¦r¨r©rªr¬r®rrrr rˆs"              rˆN)'rWÚ __future__rÚtypingrZzope.interface.verifyrZtyping_extensionsrZtwisted.internet.addressrZtwisted.internet.interfacesrrr r r r r rrZtwisted.internet.protocolrrZtwisted.internet.testingrrrrZtwisted.python.reflectrZtwisted.trial.unittestrrrXrjrtr…rˆrrrr Ús$     ,   )j$.