<s> __version__ = '<STR_LIT>' [ <NUM_LIT:11> : - <NUM_LIT:2> ] <EOL> from twisted . trial import unittest <EOL> from twisted . protocols import pcp <EOL> class DummyTransport : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _writes = [ ] <EOL> def write ( self , data ) : <EOL> self . _writes . append ( data ) <EOL> def getvalue ( self ) : <EOL> return '<STR_LIT>' . join ( self . _writes ) <EOL> class DummyProducer : <EOL> resumed = False <EOL> stopped = False <EOL> paused = False <EOL> def __init__ ( self , consumer ) : <EOL> self . consumer = consumer <EOL> def resumeProducing ( self ) : <EOL> self . resumed = True <EOL> self . paused = False <EOL> def pauseProducing ( self ) : <EOL> self . paused = True <EOL> def stopProducing ( self ) : <EOL> self . stopped = True <EOL> class DummyConsumer ( DummyTransport ) : <EOL> producer = None <EOL> finished = False <EOL> unregistered = True <EOL> def registerProducer ( self , producer , streaming ) : <EOL> self . producer = ( producer , streaming ) <EOL> def unregisterProducer ( self ) : <EOL> self . unregistered = True <EOL> def finish ( self ) : <EOL> self . finished = True <EOL> class TransportInterfaceTests ( unittest . TestCase ) : <EOL> proxyClass = pcp . BasicProducerConsumerProxy <EOL> def setUp ( self ) : <EOL> self . underlying = DummyConsumer ( ) <EOL> self . transport = self . proxyClass ( self . underlying ) <EOL> def testWrite ( self ) : <EOL> self . transport . write ( "<STR_LIT>" ) <EOL> class ConsumerInterfaceTest : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . underlying = DummyConsumer ( ) <EOL> self . consumer = self . proxyClass ( self . underlying ) <EOL> self . producer = DummyProducer ( self . consumer ) <EOL> def testRegisterPush ( self ) : <EOL> self . consumer . registerProducer ( self . producer , True ) <EOL> self . assertFalse ( self . producer . resumed ) <EOL> def testUnregister ( self ) : <EOL> self . consumer . registerProducer ( self . producer , False ) <EOL> self . consumer . unregisterProducer ( ) <EOL> self . producer . resumed = False <EOL> self . consumer . resumeProducing ( ) <EOL> self . assertFalse ( self . producer . resumed ) <EOL> def testFinish ( self ) : <EOL> self . consumer . registerProducer ( self . producer , False ) <EOL> self . consumer . finish ( ) <EOL> self . producer . resumed = False <EOL> self . consumer . resumeProducing ( ) <EOL> self . assertFalse ( self . producer . resumed ) <EOL> class ProducerInterfaceTest : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . consumer = DummyConsumer ( ) <EOL> self . producer = self . proxyClass ( self . consumer ) <EOL> def testRegistersProducer ( self ) : <EOL> self . assertEqual ( self . consumer . producer [ <NUM_LIT:0> ] , self . producer ) <EOL> def testPause ( self ) : <EOL> self . producer . pauseProducing ( ) <EOL> self . producer . write ( "<STR_LIT>" ) <EOL> self . assertFalse ( self . consumer . getvalue ( ) , <EOL> "<STR_LIT>" ) <EOL> def testResume ( self ) : <EOL> self . producer . pauseProducing ( ) <EOL> self . producer . resumeProducing ( ) <EOL> self . producer . write ( "<STR_LIT>" ) <EOL> self . assertEqual ( self . consumer . getvalue ( ) , "<STR_LIT>" ) <EOL> def testResumeNoEmptyWrite ( self ) : <EOL> self . producer . pauseProducing ( ) <EOL> self . producer . resumeProducing ( ) <EOL> self . assertEqual ( len ( self . consumer . _writes ) , <NUM_LIT:0> , <EOL> "<STR_LIT>" ) <EOL> def testResumeBuffer ( self ) : <EOL> self . producer . pauseProducing ( ) <EOL> self . producer . write ( "<STR_LIT>" ) <EOL> self . producer . resumeProducing ( ) <EOL> self . assertEqual ( self . consumer . getvalue ( ) , "<STR_LIT>" ) <EOL> def testStop ( self ) : <EOL> self . producer . stopProducing ( ) <EOL> self . producer . write ( "<STR_LIT>" ) <EOL> self . assertFalse ( self . consumer . getvalue ( ) , <EOL> "<STR_LIT>" ) <EOL> class PCP_ConsumerInterfaceTests ( ConsumerInterfaceTest , unittest . TestCase ) : <EOL> proxyClass = pcp . BasicProducerConsumerProxy <EOL> class PCPII_ConsumerInterfaceTests ( ConsumerInterfaceTest , unittest . TestCase ) : <EOL> proxyClass = pcp . ProducerConsumerProxy <EOL> class PCP_ProducerInterfaceTests ( ProducerInterfaceTest , unittest . TestCase ) : <EOL> proxyClass = pcp . BasicProducerConsumerProxy <EOL> class PCPII_ProducerInterfaceTests ( ProducerInterfaceTest , unittest . TestCase ) : <EOL> proxyClass = pcp . ProducerConsumerProxy <EOL> class ProducerProxyTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> proxyClass = pcp . BasicProducerConsumerProxy <EOL> def setUp ( self ) : <EOL> self . proxy = self . proxyClass ( None ) <EOL> self . parentProducer = DummyProducer ( self . proxy ) <EOL> self . proxy . registerProducer ( self . parentProducer , True ) <EOL> def testStop ( self ) : <EOL> self . proxy . stopProducing ( ) <EOL> self . assertTrue ( self . parentProducer . stopped ) <EOL> class ConsumerProxyTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> proxyClass = pcp . BasicProducerConsumerProxy <EOL> def setUp ( self ) : <EOL> self . underlying = DummyConsumer ( ) <EOL> self . consumer = self . proxyClass ( self . underlying ) <EOL> def testWrite ( self ) : <EOL> self . consumer . write ( "<STR_LIT>" ) <EOL> self . assertEqual ( self . underlying . getvalue ( ) , "<STR_LIT>" ) <EOL> def testFinish ( self ) : <EOL> self . consumer . finish ( ) <EOL> self . assertTrue ( self . underlying . finished ) <EOL> def testUnregister ( self ) : <EOL> self . consumer . unregisterProducer ( ) <EOL> self . assertTrue ( self . underlying . unregistered ) <EOL> class PullProducerTest : <EOL> def setUp ( self ) : <EOL> self . underlying = DummyConsumer ( ) <EOL> self . proxy = self . proxyClass ( self . underlying ) <EOL> self . parentProducer = DummyProducer ( self . proxy ) <EOL> self . proxy . registerProducer ( self . parentProducer , True ) <EOL> def testHoldWrites ( self ) : <EOL> self . proxy . write ( "<STR_LIT:hello>" ) <EOL> self . assertFalse ( self . underlying . getvalue ( ) , <EOL> "<STR_LIT>" ) <EOL> def testPull ( self ) : <EOL> self . proxy . write ( "<STR_LIT:hello>" ) <EOL> self . proxy . resumeProducing ( ) <EOL> self . assertEqual ( self . underlying . getvalue ( ) , "<STR_LIT:hello>" ) <EOL> def testMergeWrites ( self ) : <EOL> self . proxy . write ( "<STR_LIT>" ) <EOL> self . proxy . write ( "<STR_LIT>" ) <EOL> self . proxy . resumeProducing ( ) <EOL> nwrites = len ( self . underlying . _writes ) <EOL> self . assertEqual ( nwrites , <NUM_LIT:1> , "<STR_LIT>" <EOL> "<STR_LIT>" % ( nwrites , ) ) <EOL> self . assertEqual ( self . underlying . getvalue ( ) , "<STR_LIT>" ) <EOL> def testLateWrite ( self ) : <EOL> self . proxy . resumeProducing ( ) <EOL> self . proxy . write ( "<STR_LIT:data>" ) <EOL> self . assertEqual ( self . underlying . getvalue ( ) , "<STR_LIT:data>" ) <EOL> class PCP_PullProducerTests ( PullProducerTest , unittest . TestCase ) : <EOL> class proxyClass ( pcp . BasicProducerConsumerProxy ) : <EOL> iAmStreaming = False <EOL> class PCPII_PullProducerTests ( PullProducerTest , unittest . TestCase ) : <EOL> class proxyClass ( pcp . ProducerConsumerProxy ) : <EOL> iAmStreaming = False <EOL> class BufferedConsumerTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> proxyClass = pcp . ProducerConsumerProxy <EOL> def setUp ( self ) : <EOL> self . underlying = DummyConsumer ( ) <EOL> self . proxy = self . proxyClass ( self . underlying ) <EOL> self . proxy . bufferSize = <NUM_LIT:100> <EOL> self . parentProducer = DummyProducer ( self . proxy ) <EOL> self . proxy . registerProducer ( self . parentProducer , True ) <EOL> def testRegisterPull ( self ) : <EOL> self . proxy . registerProducer ( self . parentProducer , False ) <EOL> self . assertTrue ( self . parentProducer . resumed ) <EOL> def testPauseIntercept ( self ) : <EOL> self . proxy . pauseProducing ( ) <EOL> self . assertFalse ( self . parentProducer . paused ) <EOL> def testResumeIntercept ( self ) : <EOL> self . proxy . pauseProducing ( ) <EOL> self . proxy . resumeProducing ( ) <EOL> self . assertFalse ( self . parentProducer . resumed ) <EOL> def testTriggerPause ( self ) : <EOL> """<STR_LIT>""" <EOL> self . proxy . pauseProducing ( ) <EOL> self . assertFalse ( self . parentProducer . paused , "<STR_LIT>" ) <EOL> self . proxy . write ( "<STR_LIT:x>" * <NUM_LIT> ) <EOL> self . assertFalse ( self . parentProducer . paused , "<STR_LIT>" ) <EOL> self . proxy . write ( "<STR_LIT:x>" * <NUM_LIT> ) <EOL> self . assertTrue ( self . parentProducer . paused ) <EOL> def testTriggerResume ( self ) : <EOL> """<STR_LIT>""" <EOL> self . proxy . pauseProducing ( ) <EOL> self . proxy . write ( "<STR_LIT:x>" * <NUM_LIT> ) <EOL> self . assertTrue ( self . parentProducer . paused , "<STR_LIT>" ) <EOL> self . proxy . resumeProducing ( ) <EOL> self . assertFalse ( self . parentProducer . paused , <EOL> "<STR_LIT>" ) <EOL> self . assertFalse ( self . proxy . producerPaused ) <EOL> class BufferedPullTests ( unittest . TestCase ) : <EOL> class proxyClass ( pcp . ProducerConsumerProxy ) : <EOL> iAmStreaming = False <EOL> def _writeSomeData ( self , data ) : <EOL> pcp . ProducerConsumerProxy . _writeSomeData ( self , data [ : <NUM_LIT:100> ] ) <EOL> return min ( len ( data ) , <NUM_LIT:100> ) <EOL> def setUp ( self ) : <EOL> self . underlying = DummyConsumer ( ) <EOL> self . proxy = self . proxyClass ( self . underlying ) <EOL> self . proxy . bufferSize = <NUM_LIT:100> <EOL> self . parentProducer = DummyProducer ( self . proxy ) <EOL> self . proxy . registerProducer ( self . parentProducer , False ) <EOL> def testResumePull ( self ) : <EOL> self . parentProducer . resumed = False <EOL> self . proxy . resumeProducing ( ) <EOL> self . assertTrue ( self . parentProducer . resumed ) <EOL> def testLateWriteBuffering ( self ) : <EOL> self . proxy . resumeProducing ( ) <EOL> self . proxy . write ( "<STR_LIT>" * <NUM_LIT> ) <EOL> self . assertEqual ( self . underlying . getvalue ( ) , "<STR_LIT>" * <NUM_LIT:20> ) <EOL> self . assertEqual ( self . proxy . _buffer , [ "<STR_LIT>" ] ) </s>
