<s> from twisted . internet import defer , error , reactor <EOL> from twisted . protocols import loopback <EOL> from twisted . python import failure <EOL> from twisted . trial import unittest <EOL> from twirrdy import protocol <EOL> from twirrdy . unittests import DummyCacheProtocol , DummyCacheServer <EOL> class ProtocolTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . server = DummyCacheProtocol ( ) <EOL> self . client = protocol . RRDCacheProtocol ( ) <EOL> self . client . factory = self <EOL> loopback . loopbackAsync ( self . server , self . client ) <EOL> def tearDown ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , failure . Failure ) <EOL> self . assertIsInstance ( result . value , error . ConnectionDone ) <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( check ) <EOL> return d <EOL> def clientConnectionMade ( self , p ) : <EOL> pass <EOL> def testSimpleCommand ( self ) : <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( lambda x : self . assertEquals ( x , "<STR_LIT>" ) ) <EOL> return d <EOL> def testStatCommand ( self ) : <EOL> def check ( result ) : <EOL> lines = result . split ( '<STR_LIT:\n>' ) <EOL> self . assertEquals ( lines [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . assertEquals ( len ( lines ) , <NUM_LIT:10> ) <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( check ) <EOL> return d <EOL> def testManyCommands ( self ) : <EOL> dlist = [ ] <EOL> for x in xrange ( <NUM_LIT:10> ) : <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( lambda x : self . assertEquals ( x , "<STR_LIT>" ) ) <EOL> dlist . append ( d ) <EOL> return defer . DeferredList ( dlist ) <EOL> def testUnknownCommand ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , failure . Failure ) <EOL> self . assertIsInstance ( result . value , protocol . RRDCacheError ) <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( check ) <EOL> return d <EOL> def testManyUnknownCommands ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , failure . Failure ) <EOL> self . assertIsInstance ( result . value , protocol . RRDCacheError ) <EOL> dlist = [ ] <EOL> for x in xrange ( <NUM_LIT:10> ) : <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( check ) <EOL> dlist . append ( d ) <EOL> return defer . DeferredList ( dlist ) <EOL> class FactoryTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> sock = self . mktemp ( ) <EOL> serverfactory = DummyCacheServer ( ) <EOL> self . server = reactor . listenUNIX ( sock , serverfactory ) <EOL> deferred = defer . Deferred ( ) <EOL> self . client = protocol . RRDCacheClient ( deferred , True ) <EOL> reactor . connectUNIX ( sock , self . client ) <EOL> return deferred <EOL> def tearDown ( self ) : <EOL> def check ( result ) : <EOL> self . assertIsInstance ( result , failure . Failure ) <EOL> self . assertIsInstance ( result . value , error . ConnectionDone ) <EOL> self . client . stopTrying ( ) <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( check ) <EOL> d . addBoth ( lambda x : self . server . stopListening ( ) ) <EOL> return d <EOL> def testSimpleCommand ( self ) : <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( lambda x : self . assertEquals ( x , "<STR_LIT>" ) ) <EOL> return d <EOL> def testReconnect ( self ) : <EOL> final = defer . Deferred ( ) <EOL> def check_closed ( result ) : <EOL> self . assertIsInstance ( result , failure . Failure ) <EOL> self . assertIsInstance ( result . value , error . ConnectionDone ) <EOL> def setup_reconnect ( result ) : <EOL> self . assertEquals ( result , "<STR_LIT>" ) <EOL> self . client . _deferred = final <EOL> self . client . maxDelay = <NUM_LIT:0.1> <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addBoth ( check_closed ) <EOL> d . addCallback ( lambda x : final ) <EOL> d . addCallback ( test_reconnect ) <EOL> return d <EOL> def test_reconnect ( result ) : <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addCallback ( lambda x : self . assertEquals ( x , "<STR_LIT>" ) ) <EOL> return d <EOL> d = self . client . sendLine ( "<STR_LIT>" ) <EOL> d . addCallback ( setup_reconnect ) <EOL> return d </s>
