<s> """<STR_LIT>""" <EOL> from twisted . python . compat import set <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . internet . defer import Deferred <EOL> from twisted . internet . posixbase import PosixReactorBase , _Waker <EOL> from twisted . internet . protocol import ServerFactory <EOL> skipSockets = None <EOL> try : <EOL> from twisted . internet import unix <EOL> except ImportError : <EOL> skipSockets = "<STR_LIT>" <EOL> from twisted . internet . tcp import Port <EOL> from twisted . internet import reactor <EOL> from twisted . test . test_unix import ClientProto <EOL> class TrivialReactor ( PosixReactorBase ) : <EOL> def __init__ ( self ) : <EOL> self . _readers = { } <EOL> self . _writers = { } <EOL> PosixReactorBase . __init__ ( self ) <EOL> def addReader ( self , reader ) : <EOL> self . _readers [ reader ] = True <EOL> def removeReader ( self , reader ) : <EOL> del self . _readers [ reader ] <EOL> def addWriter ( self , writer ) : <EOL> self . _writers [ writer ] = True <EOL> def removeWriter ( self , writer ) : <EOL> del self . _writers [ writer ] <EOL> class PosixReactorBaseTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _checkWaker ( self , reactor ) : <EOL> self . assertIsInstance ( reactor . waker , _Waker ) <EOL> self . assertIn ( reactor . waker , reactor . _internalReaders ) <EOL> self . assertIn ( reactor . waker , reactor . _readers ) <EOL> def test_wakerIsInternalReader ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TrivialReactor ( ) <EOL> self . _checkWaker ( reactor ) <EOL> def test_removeAllSkipsInternalReaders ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TrivialReactor ( ) <EOL> extra = object ( ) <EOL> reactor . _internalReaders . add ( extra ) <EOL> reactor . addReader ( extra ) <EOL> reactor . _removeAll ( reactor . _readers , reactor . _writers ) <EOL> self . _checkWaker ( reactor ) <EOL> self . assertIn ( extra , reactor . _internalReaders ) <EOL> self . assertIn ( extra , reactor . _readers ) <EOL> def test_removeAllReturnsRemovedDescriptors ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TrivialReactor ( ) <EOL> reader = object ( ) <EOL> writer = object ( ) <EOL> reactor . addReader ( reader ) <EOL> reactor . addWriter ( writer ) <EOL> removed = reactor . _removeAll ( <EOL> reactor . _readers , reactor . _writers ) <EOL> self . assertEqual ( set ( removed ) , set ( [ reader , writer ] ) ) <EOL> self . assertNotIn ( reader , reactor . _readers ) <EOL> self . assertNotIn ( writer , reactor . _writers ) <EOL> def test_IReactorArbitraryIsDeprecated ( self ) : <EOL> """<STR_LIT>""" <EOL> from twisted . internet import interfaces <EOL> interfaces . IReactorArbitrary <EOL> warningsShown = self . flushWarnings ( <EOL> [ self . test_IReactorArbitraryIsDeprecated ] ) <EOL> self . assertEqual ( len ( warningsShown ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( warningsShown [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> warningsShown [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] ) <EOL> def test_listenWithIsDeprecated ( self ) : <EOL> """<STR_LIT>""" <EOL> class fakePort : <EOL> def __init__ ( self , * args , ** kw ) : <EOL> pass <EOL> def startListening ( self ) : <EOL> pass <EOL> reactor = TrivialReactor ( ) <EOL> reactor . listenWith ( fakePort ) <EOL> warnings = self . flushWarnings ( [ self . test_listenWithIsDeprecated ] ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> warnings [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] ) <EOL> def test_connectWithIsDeprecated ( self ) : <EOL> """<STR_LIT>""" <EOL> class fakeConnector : <EOL> def __init__ ( self , * args , ** kw ) : <EOL> pass <EOL> def connect ( self ) : <EOL> pass <EOL> reactor = TrivialReactor ( ) <EOL> reactor . connectWith ( fakeConnector ) <EOL> warnings = self . flushWarnings ( [ self . test_connectWithIsDeprecated ] ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> warnings [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] ) <EOL> class TCPPortTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( reactor , PosixReactorBase ) : <EOL> skip = "<STR_LIT>" <EOL> def test_connectionLostFailed ( self ) : <EOL> """<STR_LIT>""" <EOL> port = Port ( <NUM_LIT> , ServerFactory ( ) ) <EOL> port . connected = True <EOL> port . connectionLost = lambda reason : <NUM_LIT:1> // <NUM_LIT:0> <EOL> return self . assertFailure ( port . stopListening ( ) , ZeroDivisionError ) <EOL> class TimeoutReportReactor ( PosixReactorBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> PosixReactorBase . __init__ ( self ) <EOL> self . iterationTimeout = Deferred ( ) <EOL> self . now = <NUM_LIT:100> <EOL> def addReader ( self , reader ) : <EOL> """<STR_LIT>""" <EOL> def removeAll ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ ] <EOL> def seconds ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . now <EOL> def doIteration ( self , timeout ) : <EOL> d = self . iterationTimeout <EOL> if d is not None : <EOL> self . iterationTimeout = None <EOL> d . callback ( timeout ) <EOL> class IterationTimeoutTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _checkIterationTimeout ( self , reactor ) : <EOL> timeout = [ ] <EOL> reactor . iterationTimeout . addCallback ( timeout . append ) <EOL> reactor . iterationTimeout . addCallback ( lambda ignored : reactor . stop ( ) ) <EOL> reactor . run ( ) <EOL> return timeout [ <NUM_LIT:0> ] <EOL> def test_noCalls ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TimeoutReportReactor ( ) <EOL> timeout = self . _checkIterationTimeout ( reactor ) <EOL> self . assertEqual ( timeout , None ) <EOL> def test_delayedCall ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TimeoutReportReactor ( ) <EOL> reactor . callLater ( <NUM_LIT:100> , lambda : None ) <EOL> timeout = self . _checkIterationTimeout ( reactor ) <EOL> self . assertEqual ( timeout , <NUM_LIT:100> ) <EOL> def test_timePasses ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TimeoutReportReactor ( ) <EOL> reactor . callLater ( <NUM_LIT:100> , lambda : None ) <EOL> reactor . now += <NUM_LIT> <EOL> timeout = self . _checkIterationTimeout ( reactor ) <EOL> self . assertEqual ( timeout , <NUM_LIT> ) <EOL> def test_multipleDelayedCalls ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TimeoutReportReactor ( ) <EOL> reactor . callLater ( <NUM_LIT:50> , lambda : None ) <EOL> reactor . callLater ( <NUM_LIT:10> , lambda : None ) <EOL> reactor . callLater ( <NUM_LIT:100> , lambda : None ) <EOL> timeout = self . _checkIterationTimeout ( reactor ) <EOL> self . assertEqual ( timeout , <NUM_LIT:10> ) <EOL> def test_resetDelayedCall ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TimeoutReportReactor ( ) <EOL> call = reactor . callLater ( <NUM_LIT:50> , lambda : None ) <EOL> reactor . now += <NUM_LIT> <EOL> call . reset ( <NUM_LIT:15> ) <EOL> timeout = self . _checkIterationTimeout ( reactor ) <EOL> self . assertEqual ( timeout , <NUM_LIT:15> ) <EOL> def test_delayDelayedCall ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TimeoutReportReactor ( ) <EOL> call = reactor . callLater ( <NUM_LIT:50> , lambda : None ) <EOL> reactor . now += <NUM_LIT:10> <EOL> call . delay ( <NUM_LIT:20> ) <EOL> timeout = self . _checkIterationTimeout ( reactor ) <EOL> self . assertEqual ( timeout , <NUM_LIT> ) <EOL> def test_cancelDelayedCall ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = TimeoutReportReactor ( ) <EOL> call = reactor . callLater ( <NUM_LIT:50> , lambda : None ) <EOL> call . cancel ( ) <EOL> timeout = self . _checkIterationTimeout ( reactor ) <EOL> self . assertEqual ( timeout , None ) <EOL> class ConnectedDatagramPortTestCase ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> if skipSockets is not None : <EOL> skip = skipSockets <EOL> def test_connectionFailedDoesntCallLoseConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> def loseConnection ( ) : <EOL> """<STR_LIT>""" <EOL> self . fail ( "<STR_LIT>" ) <EOL> port = unix . ConnectedDatagramPort ( None , ClientProto ( ) ) <EOL> port . loseConnection = loseConnection <EOL> port . connectionFailed ( "<STR_LIT>" ) <EOL> def test_connectionFailedCallsStopListening ( self ) : <EOL> """<STR_LIT>""" <EOL> self . called = False <EOL> def stopListening ( ) : <EOL> """<STR_LIT>""" <EOL> self . called = True <EOL> port = unix . ConnectedDatagramPort ( None , ClientProto ( ) ) <EOL> port . stopListening = stopListening <EOL> port . connectionFailed ( "<STR_LIT>" ) <EOL> self . assertEqual ( self . called , True ) </s>
