<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import <EOL> from zope . interface import implementer <EOL> from twisted . python . compat import _PY3 , intToBytes <EOL> from twisted . trial import unittest <EOL> from twisted . trial . util import suppress as SUPPRESS <EOL> from twisted . protocols import basic , loopback <EOL> from twisted . internet import defer <EOL> from twisted . internet . protocol import Protocol <EOL> from twisted . internet . defer import Deferred <EOL> from twisted . internet . interfaces import IAddress , IPushProducer , IPullProducer <EOL> from twisted . internet import reactor , interfaces <EOL> class SimpleProtocol ( basic . LineReceiver ) : <EOL> def __init__ ( self ) : <EOL> self . conn = defer . Deferred ( ) <EOL> self . lines = [ ] <EOL> self . connLost = [ ] <EOL> def connectionMade ( self ) : <EOL> self . conn . callback ( None ) <EOL> def lineReceived ( self , line ) : <EOL> self . lines . append ( line ) <EOL> def connectionLost ( self , reason ) : <EOL> self . connLost . append ( reason ) <EOL> class DoomProtocol ( SimpleProtocol ) : <EOL> i = <NUM_LIT:0> <EOL> def lineReceived ( self , line ) : <EOL> self . i += <NUM_LIT:1> <EOL> if self . i < <NUM_LIT:4> : <EOL> self . sendLine ( b"<STR_LIT>" + intToBytes ( self . i ) ) <EOL> SimpleProtocol . lineReceived ( self , line ) <EOL> if self . lines [ - <NUM_LIT:1> ] == b"<STR_LIT>" : <EOL> self . transport . loseConnection ( ) <EOL> class LoopbackTestCaseMixin : <EOL> def testRegularFunction ( self ) : <EOL> s = SimpleProtocol ( ) <EOL> c = SimpleProtocol ( ) <EOL> def sendALine ( result ) : <EOL> s . sendLine ( b"<STR_LIT>" ) <EOL> s . transport . loseConnection ( ) <EOL> s . conn . addCallback ( sendALine ) <EOL> def check ( ignored ) : <EOL> self . assertEqual ( c . lines , [ b"<STR_LIT>" ] ) <EOL> self . assertEqual ( len ( s . connLost ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( c . connLost ) , <NUM_LIT:1> ) <EOL> d = defer . maybeDeferred ( self . loopbackFunc , s , c ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def testSneakyHiddenDoom ( self ) : <EOL> s = DoomProtocol ( ) <EOL> c = DoomProtocol ( ) <EOL> def sendALine ( result ) : <EOL> s . sendLine ( b"<STR_LIT>" ) <EOL> s . conn . addCallback ( sendALine ) <EOL> def check ( ignored ) : <EOL> self . assertEqual ( s . lines , [ b'<STR_LIT>' , b'<STR_LIT>' , b'<STR_LIT>' ] ) <EOL> self . assertEqual ( <EOL> c . lines , [ b'<STR_LIT>' , b'<STR_LIT>' , b'<STR_LIT>' , b'<STR_LIT>' ] ) <EOL> self . assertEqual ( len ( s . connLost ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( c . connLost ) , <NUM_LIT:1> ) <EOL> d = defer . maybeDeferred ( self . loopbackFunc , s , c ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> class LoopbackAsyncTestCase ( LoopbackTestCaseMixin , unittest . TestCase ) : <EOL> loopbackFunc = staticmethod ( loopback . loopbackAsync ) <EOL> def test_makeConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestProtocol ( Protocol ) : <EOL> transport = None <EOL> def makeConnection ( self , transport ) : <EOL> self . transport = transport <EOL> server = TestProtocol ( ) <EOL> client = TestProtocol ( ) <EOL> loopback . loopbackAsync ( server , client ) <EOL> self . failIfEqual ( client . transport , None ) <EOL> self . failIfEqual ( server . transport , None ) <EOL> def _hostpeertest ( self , get , testServer ) : <EOL> """<STR_LIT>""" <EOL> class TestProtocol ( Protocol ) : <EOL> def makeConnection ( self , transport ) : <EOL> Protocol . makeConnection ( self , transport ) <EOL> self . onConnection . callback ( transport ) <EOL> if testServer : <EOL> server = TestProtocol ( ) <EOL> d = server . onConnection = Deferred ( ) <EOL> client = Protocol ( ) <EOL> else : <EOL> server = Protocol ( ) <EOL> client = TestProtocol ( ) <EOL> d = client . onConnection = Deferred ( ) <EOL> loopback . loopbackAsync ( server , client ) <EOL> def connected ( transport ) : <EOL> host = getattr ( transport , get ) ( ) <EOL> self . failUnless ( IAddress . providedBy ( host ) ) <EOL> return d . addCallback ( connected ) <EOL> def test_serverHost ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hostpeertest ( "<STR_LIT>" , True ) <EOL> def test_serverPeer ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hostpeertest ( "<STR_LIT>" , True ) <EOL> def test_clientHost ( self , get = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . _hostpeertest ( "<STR_LIT>" , False ) <EOL> def test_clientPeer ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _hostpeertest ( "<STR_LIT>" , False ) <EOL> def _greetingtest ( self , write , testServer ) : <EOL> """<STR_LIT>""" <EOL> class GreeteeProtocol ( Protocol ) : <EOL> bytes = b"<STR_LIT>" <EOL> def dataReceived ( self , bytes ) : <EOL> self . bytes += bytes <EOL> if self . bytes == b"<STR_LIT>" : <EOL> self . received . callback ( None ) <EOL> class GreeterProtocol ( Protocol ) : <EOL> def connectionMade ( self ) : <EOL> if write == "<STR_LIT>" : <EOL> self . transport . write ( b"<STR_LIT>" ) <EOL> else : <EOL> self . transport . writeSequence ( [ b"<STR_LIT>" , b"<STR_LIT>" ] ) <EOL> if testServer : <EOL> server = GreeterProtocol ( ) <EOL> client = GreeteeProtocol ( ) <EOL> d = client . received = Deferred ( ) <EOL> else : <EOL> server = GreeteeProtocol ( ) <EOL> d = server . received = Deferred ( ) <EOL> client = GreeterProtocol ( ) <EOL> loopback . loopbackAsync ( server , client ) <EOL> return d <EOL> def test_clientGreeting ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _greetingtest ( "<STR_LIT>" , False ) <EOL> def test_clientGreetingSequence ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _greetingtest ( "<STR_LIT>" , False ) <EOL> def test_serverGreeting ( self , write = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . _greetingtest ( "<STR_LIT>" , True ) <EOL> def test_serverGreetingSequence ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _greetingtest ( "<STR_LIT>" , True ) <EOL> def _producertest ( self , producerClass ) : <EOL> toProduce = list ( map ( intToBytes , range ( <NUM_LIT:0> , <NUM_LIT:10> ) ) ) <EOL> class ProducingProtocol ( Protocol ) : <EOL> def connectionMade ( self ) : <EOL> self . producer = producerClass ( list ( toProduce ) ) <EOL> self . producer . start ( self . transport ) <EOL> class ReceivingProtocol ( Protocol ) : <EOL> bytes = b"<STR_LIT>" <EOL> def dataReceived ( self , data ) : <EOL> self . bytes += data <EOL> if self . bytes == b'<STR_LIT>' . join ( toProduce ) : <EOL> self . received . callback ( ( client , server ) ) <EOL> server = ProducingProtocol ( ) <EOL> client = ReceivingProtocol ( ) <EOL> client . received = Deferred ( ) <EOL> loopback . loopbackAsync ( server , client ) <EOL> return client . received <EOL> def test_pushProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> @ implementer ( IPushProducer ) <EOL> class PushProducer ( object ) : <EOL> resumed = False <EOL> def __init__ ( self , toProduce ) : <EOL> self . toProduce = toProduce <EOL> def resumeProducing ( self ) : <EOL> self . resumed = True <EOL> def start ( self , consumer ) : <EOL> self . consumer = consumer <EOL> consumer . registerProducer ( self , True ) <EOL> self . _produceAndSchedule ( ) <EOL> def _produceAndSchedule ( self ) : <EOL> if self . toProduce : <EOL> self . consumer . write ( self . toProduce . pop ( <NUM_LIT:0> ) ) <EOL> reactor . callLater ( <NUM_LIT:0> , self . _produceAndSchedule ) <EOL> else : <EOL> self . consumer . unregisterProducer ( ) <EOL> d = self . _producertest ( PushProducer ) <EOL> def finished ( results ) : <EOL> ( client , server ) = results <EOL> self . assertFalse ( <EOL> server . producer . resumed , <EOL> "<STR_LIT>" ) <EOL> d . addCallback ( finished ) <EOL> return d <EOL> def test_pullProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> @ implementer ( IPullProducer ) <EOL> class PullProducer ( object ) : <EOL> def __init__ ( self , toProduce ) : <EOL> self . toProduce = toProduce <EOL> def start ( self , consumer ) : <EOL> self . consumer = consumer <EOL> self . consumer . registerProducer ( self , False ) <EOL> def resumeProducing ( self ) : <EOL> self . consumer . write ( self . toProduce . pop ( <NUM_LIT:0> ) ) <EOL> if not self . toProduce : <EOL> self . consumer . unregisterProducer ( ) <EOL> return self . _producertest ( PullProducer ) <EOL> def test_writeNotReentrant ( self ) : <EOL> """<STR_LIT>""" <EOL> class Server ( Protocol ) : <EOL> def dataReceived ( self , bytes ) : <EOL> self . transport . write ( b"<STR_LIT>" ) <EOL> class Client ( Protocol ) : <EOL> ready = False <EOL> def connectionMade ( self ) : <EOL> reactor . callLater ( <NUM_LIT:0> , self . go ) <EOL> def go ( self ) : <EOL> self . transport . write ( b"<STR_LIT:foo>" ) <EOL> self . ready = True <EOL> def dataReceived ( self , bytes ) : <EOL> self . wasReady = self . ready <EOL> self . transport . loseConnection ( ) <EOL> server = Server ( ) <EOL> client = Client ( ) <EOL> d = loopback . loopbackAsync ( client , server ) <EOL> def cbFinished ( ignored ) : <EOL> self . assertTrue ( client . wasReady ) <EOL> d . addCallback ( cbFinished ) <EOL> return d <EOL> def test_pumpPolicy ( self ) : <EOL> """<STR_LIT>""" <EOL> pumpCalls = [ ] <EOL> def dummyPolicy ( queue , target ) : <EOL> bytes = [ ] <EOL> while queue : <EOL> bytes . append ( queue . get ( ) ) <EOL> pumpCalls . append ( ( target , bytes ) ) <EOL> client = Protocol ( ) <EOL> server = Protocol ( ) <EOL> finished = loopback . loopbackAsync ( server , client , dummyPolicy ) <EOL> self . assertEqual ( pumpCalls , [ ] ) <EOL> client . transport . write ( b"<STR_LIT:foo>" ) <EOL> client . transport . write ( b"<STR_LIT:bar>" ) <EOL> server . transport . write ( b"<STR_LIT>" ) <EOL> server . transport . write ( b"<STR_LIT>" ) <EOL> server . transport . loseConnection ( ) <EOL> def cbComplete ( ignored ) : <EOL> self . assertEqual ( <EOL> pumpCalls , <EOL> [ ( client , [ b"<STR_LIT>" , b"<STR_LIT>" , None ] ) , <EOL> ( server , [ b"<STR_LIT:foo>" , b"<STR_LIT:bar>" ] ) ] ) <EOL> finished . addCallback ( cbComplete ) <EOL> return finished <EOL> def test_identityPumpPolicy ( self ) : <EOL> """<STR_LIT>""" <EOL> bytes = [ ] <EOL> client = Protocol ( ) <EOL> client . dataReceived = bytes . append <EOL> queue = loopback . _LoopbackQueue ( ) <EOL> queue . put ( b"<STR_LIT:foo>" ) <EOL> queue . put ( b"<STR_LIT:bar>" ) <EOL> queue . put ( None ) <EOL> loopback . identityPumpPolicy ( queue , client ) <EOL> self . assertEqual ( bytes , [ b"<STR_LIT:foo>" , b"<STR_LIT:bar>" ] ) <EOL> def test_collapsingPumpPolicy ( self ) : <EOL> """<STR_LIT>""" <EOL> bytes = [ ] <EOL> client = Protocol ( ) <EOL> client . dataReceived = bytes . append <EOL> queue = loopback . _LoopbackQueue ( ) <EOL> queue . put ( b"<STR_LIT:foo>" ) <EOL> queue . put ( b"<STR_LIT:bar>" ) <EOL> queue . put ( None ) <EOL> loopback . collapsingPumpPolicy ( queue , client ) <EOL> self . assertEqual ( bytes , [ b"<STR_LIT>" ] ) <EOL> class LoopbackTCPTestCase ( LoopbackTestCaseMixin , unittest . TestCase ) : <EOL> loopbackFunc = staticmethod ( loopback . loopbackTCP ) <EOL> class LoopbackUNIXTestCase ( LoopbackTestCaseMixin , unittest . TestCase ) : <EOL> loopbackFunc = staticmethod ( loopback . loopbackUNIX ) <EOL> if interfaces . IReactorUNIX ( reactor , None ) is None : <EOL> skip = "<STR_LIT>" <EOL> elif _PY3 : <EOL> skip = "<STR_LIT>" </s>
