<s> """<STR_LIT>""" <EOL> import pickle <EOL> from zope . interface import implements <EOL> from zope . interface . verify import verifyClass <EOL> from twisted . internet . protocol import Factory <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . application import internet <EOL> from twisted . application . internet import ( <EOL> StreamServerEndpointService , TimerService ) <EOL> from twisted . internet . interfaces import IStreamServerEndpoint , IListeningPort <EOL> from twisted . internet . defer import Deferred , CancelledError <EOL> from twisted . internet import task <EOL> from twisted . python . failure import Failure <EOL> def fakeTargetFunction ( ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class FakeServer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( IStreamServerEndpoint ) <EOL> result = None <EOL> factory = None <EOL> failImmediately = None <EOL> cancelException = CancelledError ( ) <EOL> listenAttempts = <NUM_LIT:0> <EOL> def __init__ ( self ) : <EOL> self . port = FakePort ( ) <EOL> def listen ( self , factory ) : <EOL> """<STR_LIT>""" <EOL> self . listenAttempts += <NUM_LIT:1> <EOL> self . factory = factory <EOL> self . result = Deferred ( <EOL> canceller = lambda d : d . errback ( self . cancelException ) ) <EOL> if self . failImmediately is not None : <EOL> self . result . errback ( self . failImmediately ) <EOL> return self . result <EOL> def startedListening ( self ) : <EOL> """<STR_LIT>""" <EOL> self . result . callback ( self . port ) <EOL> def stoppedListening ( self ) : <EOL> """<STR_LIT>""" <EOL> self . port . deferred . callback ( None ) <EOL> verifyClass ( IStreamServerEndpoint , FakeServer ) <EOL> class FakePort ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( IListeningPort ) <EOL> deferred = None <EOL> def stopListening ( self ) : <EOL> self . deferred = Deferred ( ) <EOL> return self . deferred <EOL> verifyClass ( IStreamServerEndpoint , FakeServer ) <EOL> class EndpointServiceTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . fakeServer = FakeServer ( ) <EOL> self . factory = Factory ( ) <EOL> self . svc = StreamServerEndpointService ( self . fakeServer , self . factory ) <EOL> def test_privilegedStartService ( self ) : <EOL> """<STR_LIT>""" <EOL> self . svc . privilegedStartService ( ) <EOL> self . assertIdentical ( self . factory , self . fakeServer . factory ) <EOL> def test_synchronousRaiseRaisesSynchronously ( self , thunk = None ) : <EOL> """<STR_LIT>""" <EOL> self . fakeServer . failImmediately = ZeroDivisionError ( ) <EOL> self . svc . _raiseSynchronously = True <EOL> self . assertRaises ( ZeroDivisionError , thunk or self . svc . startService ) <EOL> def test_synchronousRaisePrivileged ( self ) : <EOL> """<STR_LIT>""" <EOL> self . test_synchronousRaiseRaisesSynchronously ( <EOL> self . svc . privilegedStartService ) <EOL> def test_failReportsError ( self ) : <EOL> """<STR_LIT>""" <EOL> self . svc . startService ( ) <EOL> self . fakeServer . result . errback ( ZeroDivisionError ( ) ) <EOL> logged = self . flushLoggedErrors ( ZeroDivisionError ) <EOL> self . assertEqual ( len ( logged ) , <NUM_LIT:1> ) <EOL> def test_synchronousFailReportsError ( self ) : <EOL> """<STR_LIT>""" <EOL> self . fakeServer . failImmediately = ZeroDivisionError ( ) <EOL> self . svc . startService ( ) <EOL> logged = self . flushLoggedErrors ( ZeroDivisionError ) <EOL> self . assertEqual ( len ( logged ) , <NUM_LIT:1> ) <EOL> def test_startServiceUnstarted ( self ) : <EOL> """<STR_LIT>""" <EOL> self . svc . startService ( ) <EOL> self . assertIdentical ( self . factory , self . fakeServer . factory ) <EOL> self . assertEqual ( self . svc . running , True ) <EOL> def test_startServiceStarted ( self ) : <EOL> """<STR_LIT>""" <EOL> self . test_privilegedStartService ( ) <EOL> self . svc . startService ( ) <EOL> self . assertEqual ( self . fakeServer . listenAttempts , <NUM_LIT:1> ) <EOL> self . assertEqual ( self . svc . running , True ) <EOL> def test_stopService ( self ) : <EOL> """<STR_LIT>""" <EOL> self . svc . privilegedStartService ( ) <EOL> self . fakeServer . startedListening ( ) <EOL> self . svc . startService ( ) <EOL> result = self . svc . stopService ( ) <EOL> l = [ ] <EOL> result . addCallback ( l . append ) <EOL> self . assertEqual ( len ( l ) , <NUM_LIT:0> ) <EOL> self . fakeServer . stoppedListening ( ) <EOL> self . assertEqual ( len ( l ) , <NUM_LIT:1> ) <EOL> self . assertFalse ( self . svc . running ) <EOL> def test_stopServiceBeforeStartFinished ( self ) : <EOL> """<STR_LIT>""" <EOL> self . svc . privilegedStartService ( ) <EOL> result = self . svc . stopService ( ) <EOL> l = [ ] <EOL> result . addBoth ( l . append ) <EOL> self . assertEqual ( l , [ None ] ) <EOL> self . assertEqual ( self . flushLoggedErrors ( CancelledError ) , [ ] ) <EOL> def test_stopServiceCancelStartError ( self ) : <EOL> """<STR_LIT>""" <EOL> self . fakeServer . cancelException = ZeroDivisionError ( ) <EOL> self . svc . privilegedStartService ( ) <EOL> result = self . svc . stopService ( ) <EOL> l = [ ] <EOL> result . addCallback ( l . append ) <EOL> self . assertEqual ( l , [ None ] ) <EOL> stoppingErrors = self . flushLoggedErrors ( ZeroDivisionError ) <EOL> self . assertEqual ( len ( stoppingErrors ) , <NUM_LIT:1> ) <EOL> class TimerServiceTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . timer = TimerService ( <NUM_LIT:2> , self . call ) <EOL> self . clock = self . timer . clock = task . Clock ( ) <EOL> self . deferred = Deferred ( ) <EOL> def call ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . deferred <EOL> def test_startService ( self ) : <EOL> """<STR_LIT>""" <EOL> self . timer . startService ( ) <EOL> self . assertTrue ( self . timer . running , "<STR_LIT>" ) <EOL> self . assertIsInstance ( self . timer . _loop , task . LoopingCall ) <EOL> self . assertIdentical ( self . clock , self . timer . _loop . clock ) <EOL> self . assertTrue ( self . timer . _loop . running , "<STR_LIT>" ) <EOL> def test_startServiceRunsCallImmediately ( self ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> self . timer . call = ( result . append , ( None , ) , { } ) <EOL> self . timer . startService ( ) <EOL> self . assertEqual ( [ None ] , result ) <EOL> def test_startServiceUsesGlobalReactor ( self ) : <EOL> """<STR_LIT>""" <EOL> otherClock = task . Clock ( ) <EOL> def getOtherClock ( maybeReactor ) : <EOL> return otherClock <EOL> self . patch ( internet , "<STR_LIT>" , getOtherClock ) <EOL> self . timer . startService ( ) <EOL> self . assertIdentical ( otherClock , self . timer . _loop . clock ) <EOL> def test_stopServiceWaits ( self ) : <EOL> """<STR_LIT>""" <EOL> self . timer . startService ( ) <EOL> d = self . timer . stopService ( ) <EOL> self . assertNoResult ( d ) <EOL> self . assertEqual ( True , self . timer . running ) <EOL> self . deferred . callback ( object ( ) ) <EOL> self . assertIdentical ( self . successResultOf ( d ) , None ) <EOL> def test_stopServiceImmediately ( self ) : <EOL> """<STR_LIT>""" <EOL> self . timer . startService ( ) <EOL> self . deferred . callback ( object ( ) ) <EOL> d = self . timer . stopService ( ) <EOL> self . assertIdentical ( self . successResultOf ( d ) , None ) <EOL> def test_failedCallLogsError ( self ) : <EOL> """<STR_LIT>""" <EOL> self . timer . startService ( ) <EOL> self . deferred . errback ( Failure ( ZeroDivisionError ( ) ) ) <EOL> errors = self . flushLoggedErrors ( ZeroDivisionError ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( errors ) ) <EOL> d = self . timer . stopService ( ) <EOL> self . assertIdentical ( self . successResultOf ( d ) , None ) <EOL> def test_pickleTimerServiceNotPickleLoop ( self ) : <EOL> """<STR_LIT>""" <EOL> timer = TimerService ( <NUM_LIT:1> , fakeTargetFunction ) <EOL> timer . startService ( ) <EOL> dumpedTimer = pickle . dumps ( timer ) <EOL> timer . stopService ( ) <EOL> loadedTimer = pickle . loads ( dumpedTimer ) <EOL> nothing = object ( ) <EOL> value = getattr ( loadedTimer , "<STR_LIT>" , nothing ) <EOL> self . assertIdentical ( nothing , value ) <EOL> def test_pickleTimerServiceNotPickleLoopFinished ( self ) : <EOL> """<STR_LIT>""" <EOL> timer = TimerService ( <NUM_LIT:1> , fakeTargetFunction ) <EOL> timer . startService ( ) <EOL> dumpedTimer = pickle . dumps ( timer ) <EOL> timer . stopService ( ) <EOL> loadedTimer = pickle . loads ( dumpedTimer ) <EOL> nothing = object ( ) <EOL> value = getattr ( loadedTimer , "<STR_LIT>" , nothing ) <EOL> self . assertIdentical ( nothing , value ) </s>
