<s> """<STR_LIT>""" <EOL> from twisted . internet . error import ConnectionDone <EOL> from twisted . protocols . memcache import MemCacheProtocol , NoSuchCommand <EOL> from twisted . protocols . memcache import ClientError , ServerError <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . test . proto_helpers import StringTransportWithDisconnection <EOL> from twisted . internet . task import Clock <EOL> from twisted . internet . defer import Deferred , gatherResults , TimeoutError <EOL> from twisted . internet . defer import DeferredList <EOL> class CommandMixin : <EOL> """<STR_LIT>""" <EOL> def _test ( self , d , send , recv , result ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def test_get ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . get ( "<STR_LIT:foo>" ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , ( <NUM_LIT:0> , "<STR_LIT:bar>" ) ) <EOL> def test_emptyGet ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . get ( "<STR_LIT:foo>" ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , ( <NUM_LIT:0> , None ) ) <EOL> def test_getMultiple ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . getMultiple ( [ '<STR_LIT:foo>' , '<STR_LIT>' ] ) , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : ( <NUM_LIT:0> , '<STR_LIT>' ) , '<STR_LIT:foo>' : ( <NUM_LIT:0> , '<STR_LIT:bar>' ) } ) <EOL> def test_getMultipleWithEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . getMultiple ( [ '<STR_LIT:foo>' , '<STR_LIT>' ] ) , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : ( <NUM_LIT:1> , '<STR_LIT:bar>' ) , '<STR_LIT:foo>' : ( <NUM_LIT:0> , None ) } ) <EOL> def test_set ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . set ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , True ) <EOL> def test_add ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . add ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , True ) <EOL> def test_replace ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . replace ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , True ) <EOL> def test_errorAdd ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . add ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , False ) <EOL> def test_errorReplace ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . replace ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , False ) <EOL> def test_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . delete ( "<STR_LIT:bar>" ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , True ) <EOL> def test_errorDelete ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . delete ( "<STR_LIT:bar>" ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , False ) <EOL> def test_increment ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . increment ( "<STR_LIT:foo>" ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , <NUM_LIT:4> ) <EOL> def test_decrement ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( <EOL> self . proto . decrement ( "<STR_LIT:foo>" ) , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> def test_incrementVal ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . increment ( "<STR_LIT:foo>" , <NUM_LIT:8> ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , <NUM_LIT:4> ) <EOL> def test_decrementVal ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . decrement ( "<STR_LIT:foo>" , <NUM_LIT:3> ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , <NUM_LIT:5> ) <EOL> def test_stats ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . stats ( ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> { "<STR_LIT:foo>" : "<STR_LIT:bar>" , "<STR_LIT>" : "<STR_LIT>" } ) <EOL> def test_statsWithArgument ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . stats ( "<STR_LIT>" ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> { "<STR_LIT:foo>" : "<STR_LIT:bar>" , "<STR_LIT>" : "<STR_LIT>" } ) <EOL> def test_version ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . version ( ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> def test_flushAll ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . flushAll ( ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , True ) <EOL> class MemCacheTestCase ( CommandMixin , TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . proto = MemCacheProtocol ( ) <EOL> self . clock = Clock ( ) <EOL> self . proto . callLater = self . clock . callLater <EOL> self . transport = StringTransportWithDisconnection ( ) <EOL> self . transport . protocol = self . proto <EOL> self . proto . makeConnection ( self . transport ) <EOL> def _test ( self , d , send , recv , result ) : <EOL> """<STR_LIT>""" <EOL> def cb ( res ) : <EOL> self . assertEqual ( res , result ) <EOL> self . assertEqual ( self . transport . value ( ) , send ) <EOL> d . addCallback ( cb ) <EOL> self . proto . dataReceived ( recv ) <EOL> return d <EOL> def test_invalidGetResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> self . proto . get ( "<STR_LIT:foo>" ) <EOL> s = "<STR_LIT>" <EOL> self . assertRaises ( RuntimeError , <EOL> self . proto . dataReceived , <EOL> "<STR_LIT>" % ( len ( s ) , s ) ) <EOL> def test_invalidMultipleGetResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> self . proto . getMultiple ( [ "<STR_LIT:foo>" , "<STR_LIT:bar>" ] ) <EOL> s = "<STR_LIT>" <EOL> self . assertRaises ( RuntimeError , <EOL> self . proto . dataReceived , <EOL> "<STR_LIT>" % ( len ( s ) , s ) ) <EOL> def test_timeOut ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . proto . get ( "<STR_LIT:foo>" ) <EOL> d2 = self . proto . get ( "<STR_LIT:bar>" ) <EOL> d3 = Deferred ( ) <EOL> self . proto . connectionLost = d3 . callback <EOL> self . clock . advance ( self . proto . persistentTimeOut ) <EOL> self . assertFailure ( d1 , TimeoutError ) <EOL> self . assertFailure ( d2 , TimeoutError ) <EOL> def checkMessage ( error ) : <EOL> self . assertEqual ( str ( error ) , "<STR_LIT>" ) <EOL> d1 . addCallback ( checkMessage ) <EOL> return gatherResults ( [ d1 , d2 , d3 ] ) <EOL> def test_timeoutRemoved ( self ) : <EOL> """<STR_LIT>""" <EOL> d = self . proto . get ( "<STR_LIT:foo>" ) <EOL> self . clock . advance ( self . proto . persistentTimeOut - <NUM_LIT:1> ) <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> def check ( result ) : <EOL> self . assertEqual ( result , ( <NUM_LIT:0> , "<STR_LIT:bar>" ) ) <EOL> self . assertEqual ( len ( self . clock . calls ) , <NUM_LIT:0> ) <EOL> d . addCallback ( check ) <EOL> return d <EOL> def test_timeOutRaw ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . proto . get ( "<STR_LIT:foo>" ) <EOL> d2 = Deferred ( ) <EOL> self . proto . connectionLost = d2 . callback <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> self . clock . advance ( self . proto . persistentTimeOut ) <EOL> self . assertFailure ( d1 , TimeoutError ) <EOL> return gatherResults ( [ d1 , d2 ] ) <EOL> def test_timeOutStat ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . proto . stats ( ) <EOL> d2 = Deferred ( ) <EOL> self . proto . connectionLost = d2 . callback <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> self . clock . advance ( self . proto . persistentTimeOut ) <EOL> self . assertFailure ( d1 , TimeoutError ) <EOL> return gatherResults ( [ d1 , d2 ] ) <EOL> def test_timeoutPipelining ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . proto . get ( "<STR_LIT:foo>" ) <EOL> d2 = self . proto . get ( "<STR_LIT:bar>" ) <EOL> d3 = Deferred ( ) <EOL> self . proto . connectionLost = d3 . callback <EOL> self . clock . advance ( self . proto . persistentTimeOut - <NUM_LIT:1> ) <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> def check ( result ) : <EOL> self . assertEqual ( result , ( <NUM_LIT:0> , "<STR_LIT:bar>" ) ) <EOL> self . assertEqual ( len ( self . clock . calls ) , <NUM_LIT:1> ) <EOL> for i in range ( self . proto . persistentTimeOut ) : <EOL> self . clock . advance ( <NUM_LIT:1> ) <EOL> return self . assertFailure ( d2 , TimeoutError ) . addCallback ( checkTime ) <EOL> def checkTime ( ignored ) : <EOL> self . assertEqual ( <EOL> self . clock . seconds ( ) , <NUM_LIT:2> * self . proto . persistentTimeOut - <NUM_LIT:1> ) <EOL> d1 . addCallback ( check ) <EOL> return d1 <EOL> def test_timeoutNotReset ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . proto . get ( "<STR_LIT:foo>" ) <EOL> d3 = Deferred ( ) <EOL> self . proto . connectionLost = d3 . callback <EOL> self . clock . advance ( self . proto . persistentTimeOut - <NUM_LIT:1> ) <EOL> d2 = self . proto . get ( "<STR_LIT:bar>" ) <EOL> self . clock . advance ( <NUM_LIT:1> ) <EOL> self . assertFailure ( d1 , TimeoutError ) <EOL> self . assertFailure ( d2 , TimeoutError ) <EOL> return gatherResults ( [ d1 , d2 , d3 ] ) <EOL> def test_timeoutCleanDeferreds ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . proto . get ( "<STR_LIT:foo>" ) <EOL> self . clock . advance ( self . proto . persistentTimeOut ) <EOL> self . assertFailure ( d1 , TimeoutError ) <EOL> d2 = self . proto . get ( "<STR_LIT:bar>" ) <EOL> self . assertFailure ( d2 , RuntimeError ) <EOL> return gatherResults ( [ d1 , d2 ] ) <EOL> def test_connectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . proto . get ( "<STR_LIT:foo>" ) <EOL> d2 = self . proto . get ( "<STR_LIT:bar>" ) <EOL> self . transport . loseConnection ( ) <EOL> done = DeferredList ( [ d1 , d2 ] , consumeErrors = True ) <EOL> def checkFailures ( results ) : <EOL> for success , result in results : <EOL> self . assertFalse ( success ) <EOL> result . trap ( ConnectionDone ) <EOL> return done . addCallback ( checkFailures ) <EOL> def test_tooLongKey ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . assertFailure ( self . proto . set ( "<STR_LIT:a>" * <NUM_LIT> , "<STR_LIT:bar>" ) , ClientError ) <EOL> d2 = self . assertFailure ( self . proto . increment ( "<STR_LIT:a>" * <NUM_LIT> ) , ClientError ) <EOL> d3 = self . assertFailure ( self . proto . get ( "<STR_LIT:a>" * <NUM_LIT> ) , ClientError ) <EOL> d4 = self . assertFailure ( <EOL> self . proto . append ( "<STR_LIT:a>" * <NUM_LIT> , "<STR_LIT:bar>" ) , ClientError ) <EOL> d5 = self . assertFailure ( <EOL> self . proto . prepend ( "<STR_LIT:a>" * <NUM_LIT> , "<STR_LIT:bar>" ) , ClientError ) <EOL> d6 = self . assertFailure ( <EOL> self . proto . getMultiple ( [ "<STR_LIT:foo>" , "<STR_LIT:a>" * <NUM_LIT> ] ) , ClientError ) <EOL> return gatherResults ( [ d1 , d2 , d3 , d4 , d5 , d6 ] ) <EOL> def test_invalidCommand ( self ) : <EOL> """<STR_LIT>""" <EOL> d = self . proto . _set ( "<STR_LIT>" , "<STR_LIT:foo>" , "<STR_LIT:bar>" , <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> self . assertEqual ( self . transport . value ( ) , "<STR_LIT>" ) <EOL> self . assertFailure ( d , NoSuchCommand ) <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> return d <EOL> def test_clientError ( self ) : <EOL> """<STR_LIT>""" <EOL> a = "<STR_LIT>" <EOL> d = self . proto . set ( "<STR_LIT:foo>" , a ) <EOL> self . assertEqual ( self . transport . value ( ) , <EOL> "<STR_LIT>" ) <EOL> self . assertFailure ( d , ClientError ) <EOL> def check ( err ) : <EOL> self . assertEqual ( str ( err ) , "<STR_LIT>" ) <EOL> d . addCallback ( check ) <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> return d <EOL> def test_serverError ( self ) : <EOL> """<STR_LIT>""" <EOL> a = "<STR_LIT>" <EOL> d = self . proto . set ( "<STR_LIT:foo>" , a ) <EOL> self . assertEqual ( self . transport . value ( ) , <EOL> "<STR_LIT>" ) <EOL> self . assertFailure ( d , ServerError ) <EOL> def check ( err ) : <EOL> self . assertEqual ( str ( err ) , "<STR_LIT>" ) <EOL> d . addCallback ( check ) <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> return d <EOL> def test_unicodeKey ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . assertFailure ( self . proto . set ( u"<STR_LIT:foo>" , "<STR_LIT:bar>" ) , ClientError ) <EOL> d2 = self . assertFailure ( self . proto . increment ( u"<STR_LIT>" ) , ClientError ) <EOL> d3 = self . assertFailure ( self . proto . get ( <NUM_LIT:1> ) , ClientError ) <EOL> d4 = self . assertFailure ( self . proto . delete ( u"<STR_LIT:bar>" ) , ClientError ) <EOL> d5 = self . assertFailure ( self . proto . append ( u"<STR_LIT:foo>" , "<STR_LIT:bar>" ) , ClientError ) <EOL> d6 = self . assertFailure ( self . proto . prepend ( u"<STR_LIT:foo>" , "<STR_LIT:bar>" ) , ClientError ) <EOL> d7 = self . assertFailure ( <EOL> self . proto . getMultiple ( [ "<STR_LIT>" , <NUM_LIT:1> ] ) , ClientError ) <EOL> return gatherResults ( [ d1 , d2 , d3 , d4 , d5 , d6 , d7 ] ) <EOL> def test_unicodeValue ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . assertFailure ( self . proto . set ( "<STR_LIT:foo>" , u"<STR_LIT:bar>" ) , ClientError ) <EOL> def test_pipelining ( self ) : <EOL> """<STR_LIT>""" <EOL> d1 = self . proto . get ( "<STR_LIT:foo>" ) <EOL> d1 . addCallback ( self . assertEqual , ( <NUM_LIT:0> , "<STR_LIT:bar>" ) ) <EOL> d2 = self . proto . set ( "<STR_LIT:bar>" , "<STR_LIT>" ) <EOL> d2 . addCallback ( self . assertEqual , True ) <EOL> d3 = self . proto . get ( "<STR_LIT>" ) <EOL> d3 . addCallback ( self . assertEqual , ( <NUM_LIT:0> , "<STR_LIT>" ) ) <EOL> self . assertEqual ( self . transport . value ( ) , <EOL> "<STR_LIT>" ) <EOL> self . proto . dataReceived ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return gatherResults ( [ d1 , d2 , d3 ] ) <EOL> def test_getInChunks ( self ) : <EOL> """<STR_LIT>""" <EOL> d = self . proto . get ( "<STR_LIT:foo>" ) <EOL> d . addCallback ( self . assertEqual , ( <NUM_LIT:0> , "<STR_LIT>" ) ) <EOL> self . assertEqual ( self . transport . value ( ) , "<STR_LIT>" ) <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> self . proto . dataReceived ( "<STR_LIT>" ) <EOL> self . proto . dataReceived ( "<STR_LIT:\r\n>" ) <EOL> return d <EOL> def test_append ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . append ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , True ) <EOL> def test_prepend ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . prepend ( "<STR_LIT:foo>" , "<STR_LIT:bar>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , True ) <EOL> def test_gets ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . get ( "<STR_LIT:foo>" , True ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , ( <NUM_LIT:0> , "<STR_LIT>" , "<STR_LIT:bar>" ) ) <EOL> def test_emptyGets ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . get ( "<STR_LIT:foo>" , True ) , "<STR_LIT>" , <EOL> "<STR_LIT>" , ( <NUM_LIT:0> , "<STR_LIT>" , None ) ) <EOL> def test_getsMultiple ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . getMultiple ( [ "<STR_LIT:foo>" , "<STR_LIT:bar>" ] , True ) , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:bar>' : ( <NUM_LIT:0> , '<STR_LIT>' , '<STR_LIT>' ) , '<STR_LIT:foo>' : ( <NUM_LIT:0> , '<STR_LIT>' , '<STR_LIT>' ) } ) <EOL> def test_getsMultipleWithEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . getMultiple ( [ "<STR_LIT:foo>" , "<STR_LIT:bar>" ] , True ) , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:bar>' : ( <NUM_LIT:0> , '<STR_LIT>' , None ) , '<STR_LIT:foo>' : ( <NUM_LIT:0> , '<STR_LIT>' , '<STR_LIT>' ) } ) <EOL> def test_checkAndSet ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . checkAndSet ( "<STR_LIT:foo>" , "<STR_LIT:bar>" , cas = "<STR_LIT>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , True ) <EOL> def test_casUnknowKey ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _test ( self . proto . checkAndSet ( "<STR_LIT:foo>" , "<STR_LIT:bar>" , cas = "<STR_LIT>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , False ) <EOL> class CommandFailureTests ( CommandMixin , TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . proto = MemCacheProtocol ( ) <EOL> self . clock = Clock ( ) <EOL> self . proto . callLater = self . clock . callLater <EOL> self . transport = StringTransportWithDisconnection ( ) <EOL> self . transport . protocol = self . proto <EOL> self . proto . makeConnection ( self . transport ) <EOL> self . transport . loseConnection ( ) <EOL> def _test ( self , d , send , recv , result ) : <EOL> """<STR_LIT>""" <EOL> return self . assertFailure ( d , RuntimeError ) </s>
