<s> """<STR_LIT>""" <EOL> import copy , os , pickle <EOL> from StringIO import StringIO <EOL> from twisted . trial import unittest , util <EOL> from twisted . application import service , internet , app <EOL> from twisted . persisted import sob <EOL> from twisted . python import usage <EOL> from twisted . internet import interfaces , defer <EOL> from twisted . protocols import wire , basic <EOL> from twisted . internet import protocol , reactor <EOL> from twisted . application import reactors <EOL> from twisted . test . proto_helpers import MemoryReactor <EOL> class Dummy : <EOL> processName = None <EOL> class TestService ( unittest . TestCase ) : <EOL> def testName ( self ) : <EOL> s = service . Service ( ) <EOL> s . setName ( "<STR_LIT:hello>" ) <EOL> self . assertEqual ( s . name , "<STR_LIT:hello>" ) <EOL> def testParent ( self ) : <EOL> s = service . Service ( ) <EOL> p = service . MultiService ( ) <EOL> s . setServiceParent ( p ) <EOL> self . assertEqual ( list ( p ) , [ s ] ) <EOL> self . assertEqual ( s . parent , p ) <EOL> def testApplicationAsParent ( self ) : <EOL> s = service . Service ( ) <EOL> p = service . Application ( "<STR_LIT>" ) <EOL> s . setServiceParent ( p ) <EOL> self . assertEqual ( list ( service . IServiceCollection ( p ) ) , [ s ] ) <EOL> self . assertEqual ( s . parent , service . IServiceCollection ( p ) ) <EOL> def testNamedChild ( self ) : <EOL> s = service . Service ( ) <EOL> p = service . MultiService ( ) <EOL> s . setName ( "<STR_LIT:hello>" ) <EOL> s . setServiceParent ( p ) <EOL> self . assertEqual ( list ( p ) , [ s ] ) <EOL> self . assertEqual ( s . parent , p ) <EOL> self . assertEqual ( p . getServiceNamed ( "<STR_LIT:hello>" ) , s ) <EOL> def testDoublyNamedChild ( self ) : <EOL> s = service . Service ( ) <EOL> p = service . MultiService ( ) <EOL> s . setName ( "<STR_LIT:hello>" ) <EOL> s . setServiceParent ( p ) <EOL> self . failUnlessRaises ( RuntimeError , s . setName , "<STR_LIT>" ) <EOL> def testDuplicateNamedChild ( self ) : <EOL> s = service . Service ( ) <EOL> p = service . MultiService ( ) <EOL> s . setName ( "<STR_LIT:hello>" ) <EOL> s . setServiceParent ( p ) <EOL> s = service . Service ( ) <EOL> s . setName ( "<STR_LIT:hello>" ) <EOL> self . failUnlessRaises ( RuntimeError , s . setServiceParent , p ) <EOL> def testDisowning ( self ) : <EOL> s = service . Service ( ) <EOL> p = service . MultiService ( ) <EOL> s . setServiceParent ( p ) <EOL> self . assertEqual ( list ( p ) , [ s ] ) <EOL> self . assertEqual ( s . parent , p ) <EOL> s . disownServiceParent ( ) <EOL> self . assertEqual ( list ( p ) , [ ] ) <EOL> self . assertEqual ( s . parent , None ) <EOL> def testRunning ( self ) : <EOL> s = service . Service ( ) <EOL> self . assert_ ( not s . running ) <EOL> s . startService ( ) <EOL> self . assert_ ( s . running ) <EOL> s . stopService ( ) <EOL> self . assert_ ( not s . running ) <EOL> def testRunningChildren1 ( self ) : <EOL> s = service . Service ( ) <EOL> p = service . MultiService ( ) <EOL> s . setServiceParent ( p ) <EOL> self . assert_ ( not s . running ) <EOL> self . assert_ ( not p . running ) <EOL> p . startService ( ) <EOL> self . assert_ ( s . running ) <EOL> self . assert_ ( p . running ) <EOL> p . stopService ( ) <EOL> self . assert_ ( not s . running ) <EOL> self . assert_ ( not p . running ) <EOL> def testRunningChildren2 ( self ) : <EOL> s = service . Service ( ) <EOL> def checkRunning ( ) : <EOL> self . assert_ ( s . running ) <EOL> t = service . Service ( ) <EOL> t . stopService = checkRunning <EOL> t . startService = checkRunning <EOL> p = service . MultiService ( ) <EOL> s . setServiceParent ( p ) <EOL> t . setServiceParent ( p ) <EOL> p . startService ( ) <EOL> p . stopService ( ) <EOL> def testAddingIntoRunning ( self ) : <EOL> p = service . MultiService ( ) <EOL> p . startService ( ) <EOL> s = service . Service ( ) <EOL> self . assert_ ( not s . running ) <EOL> s . setServiceParent ( p ) <EOL> self . assert_ ( s . running ) <EOL> s . disownServiceParent ( ) <EOL> self . assert_ ( not s . running ) <EOL> def testPrivileged ( self ) : <EOL> s = service . Service ( ) <EOL> def pss ( ) : <EOL> s . privilegedStarted = <NUM_LIT:1> <EOL> s . privilegedStartService = pss <EOL> s1 = service . Service ( ) <EOL> p = service . MultiService ( ) <EOL> s . setServiceParent ( p ) <EOL> s1 . setServiceParent ( p ) <EOL> p . privilegedStartService ( ) <EOL> self . assert_ ( s . privilegedStarted ) <EOL> def testCopying ( self ) : <EOL> s = service . Service ( ) <EOL> s . startService ( ) <EOL> s1 = copy . copy ( s ) <EOL> self . assert_ ( not s1 . running ) <EOL> self . assert_ ( s . running ) <EOL> if hasattr ( os , "<STR_LIT>" ) : <EOL> curuid = os . getuid ( ) <EOL> curgid = os . getgid ( ) <EOL> else : <EOL> curuid = curgid = <NUM_LIT:0> <EOL> class TestProcess ( unittest . TestCase ) : <EOL> def testID ( self ) : <EOL> p = service . Process ( <NUM_LIT:5> , <NUM_LIT:6> ) <EOL> self . assertEqual ( p . uid , <NUM_LIT:5> ) <EOL> self . assertEqual ( p . gid , <NUM_LIT:6> ) <EOL> def testDefaults ( self ) : <EOL> p = service . Process ( <NUM_LIT:5> ) <EOL> self . assertEqual ( p . uid , <NUM_LIT:5> ) <EOL> self . assertEqual ( p . gid , None ) <EOL> p = service . Process ( gid = <NUM_LIT:5> ) <EOL> self . assertEqual ( p . uid , None ) <EOL> self . assertEqual ( p . gid , <NUM_LIT:5> ) <EOL> p = service . Process ( ) <EOL> self . assertEqual ( p . uid , None ) <EOL> self . assertEqual ( p . gid , None ) <EOL> def testProcessName ( self ) : <EOL> p = service . Process ( ) <EOL> self . assertEqual ( p . processName , None ) <EOL> p . processName = '<STR_LIT:hello>' <EOL> self . assertEqual ( p . processName , '<STR_LIT:hello>' ) <EOL> class TestInterfaces ( unittest . TestCase ) : <EOL> def testService ( self ) : <EOL> self . assert_ ( service . IService . providedBy ( service . Service ( ) ) ) <EOL> def testMultiService ( self ) : <EOL> self . assert_ ( service . IService . providedBy ( service . MultiService ( ) ) ) <EOL> self . assert_ ( service . IServiceCollection . providedBy ( service . MultiService ( ) ) ) <EOL> def testProcess ( self ) : <EOL> self . assert_ ( service . IProcess . providedBy ( service . Process ( ) ) ) <EOL> class TestApplication ( unittest . TestCase ) : <EOL> def testConstructor ( self ) : <EOL> service . Application ( "<STR_LIT:hello>" ) <EOL> service . Application ( "<STR_LIT:hello>" , <NUM_LIT:5> ) <EOL> service . Application ( "<STR_LIT:hello>" , <NUM_LIT:5> , <NUM_LIT:6> ) <EOL> def testProcessComponent ( self ) : <EOL> a = service . Application ( "<STR_LIT:hello>" ) <EOL> self . assertEqual ( service . IProcess ( a ) . uid , None ) <EOL> self . assertEqual ( service . IProcess ( a ) . gid , None ) <EOL> a = service . Application ( "<STR_LIT:hello>" , <NUM_LIT:5> ) <EOL> self . assertEqual ( service . IProcess ( a ) . uid , <NUM_LIT:5> ) <EOL> self . assertEqual ( service . IProcess ( a ) . gid , None ) <EOL> a = service . Application ( "<STR_LIT:hello>" , <NUM_LIT:5> , <NUM_LIT:6> ) <EOL> self . assertEqual ( service . IProcess ( a ) . uid , <NUM_LIT:5> ) <EOL> self . assertEqual ( service . IProcess ( a ) . gid , <NUM_LIT:6> ) <EOL> def testServiceComponent ( self ) : <EOL> a = service . Application ( "<STR_LIT:hello>" ) <EOL> self . assert_ ( service . IService ( a ) is service . IServiceCollection ( a ) ) <EOL> self . assertEqual ( service . IService ( a ) . name , "<STR_LIT:hello>" ) <EOL> self . assertEqual ( service . IService ( a ) . parent , None ) <EOL> def testPersistableComponent ( self ) : <EOL> a = service . Application ( "<STR_LIT:hello>" ) <EOL> p = sob . IPersistable ( a ) <EOL> self . assertEqual ( p . style , '<STR_LIT>' ) <EOL> self . assertEqual ( p . name , '<STR_LIT:hello>' ) <EOL> self . assert_ ( p . original is a ) <EOL> class TestLoading ( unittest . TestCase ) : <EOL> def test_simpleStoreAndLoad ( self ) : <EOL> a = service . Application ( "<STR_LIT:hello>" ) <EOL> p = sob . IPersistable ( a ) <EOL> for style in '<STR_LIT>' . split ( ) : <EOL> p . setStyle ( style ) <EOL> p . save ( ) <EOL> a1 = service . loadApplication ( "<STR_LIT>" + style [ <NUM_LIT:0> ] , style ) <EOL> self . assertEqual ( service . IService ( a1 ) . name , "<STR_LIT:hello>" ) <EOL> f = open ( "<STR_LIT>" , '<STR_LIT:w>' ) <EOL> f . writelines ( [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] ) <EOL> f . close ( ) <EOL> a1 = service . loadApplication ( "<STR_LIT>" , '<STR_LIT>' ) <EOL> self . assertEqual ( service . IService ( a1 ) . name , "<STR_LIT:hello>" ) <EOL> class TestAppSupport ( unittest . TestCase ) : <EOL> def testPassphrase ( self ) : <EOL> self . assertEqual ( app . getPassphrase ( <NUM_LIT:0> ) , None ) <EOL> def testLoadApplication ( self ) : <EOL> """<STR_LIT>""" <EOL> a = service . Application ( "<STR_LIT:hello>" ) <EOL> baseconfig = { '<STR_LIT:file>' : None , '<STR_LIT:source>' : None , '<STR_LIT>' : None } <EOL> for style in '<STR_LIT>' . split ( ) : <EOL> config = baseconfig . copy ( ) <EOL> config [ { '<STR_LIT>' : '<STR_LIT:file>' } . get ( style , style ) ] = '<STR_LIT>' <EOL> sob . IPersistable ( a ) . setStyle ( style ) <EOL> sob . IPersistable ( a ) . save ( filename = '<STR_LIT>' ) <EOL> a1 = app . getApplication ( config , None ) <EOL> self . assertEqual ( service . IService ( a1 ) . name , "<STR_LIT:hello>" ) <EOL> config = baseconfig . copy ( ) <EOL> config [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> f = open ( "<STR_LIT>" , '<STR_LIT:w>' ) <EOL> f . writelines ( [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] ) <EOL> f . close ( ) <EOL> a1 = app . getApplication ( config , None ) <EOL> self . assertEqual ( service . IService ( a1 ) . name , "<STR_LIT:hello>" ) <EOL> def test_convertStyle ( self ) : <EOL> appl = service . Application ( "<STR_LIT>" ) <EOL> for instyle in '<STR_LIT>' . split ( ) : <EOL> for outstyle in '<STR_LIT>' . split ( ) : <EOL> sob . IPersistable ( appl ) . setStyle ( instyle ) <EOL> sob . IPersistable ( appl ) . save ( filename = "<STR_LIT>" ) <EOL> app . convertStyle ( "<STR_LIT>" , instyle , None , <EOL> "<STR_LIT>" , outstyle , <NUM_LIT:0> ) <EOL> appl2 = service . loadApplication ( "<STR_LIT>" , outstyle ) <EOL> self . assertEqual ( service . IService ( appl2 ) . name , "<STR_LIT>" ) <EOL> def test_startApplication ( self ) : <EOL> appl = service . Application ( "<STR_LIT>" ) <EOL> app . startApplication ( appl , <NUM_LIT:0> ) <EOL> self . assert_ ( service . IService ( appl ) . running ) <EOL> class Foo ( basic . LineReceiver ) : <EOL> def connectionMade ( self ) : <EOL> self . transport . write ( '<STR_LIT>' ) <EOL> def lineReceived ( self , line ) : <EOL> self . factory . line = line <EOL> self . transport . loseConnection ( ) <EOL> def connectionLost ( self , reason ) : <EOL> self . factory . d . callback ( self . factory . line ) <EOL> class DummyApp : <EOL> processName = None <EOL> def addService ( self , service ) : <EOL> self . services [ service . name ] = service <EOL> def removeService ( self , service ) : <EOL> del self . services [ service . name ] <EOL> class TimerTarget : <EOL> def __init__ ( self ) : <EOL> self . l = [ ] <EOL> def append ( self , what ) : <EOL> self . l . append ( what ) <EOL> class TestEcho ( wire . Echo ) : <EOL> def connectionLost ( self , reason ) : <EOL> self . d . callback ( True ) <EOL> class TestInternet2 ( unittest . TestCase ) : <EOL> def testTCP ( self ) : <EOL> s = service . MultiService ( ) <EOL> s . startService ( ) <EOL> factory = protocol . ServerFactory ( ) <EOL> factory . protocol = TestEcho <EOL> TestEcho . d = defer . Deferred ( ) <EOL> t = internet . TCPServer ( <NUM_LIT:0> , factory ) <EOL> t . setServiceParent ( s ) <EOL> num = t . _port . getHost ( ) . port <EOL> factory = protocol . ClientFactory ( ) <EOL> factory . d = defer . Deferred ( ) <EOL> factory . protocol = Foo <EOL> factory . line = None <EOL> internet . TCPClient ( '<STR_LIT:127.0.0.1>' , num , factory ) . setServiceParent ( s ) <EOL> factory . d . addCallback ( self . assertEqual , '<STR_LIT>' ) <EOL> factory . d . addCallback ( lambda x : s . stopService ( ) ) <EOL> factory . d . addCallback ( lambda x : TestEcho . d ) <EOL> return factory . d <EOL> def test_UDP ( self ) : <EOL> """<STR_LIT>""" <EOL> if not interfaces . IReactorUDP ( reactor , None ) : <EOL> raise unittest . SkipTest ( "<STR_LIT>" ) <EOL> p = protocol . DatagramProtocol ( ) <EOL> t = internet . UDPServer ( <NUM_LIT:0> , p ) <EOL> t . startService ( ) <EOL> num = t . _port . getHost ( ) . port <EOL> self . assertNotEquals ( num , <NUM_LIT:0> ) <EOL> def onStop ( ignored ) : <EOL> t = internet . UDPServer ( num , p ) <EOL> t . startService ( ) <EOL> return t . stopService ( ) <EOL> return defer . maybeDeferred ( t . stopService ) . addCallback ( onStop ) <EOL> def testPrivileged ( self ) : <EOL> factory = protocol . ServerFactory ( ) <EOL> factory . protocol = TestEcho <EOL> TestEcho . d = defer . Deferred ( ) <EOL> t = internet . TCPServer ( <NUM_LIT:0> , factory ) <EOL> t . privileged = <NUM_LIT:1> <EOL> t . privilegedStartService ( ) <EOL> num = t . _port . getHost ( ) . port <EOL> factory = protocol . ClientFactory ( ) <EOL> factory . d = defer . Deferred ( ) <EOL> factory . protocol = Foo <EOL> factory . line = None <EOL> c = internet . TCPClient ( '<STR_LIT:127.0.0.1>' , num , factory ) <EOL> c . startService ( ) <EOL> factory . d . addCallback ( self . assertEqual , '<STR_LIT>' ) <EOL> factory . d . addCallback ( lambda x : c . stopService ( ) ) <EOL> factory . d . addCallback ( lambda x : t . stopService ( ) ) <EOL> factory . d . addCallback ( lambda x : TestEcho . d ) <EOL> return factory . d <EOL> def testConnectionGettingRefused ( self ) : <EOL> factory = protocol . ServerFactory ( ) <EOL> factory . protocol = wire . Echo <EOL> t = internet . TCPServer ( <NUM_LIT:0> , factory ) <EOL> t . startService ( ) <EOL> num = t . _port . getHost ( ) . port <EOL> t . stopService ( ) <EOL> d = defer . Deferred ( ) <EOL> factory = protocol . ClientFactory ( ) <EOL> factory . clientConnectionFailed = lambda * args : d . callback ( None ) <EOL> c = internet . TCPClient ( '<STR_LIT:127.0.0.1>' , num , factory ) <EOL> c . startService ( ) <EOL> return d <EOL> def testUNIX ( self ) : <EOL> if not interfaces . IReactorUNIX ( reactor , None ) : <EOL> raise unittest . SkipTest , "<STR_LIT>" <EOL> s = service . MultiService ( ) <EOL> s . startService ( ) <EOL> factory = protocol . ServerFactory ( ) <EOL> factory . protocol = TestEcho <EOL> TestEcho . d = defer . Deferred ( ) <EOL> t = internet . UNIXServer ( '<STR_LIT>' , factory ) <EOL> t . setServiceParent ( s ) <EOL> factory = protocol . ClientFactory ( ) <EOL> factory . protocol = Foo <EOL> factory . d = defer . Deferred ( ) <EOL> factory . line = None <EOL> internet . UNIXClient ( '<STR_LIT>' , factory ) . setServiceParent ( s ) <EOL> factory . d . addCallback ( self . assertEqual , '<STR_LIT>' ) <EOL> factory . d . addCallback ( lambda x : s . stopService ( ) ) <EOL> factory . d . addCallback ( lambda x : TestEcho . d ) <EOL> factory . d . addCallback ( self . _cbTestUnix , factory , s ) <EOL> return factory . d <EOL> def _cbTestUnix ( self , ignored , factory , s ) : <EOL> TestEcho . d = defer . Deferred ( ) <EOL> factory . line = None <EOL> factory . d = defer . Deferred ( ) <EOL> s . startService ( ) <EOL> factory . d . addCallback ( self . assertEqual , '<STR_LIT>' ) <EOL> factory . d . addCallback ( lambda x : s . stopService ( ) ) <EOL> factory . d . addCallback ( lambda x : TestEcho . d ) <EOL> return factory . d <EOL> def testVolatile ( self ) : <EOL> if not interfaces . IReactorUNIX ( reactor , None ) : <EOL> raise unittest . SkipTest , "<STR_LIT>" <EOL> factory = protocol . ServerFactory ( ) <EOL> factory . protocol = wire . Echo <EOL> t = internet . UNIXServer ( '<STR_LIT>' , factory ) <EOL> t . startService ( ) <EOL> self . failIfIdentical ( t . _port , None ) <EOL> t1 = copy . copy ( t ) <EOL> self . assertIdentical ( t1 . _port , None ) <EOL> t . stopService ( ) <EOL> self . assertIdentical ( t . _port , None ) <EOL> self . failIf ( t . running ) <EOL> factory = protocol . ClientFactory ( ) <EOL> factory . protocol = wire . Echo <EOL> t = internet . UNIXClient ( '<STR_LIT>' , factory ) <EOL> t . startService ( ) <EOL> self . failIfIdentical ( t . _connection , None ) <EOL> t1 = copy . copy ( t ) <EOL> self . assertIdentical ( t1 . _connection , None ) <EOL> t . stopService ( ) <EOL> self . assertIdentical ( t . _connection , None ) <EOL> self . failIf ( t . running ) <EOL> def testStoppingServer ( self ) : <EOL> if not interfaces . IReactorUNIX ( reactor , None ) : <EOL> raise unittest . SkipTest , "<STR_LIT>" <EOL> factory = protocol . ServerFactory ( ) <EOL> factory . protocol = wire . Echo <EOL> t = internet . UNIXServer ( '<STR_LIT>' , factory ) <EOL> t . startService ( ) <EOL> t . stopService ( ) <EOL> self . failIf ( t . running ) <EOL> factory = protocol . ClientFactory ( ) <EOL> d = defer . Deferred ( ) <EOL> factory . clientConnectionFailed = lambda * args : d . callback ( None ) <EOL> reactor . connectUNIX ( '<STR_LIT>' , factory ) <EOL> return d <EOL> def testPickledTimer ( self ) : <EOL> target = TimerTarget ( ) <EOL> t0 = internet . TimerService ( <NUM_LIT:1> , target . append , "<STR_LIT:hello>" ) <EOL> t0 . startService ( ) <EOL> s = pickle . dumps ( t0 ) <EOL> t0 . stopService ( ) <EOL> t = pickle . loads ( s ) <EOL> self . failIf ( t . running ) <EOL> def testBrokenTimer ( self ) : <EOL> d = defer . Deferred ( ) <EOL> t = internet . TimerService ( <NUM_LIT:1> , lambda : <NUM_LIT:1> // <NUM_LIT:0> ) <EOL> oldFailed = t . _failed <EOL> def _failed ( why ) : <EOL> oldFailed ( why ) <EOL> d . callback ( None ) <EOL> t . _failed = _failed <EOL> t . startService ( ) <EOL> d . addCallback ( lambda x : t . stopService ) <EOL> d . addCallback ( lambda x : self . assertEqual ( <EOL> [ ZeroDivisionError ] , <EOL> [ o . value . __class__ for o in self . flushLoggedErrors ( ZeroDivisionError ) ] ) ) <EOL> return d <EOL> def test_genericServerDeprecated ( self ) : <EOL> """<STR_LIT>""" <EOL> internet . GenericServer ( ) <EOL> warnings = self . flushWarnings ( <EOL> offendingFunctions = [ self . test_genericServerDeprecated ] ) <EOL> self . assertEqual ( <EOL> warnings [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> def test_genericClientDeprecated ( self ) : <EOL> """<STR_LIT>""" <EOL> internet . GenericClient ( ) <EOL> warnings = self . flushWarnings ( <EOL> offendingFunctions = [ self . test_genericClientDeprecated ] ) <EOL> self . assertEqual ( <EOL> warnings [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( warnings [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , DeprecationWarning ) <EOL> self . assertEqual ( len ( warnings ) , <NUM_LIT:1> ) <EOL> def test_everythingThere ( self ) : <EOL> """<STR_LIT>""" <EOL> trans = '<STR_LIT>' . split ( ) <EOL> for tran in trans [ : ] : <EOL> if not getattr ( interfaces , "<STR_LIT>" + tran ) ( reactor , None ) : <EOL> trans . remove ( tran ) <EOL> if interfaces . IReactorArbitrary ( reactor , None ) is not None : <EOL> trans . insert ( <NUM_LIT:0> , "<STR_LIT>" ) <EOL> for tran in trans : <EOL> for side in '<STR_LIT>' . split ( ) : <EOL> if tran == "<STR_LIT>" and side == "<STR_LIT>" : <EOL> continue <EOL> self . assertTrue ( hasattr ( internet , tran + side ) ) <EOL> method = getattr ( internet , tran + side ) . method <EOL> prefix = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } [ side ] <EOL> self . assertTrue ( hasattr ( reactor , prefix + method ) or <EOL> ( prefix == "<STR_LIT>" and method == "<STR_LIT>" ) ) <EOL> o = getattr ( internet , tran + side ) ( ) <EOL> self . assertEqual ( service . IService ( o ) , o ) <EOL> test_everythingThere . suppress = [ <EOL> util . suppress ( message = '<STR_LIT>' , <EOL> category = DeprecationWarning ) , <EOL> util . suppress ( message = '<STR_LIT>' , <EOL> category = DeprecationWarning ) , <EOL> util . suppress ( message = '<STR_LIT>' <EOL> '<STR_LIT>' ) ] <EOL> def test_importAll ( self ) : <EOL> """<STR_LIT>""" <EOL> for cls in internet . __all__ : <EOL> self . assertTrue ( <EOL> hasattr ( internet , cls ) , <EOL> '<STR_LIT>' % ( cls , ) ) <EOL> def test_reactorParametrizationInServer ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = MemoryReactor ( ) <EOL> factory = object ( ) <EOL> t = internet . TCPServer ( <NUM_LIT> , factory , reactor = reactor ) <EOL> t . startService ( ) <EOL> self . assertEqual ( reactor . tcpServers . pop ( ) [ : <NUM_LIT:2> ] , ( <NUM_LIT> , factory ) ) <EOL> def test_reactorParametrizationInClient ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = MemoryReactor ( ) <EOL> factory = protocol . ClientFactory ( ) <EOL> t = internet . TCPClient ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> , factory , reactor = reactor ) <EOL> t . startService ( ) <EOL> self . assertEqual ( <EOL> reactor . tcpClients . pop ( ) [ : <NUM_LIT:3> ] , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> , factory ) ) <EOL> def test_reactorParametrizationInServerMultipleStart ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = MemoryReactor ( ) <EOL> factory = protocol . Factory ( ) <EOL> t = internet . TCPServer ( <NUM_LIT> , factory , reactor = reactor ) <EOL> t . startService ( ) <EOL> self . assertEqual ( reactor . tcpServers . pop ( ) [ : <NUM_LIT:2> ] , ( <NUM_LIT> , factory ) ) <EOL> t . stopService ( ) <EOL> t . startService ( ) <EOL> self . assertEqual ( reactor . tcpServers . pop ( ) [ : <NUM_LIT:2> ] , ( <NUM_LIT> , factory ) ) <EOL> def test_reactorParametrizationInClientMultipleStart ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = MemoryReactor ( ) <EOL> factory = protocol . ClientFactory ( ) <EOL> t = internet . TCPClient ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> , factory , reactor = reactor ) <EOL> t . startService ( ) <EOL> self . assertEqual ( <EOL> reactor . tcpClients . pop ( ) [ : <NUM_LIT:3> ] , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> , factory ) ) <EOL> t . stopService ( ) <EOL> t . startService ( ) <EOL> self . assertEqual ( <EOL> reactor . tcpClients . pop ( ) [ : <NUM_LIT:3> ] , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> , factory ) ) <EOL> class TestTimerBasic ( unittest . TestCase ) : <EOL> def testTimerRuns ( self ) : <EOL> d = defer . Deferred ( ) <EOL> self . t = internet . TimerService ( <NUM_LIT:1> , d . callback , '<STR_LIT:hello>' ) <EOL> self . t . startService ( ) <EOL> d . addCallback ( self . assertEqual , '<STR_LIT:hello>' ) <EOL> d . addCallback ( lambda x : self . t . stopService ( ) ) <EOL> d . addCallback ( lambda x : self . failIf ( self . t . running ) ) <EOL> return d <EOL> def tearDown ( self ) : <EOL> return self . t . stopService ( ) <EOL> def testTimerRestart ( self ) : <EOL> d1 = defer . Deferred ( ) <EOL> d2 = defer . Deferred ( ) <EOL> work = [ ( d2 , "<STR_LIT:bar>" ) , ( d1 , "<STR_LIT:foo>" ) ] <EOL> def trigger ( ) : <EOL> d , arg = work . pop ( ) <EOL> d . callback ( arg ) <EOL> self . t = internet . TimerService ( <NUM_LIT:1> , trigger ) <EOL> self . t . startService ( ) <EOL> def onFirstResult ( result ) : <EOL> self . assertEqual ( result , '<STR_LIT:foo>' ) <EOL> return self . t . stopService ( ) <EOL> def onFirstStop ( ignored ) : <EOL> self . failIf ( self . t . running ) <EOL> self . t . startService ( ) <EOL> return d2 <EOL> def onSecondResult ( result ) : <EOL> self . assertEqual ( result , '<STR_LIT:bar>' ) <EOL> self . t . stopService ( ) <EOL> d1 . addCallback ( onFirstResult ) <EOL> d1 . addCallback ( onFirstStop ) <EOL> d1 . addCallback ( onSecondResult ) <EOL> return d1 <EOL> def testTimerLoops ( self ) : <EOL> l = [ ] <EOL> def trigger ( data , number , d ) : <EOL> l . append ( data ) <EOL> if len ( l ) == number : <EOL> d . callback ( l ) <EOL> d = defer . Deferred ( ) <EOL> self . t = internet . TimerService ( <NUM_LIT> , trigger , "<STR_LIT:hello>" , <NUM_LIT:10> , d ) <EOL> self . t . startService ( ) <EOL> d . addCallback ( self . assertEqual , [ '<STR_LIT:hello>' ] * <NUM_LIT:10> ) <EOL> d . addCallback ( lambda x : self . t . stopService ( ) ) <EOL> return d <EOL> class FakeReactor ( reactors . Reactor ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , install , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> reactors . Reactor . __init__ ( self , * args , ** kwargs ) <EOL> self . install = install <EOL> class PluggableReactorTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . pluginCalls = [ ] <EOL> self . pluginResults = [ ] <EOL> self . originalFunction = reactors . getPlugins <EOL> reactors . getPlugins = self . _getPlugins <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> reactors . getPlugins = self . originalFunction <EOL> def _getPlugins ( self , interface , package = None ) : <EOL> """<STR_LIT>""" <EOL> self . pluginCalls . append ( ( interface , package ) ) <EOL> return list ( self . pluginResults ) <EOL> def test_getPluginReactorTypes ( self ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> package = __name__ + '<STR_LIT>' <EOL> description = '<STR_LIT:description>' <EOL> self . pluginResults = [ reactors . Reactor ( name , package , description ) ] <EOL> reactorTypes = reactors . getReactorTypes ( ) <EOL> self . assertEqual ( <EOL> self . pluginCalls , <EOL> [ ( reactors . IReactorInstaller , None ) ] ) <EOL> for r in reactorTypes : <EOL> if r . shortName == name : <EOL> self . assertEqual ( r . description , description ) <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_reactorInstallation ( self ) : <EOL> """<STR_LIT>""" <EOL> installed = [ ] <EOL> def install ( ) : <EOL> installed . append ( True ) <EOL> installer = FakeReactor ( install , <EOL> '<STR_LIT>' , __name__ , '<STR_LIT>' ) <EOL> installer . install ( ) <EOL> self . assertEqual ( installed , [ True ] ) <EOL> def test_installReactor ( self ) : <EOL> """<STR_LIT>""" <EOL> installed = [ ] <EOL> def install ( ) : <EOL> installed . append ( True ) <EOL> name = '<STR_LIT>' <EOL> package = __name__ <EOL> description = '<STR_LIT:description>' <EOL> self . pluginResults = [ FakeReactor ( install , name , package , description ) ] <EOL> reactors . installReactor ( name ) <EOL> self . assertEqual ( installed , [ True ] ) <EOL> def test_installNonExistentReactor ( self ) : <EOL> """<STR_LIT>""" <EOL> self . pluginResults = [ ] <EOL> self . assertRaises ( <EOL> reactors . NoSuchReactor , <EOL> reactors . installReactor , '<STR_LIT>' ) <EOL> def test_installNotAvailableReactor ( self ) : <EOL> """<STR_LIT>""" <EOL> def install ( ) : <EOL> raise ImportError ( "<STR_LIT>" ) <EOL> name = '<STR_LIT>' <EOL> package = __name__ <EOL> description = '<STR_LIT:description>' <EOL> self . pluginResults = [ FakeReactor ( install , name , package , description ) ] <EOL> self . assertRaises ( ImportError , reactors . installReactor , name ) <EOL> def test_reactorSelectionMixin ( self ) : <EOL> """<STR_LIT>""" <EOL> executed = [ ] <EOL> INSTALL_EVENT = '<STR_LIT>' <EOL> SUBCOMMAND_EVENT = '<STR_LIT>' <EOL> class ReactorSelectionOptions ( usage . Options , app . ReactorSelectionMixin ) : <EOL> def subCommands ( self ) : <EOL> executed . append ( SUBCOMMAND_EVENT ) <EOL> return [ ( '<STR_LIT>' , None , lambda : self , '<STR_LIT>' ) ] <EOL> subCommands = property ( subCommands ) <EOL> def install ( ) : <EOL> executed . append ( INSTALL_EVENT ) <EOL> self . pluginResults = [ <EOL> FakeReactor ( install , '<STR_LIT>' , __name__ , '<STR_LIT>' ) <EOL> ] <EOL> options = ReactorSelectionOptions ( ) <EOL> options . parseOptions ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( executed [ <NUM_LIT:0> ] , INSTALL_EVENT ) <EOL> self . assertEqual ( executed . count ( INSTALL_EVENT ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( options [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> def test_reactorSelectionMixinNonExistent ( self ) : <EOL> """<STR_LIT>""" <EOL> class ReactorSelectionOptions ( usage . Options , app . ReactorSelectionMixin ) : <EOL> pass <EOL> self . pluginResults = [ ] <EOL> options = ReactorSelectionOptions ( ) <EOL> options . messageOutput = StringIO ( ) <EOL> e = self . assertRaises ( usage . UsageError , options . parseOptions , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertIn ( "<STR_LIT>" , e . args [ <NUM_LIT:0> ] ) <EOL> self . assertIn ( "<STR_LIT>" , e . args [ <NUM_LIT:0> ] ) <EOL> def test_reactorSelectionMixinNotAvailable ( self ) : <EOL> """<STR_LIT>""" <EOL> class ReactorSelectionOptions ( usage . Options , app . ReactorSelectionMixin ) : <EOL> pass <EOL> message = "<STR_LIT>" <EOL> def install ( ) : <EOL> raise ImportError ( message ) <EOL> name = '<STR_LIT>' <EOL> package = __name__ <EOL> description = '<STR_LIT:description>' <EOL> self . pluginResults = [ FakeReactor ( install , name , package , description ) ] <EOL> options = ReactorSelectionOptions ( ) <EOL> options . messageOutput = StringIO ( ) <EOL> e = self . assertRaises ( usage . UsageError , options . parseOptions , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertIn ( message , e . args [ <NUM_LIT:0> ] ) <EOL> self . assertIn ( "<STR_LIT>" , e . args [ <NUM_LIT:0> ] ) </s>
