o >hA@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@seZdZdZd+ddZd+ddZd+dd Zd+d d Zd+d d Zd+ddZ d+ddZ d+ddZ d+ddZ d+ddZ d+ddZd+ddZd+ddZd+ddZd+d d!Zd+d"d#Zd+d$d%Zd+d&d'Zd+d(d)Zd*S),StringTransportTestsz@ Tests for L{twisted.internet.testing.StringTransport}. returnNonecCs t|_dSN)r transportselfr/var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/site-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) 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)objectrregisterProducerassertIsproducer streamingrr'r(rrr 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'rrr 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()r oldProducer newProducerrrr 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.rrrr test_invalidUnregisterProducer^sz3StringTransportTests.test_invalidUnregisterProducercCs||jjddS)zO L{StringTransport.producerState} is initially C{'producing'}. producingN) assertEqualr producerStaterrrr test_initialProducerStateesz.StringTransportTests.test_initialProducerStatecC|j||jjddS)zy L{StringTransport.pauseProducing} changes the C{producerState} of the transport to C{'paused'}. pausedN)rpauseProducingr5r6rrrr test_pauseProducingk z(StringTransportTests.test_pauseProducingcCs(|j|j||jjddS)z} L{StringTransport.resumeProducing} changes the C{producerState} of the transport to C{'producing'}. r4N)rr:resumeProducingr5r6rrrr test_resumeProducingss  z)StringTransportTests.test_resumeProducingcCr8)z{ L{StringTransport.stopProducing} changes the C{'producerState'} of the transport to C{'stopped'}. stoppedN)r stopProducingr5r6rrrr test_stopProducing|r<z'StringTransportTests.test_stopProducingcC|j|t|jjdS)zu L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport has been stopped. N)rr@r+r,r:rrrr test_stoppedTransportCannotPauser<z5StringTransportTests.test_stoppedTransportCannotPausecCrB)zv L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport has been stopped. N)rr@r+r,r=rrrr !test_stoppedTransportCannotResumer<z6StringTransportTests.test_stoppedTransportCannotResumecCrB)zz L{StringTransport.pauseProducing} raises L{RuntimeError} if the transport is being disconnected. N)rloseConnectionr+r,r:rrrr &test_disconnectingTransportCannotPauser<z;StringTransportTests.test_disconnectingTransportCannotPausecCrB)z{ L{StringTransport.resumeProducing} raises L{RuntimeError} if the transport is being disconnected. N)rrEr+r,r=rrrr 'test_disconnectingTransportCannotResumer<zrArCrDrFrGrJrNrQrTrVrrrr r&s*             rc@sLeZdZdZdddZdddZddd Zdd d Zdd d ZdddZ dS) ReactorTestszA Tests for L{MemoryReactor} and L{RaisingMemoryReactor}. rrcC(t}tt|tt|tt|dS)zt L{MemoryReactor} provides all of the attributes described by the interfaces it advertises. N)rrr r r)r memoryReactorrrr test_memoryReactorProvides  z'ReactorTests.test_memoryReactorProvidescCr])z{ L{RaisingMemoryReactor} provides all of the attributes described by the interfaces it advertises. N)rrr r r)rraisingReactorrrr test_raisingReactorProvidesr`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) r connectTCPr connectSSLrrgetDestinationrr5hostport connectUNIXname)rr^ connectorrMrrr 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) r listenTCPr listenSSLrr rKrr5rfrg listenUNIXri)rr^rgrMrrr test_listenDefaultHosts     z#ReactorTests.test_listenDefaultHostcCPt}t}|||||||g||||gdS)z> Adding, removing, and listing readers works. N)r$r addReaderr5 getReaders removeReader)rreaderreactorrrr test_readers   zReactorTests.test_readerscCrp)z> Adding, removing, and listing writers works. N)r$r addWriterr5 getWriters removeWriter)rwriterrurrr test_writers'rwzReactorTests.test_writersNrW) rXrYrZr[r_rbrkrorvr|rrrr r\s   r\c@s8eZdZdZdddZdd d Zdd d ZdddZdS) 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. Nr'rr)rrr r%Bs zTestConsumer.registerProducercCsd|_d|_dS)zC Forget the producer we had previously registered. Nrrrrr r.Ls zTestConsumer.unregisterProducerdatabytescCs|j|dS)zz Some data was written to the consumer: stores it for later use. @param data: The data to write. N)r~append)rrrrr writeSszTestConsumer.writeNrW)r'r$r(rrr)rrrr)rXrYrZr[rr%r.rrrrr r}8s    r}c@s$eZdZdZd ddZd ddZdS) NonStreamingProducerTestszF Tests for the L{NonStreamingProducer} to validate behaviour. rrcCst}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)r}rr%r&r'consumerrHrranger=r/r5r~r+r,)rrr'_expectedWritesrrr test_producesOnly10Timesas       z2NonStreamingProducerTests.test_producesOnly10TimescCs4t}t|}||d||t|jdS)zb When the L{NonStreamingProducer} is paused, it raises a L{RuntimeError}. FN)r}rr%r=r+r,r:)rrr'rrr test_cannotPauseProduction{s  z4NonStreamingProducerTests.test_cannotPauseProductionNrW)rXrYrZr[rrrrrr r\s rc@seZdZedddZdS) _SupportsNamerstrcCsdSrrrrrr rXsz_SupportsName.__name__N)rr)rXrYrZpropertyrrrr rsrc@seZdZdZd'dd Zd(d d Zd(d d Zd(ddZd(ddZd(ddZ d(ddZ d(ddZ d(ddZ d(ddZ d(ddZd(ddZd(d d!Zd(d"d#Zd(d$d%Zd&S))DeprecationTestsz8 Deprecations in L{twisted.test.proto_helpers}. testCallable[..., object]objrrrcCsdd|j}||g}|t|dd|dt||||dd||t|dS)Nztwisted.internet.testing.rcategorymessage)rX flushWarningsr5DeprecationWarninglenassertInr&r)rrrnew_pathwarningsrrr helpers  zDeprecationTests.helpercCddlm}||j|dS)Nr)AccumulatingProtocol)twisted.test.proto_helpersrrtest_accumulatingProtocol)rrrrr r z*DeprecationTests.test_accumulatingProtocolcCr)Nr)LineSendingProtocol)rrrtest_lineSendingProtocol)rrrrr rrz)DeprecationTests.test_lineSendingProtocolcCr)Nr)FakeDatagramTransport)rrrtest_fakeDatagramTransport)rrrrr rrz+DeprecationTests.test_fakeDatagramTransportcCr)Nr)r)rrrtest_stringTransport)rrrrr rrz%DeprecationTests.test_stringTransportcCr)Nr) StringTransportWithDisconnection)rrr%test_stringTransportWithDisconnection)rrrrr rs z6DeprecationTests.test_stringTransportWithDisconnectioncCr)Nr)StringIOWithoutClosing)rrrtest_stringIOWithoutClosing)rrrrr rrz,DeprecationTests.test_stringIOWithoutClosingcCr)Nr)_FakeConnector)rrrtest__fakeConnector)rrrrr rrz$DeprecationTests.test__fakeConnectorcCr)Nr) _FakePort)rrrtest__fakePort)rrrrr rrzDeprecationTests.test__fakePortcCr)Nr)r)rrrtest_memoryReactor)rrrrr rrz#DeprecationTests.test_memoryReactorcCr)Nr)MemoryReactorClock)rrrtest_memoryReactorClock)rrrrr rrz(DeprecationTests.test_memoryReactorClockcCr)Nr)r)rrrtest_raisingMemoryReactor)rrrrr rrz*DeprecationTests.test_raisingMemoryReactorcCr)Nr)r)rrrtest_nonStreamingProducer)rrrrr rrz*DeprecationTests.test_nonStreamingProducercCr)Nr)waitUntilAllDisconnected)rrrtest_waitUntilAllDisconnected)rrrrr rrz.DeprecationTests.test_waitUntilAllDisconnectedcCr)Nr)EventLoggingObserver)rrrtest_eventLoggingObserver)rrrrr rrz*DeprecationTests.test_eventLoggingObserverN)rrrrrrrW)rXrYrZr[rrrrrrrrrrrrrrrrrrr rs"              rN)'r[ __future__rtypingrzope.interface.verifyrtyping_extensionsrtwisted.internet.addressrtwisted.internet.interfacesrrr r r r r rrtwisted.internet.protocolrrtwisted.internet.testingrrrrtwisted.python.reflectrtwisted.trial.unittestrrr\r}rrrrrrr s$     ,   )j$.