<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN"
"http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head><meta http-equiv="content-type" content="text/html; charset=utf-8" /><style type="text/css"><!--
#msg dl { border: 1px #006 solid; background: #369; padding: 6px; color: #fff; }
#msg dt { float: left; width: 6em; font-weight: bold; }
#msg dt:after { content:':';}
#msg dl, #msg dt, #msg ul, #msg li, #header, #footer { font-family: verdana,arial,helvetica,sans-serif; font-size: 10pt; }
#msg dl a { font-weight: bold}
#msg dl a:link { color:#fc3; }
#msg dl a:active { color:#ff0; }
#msg dl a:visited { color:#cc6; }
h3 { font-family: verdana,arial,helvetica,sans-serif; font-size: 10pt; font-weight: bold; }
#msg pre, #msg p { overflow: auto; background: #ffc; border: 1px #fc0 solid; padding: 6px; }
#msg ul { overflow: auto; }
#header, #footer { color: #fff; background: #636; border: 1px #300 solid; padding: 6px; }
#patch { width: 100%; }
#patch h4 {font-family: verdana,arial,helvetica,sans-serif;font-size:10pt;padding:8px;background:#369;color:#fff;margin:0;}
#patch .propset h4, #patch .binary h4 {margin:0;}
#patch pre {padding:0;line-height:1.2em;margin:0;}
#patch .diff {width:100%;background:#eee;padding: 0 0 10px 0;overflow:auto;}
#patch .propset .diff, #patch .binary .diff {padding:10px 0;}
#patch span {display:block;padding:0 10px;}
#patch .modfile, #patch .addfile, #patch .delfile, #patch .propset, #patch .binary, #patch .copfile {border:1px solid #ccc;margin:10px 0;}
#patch ins {background:#dfd;text-decoration:none;display:block;padding:0 10px;}
#patch del {background:#fdd;text-decoration:none;display:block;padding:0 10px;}
#patch .lines, .info {color:#888;background:#fff;}
--></style>
<title>[2439] CalendarServer/branches/unified-cache/twistedcaldav</title>
</head>
<body>
<div id="msg">
<dl>
<dt>Revision</dt> <dd><a href="http://trac.macosforge.org/projects/calendarserver/changeset/2439">2439</a></dd>
<dt>Author</dt> <dd>dreid@apple.com</dd>
<dt>Date</dt> <dd>2008-05-21 16:11:19 -0700 (Wed, 21 May 2008)</dd>
</dl>
<h3>Log Message</h3>
<pre>Add twisted memcached protocol implementation</pre>
<h3>Added Paths</h3>
<ul>
<li><a href="#CalendarServerbranchesunifiedcachetwistedcaldavmemcachepy">CalendarServer/branches/unified-cache/twistedcaldav/memcache.py</a></li>
<li><a href="#CalendarServerbranchesunifiedcachetwistedcaldavtesttest_memcachepy">CalendarServer/branches/unified-cache/twistedcaldav/test/test_memcache.py</a></li>
</ul>
</div>
<div id="patch">
<h3>Diff</h3>
<a id="CalendarServerbranchesunifiedcachetwistedcaldavmemcachepy"></a>
<div class="addfile"><h4>Added: CalendarServer/branches/unified-cache/twistedcaldav/memcache.py (0 => 2439)</h4>
<pre class="diff"><span>
<span class="info">--- CalendarServer/branches/unified-cache/twistedcaldav/memcache.py (rev 0)
+++ CalendarServer/branches/unified-cache/twistedcaldav/memcache.py 2008-05-21 23:11:19 UTC (rev 2439)
</span><span class="lines">@@ -0,0 +1,657 @@
</span><ins>+# -*- test-case-name: twisted.test.test_memcache -*-
+# Copyright (c) 2007 Twisted Matrix Laboratories.
+# See LICENSE for details.
+
+"""
+Memcache client protocol. Memcached is a caching server, storing data in the
+form of pairs key/value, and memcache is the protocol to talk with it.
+
+To connect to a server, create a factory for L{MemCacheProtocol}::
+
+ from twisted.internet import reactor, protocol
+ from twisted.protocols.memcache import MemCacheProtocol, DEFAULT_PORT
+ d = protocol.ClientCreator(reactor, MemCacheProtocol
+ ).connectTCP("localhost", DEFAULT_PORT)
+ def doSomething(proto):
+ # Here you call the memcache operations
+ return proto.set("mykey", "a lot of data")
+ d.addCallback(doSomething)
+ reactor.run()
+
+All the operations of the memcache protocol are present, but
+L{MemCacheProtocol.set} and L{MemCacheProtocol.get} are the more important.
+
+See U{http://code.sixapart.com/svn/memcached/trunk/server/doc/protocol.txt} for
+more information about the protocol.
+"""
+
+try:
+ from collections import deque
+except ImportError:
+ class deque(list):
+ def popleft(self):
+ return self.pop(0)
+
+
+from twisted.protocols.basic import LineReceiver
+from twisted.protocols.policies import TimeoutMixin
+from twisted.internet.defer import Deferred, fail, TimeoutError
+from twisted.python import log
+
+
+
+DEFAULT_PORT = 11211
+
+
+
+class NoSuchCommand(Exception):
+ """
+ Exception raised when a non existent command is called.
+ """
+
+
+
+class ClientError(Exception):
+ """
+ Error caused by an invalid client call.
+ """
+
+
+
+class ServerError(Exception):
+ """
+ Problem happening on the server.
+ """
+
+
+
+class Command(object):
+ """
+ Wrap a client action into an object, that holds the values used in the
+ protocol.
+
+ @ivar _deferred: the L{Deferred} object that will be fired when the result
+ arrives.
+ @type _deferred: L{Deferred}
+
+ @ivar command: name of the command sent to the server.
+ @type command: C{str}
+ """
+
+ def __init__(self, command, **kwargs):
+ """
+ Create a command.
+
+ @param command: the name of the command.
+ @type command: C{str}
+
+ @param kwargs: this values will be stored as attributes of the object
+ for future use
+ """
+ self.command = command
+ self._deferred = Deferred()
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+
+
+ def success(self, value):
+ """
+ Shortcut method to fire the underlying deferred.
+ """
+ self._deferred.callback(value)
+
+
+ def fail(self, error):
+ """
+ Make the underlying deferred fails.
+ """
+ self._deferred.errback(error)
+
+
+
+class MemCacheProtocol(LineReceiver, TimeoutMixin):
+ """
+ MemCache protocol: connect to a memcached server to store/retrieve values.
+
+ @ivar persistentTimeOut: the timeout period used to wait for a response.
+ @type persistentTimeOut: C{int}
+
+ @ivar _current: current list of requests waiting for an answer from the
+ server.
+ @type _current: C{deque} of L{Command}
+
+ @ivar _lenExpected: amount of data expected in raw mode, when reading for
+ a value.
+ @type _lenExpected: C{int}
+
+ @ivar _getBuffer: current buffer of data, used to store temporary data
+ when reading in raw mode.
+ @type _getBuffer: C{list}
+
+ @ivar _bufferLength: the total amount of bytes in C{_getBuffer}.
+ @type _bufferLength: C{int}
+ """
+ MAX_KEY_LENGTH = 250
+
+ def __init__(self, timeOut=60):
+ """
+ Create the protocol.
+
+ @param timeOut: the timeout to wait before detecting that the
+ connection is dead and close it. It's expressed in seconds.
+ @type timeOut: C{int}
+ """
+ self._current = deque()
+ self._lenExpected = None
+ self._getBuffer = None
+ self._bufferLength = None
+ self.persistentTimeOut = self.timeOut = timeOut
+
+
+ def timeoutConnection(self):
+ """
+ Close the connection in case of timeout.
+ """
+ for cmd in self._current:
+ cmd.fail(TimeoutError("Connection timeout"))
+ self.transport.loseConnection()
+
+
+ def sendLine(self, line):
+ """
+ Override sendLine to add a timeout to response.
+ """
+ if not self._current:
+ self.setTimeout(self.persistentTimeOut)
+ LineReceiver.sendLine(self, line)
+
+
+ def rawDataReceived(self, data):
+ """
+ Collect data for a get.
+ """
+ self.resetTimeout()
+ self._getBuffer.append(data)
+ self._bufferLength += len(data)
+ if self._bufferLength >= self._lenExpected + 2:
+ data = "".join(self._getBuffer)
+ buf = data[:self._lenExpected]
+ rem = data[self._lenExpected + 2:]
+ val = buf
+ self._lenExpected = None
+ self._getBuffer = None
+ self._bufferLength = None
+ cmd = self._current[0]
+ cmd.value = val
+ self.setLineMode(rem)
+
+
+ def cmd_STORED(self):
+ """
+ Manage a success response to a set operation.
+ """
+ self._current.popleft().success(True)
+
+
+ def cmd_NOT_STORED(self):
+ """
+ Manage a specific 'not stored' response to a set operation: this is not
+ an error, but some condition wasn't met.
+ """
+ self._current.popleft().success(False)
+
+
+ def cmd_END(self):
+ """
+ This the end token to a get or a stat operation.
+ """
+ cmd = self._current.popleft()
+ if cmd.command == "get":
+ cmd.success((cmd.flags, cmd.value))
+ elif cmd.command == "gets":
+ cmd.success((cmd.flags, cmd.cas, cmd.value))
+ elif cmd.command == "stats":
+ cmd.success(cmd.values)
+
+
+ def cmd_NOT_FOUND(self):
+ """
+ Manage error response for incr/decr/delete.
+ """
+ self._current.popleft().success(False)
+
+
+ def cmd_VALUE(self, line):
+ """
+ Prepare the reading a value after a get.
+ """
+ cmd = self._current[0]
+ if cmd.command == "get":
+ key, flags, length = line.split()
+ cas = ""
+ else:
+ key, flags, length, cas = line.split()
+ self._lenExpected = int(length)
+ self._getBuffer = []
+ self._bufferLength = 0
+ if cmd.key != key:
+ raise RuntimeError("Unexpected commands answer.")
+ cmd.flags = int(flags)
+ cmd.length = self._lenExpected
+ cmd.cas = cas
+ self.setRawMode()
+
+
+ def cmd_STAT(self, line):
+ """
+ Reception of one stat line.
+ """
+ cmd = self._current[0]
+ key, val = line.split(" ", 1)
+ cmd.values[key] = val
+
+
+ def cmd_VERSION(self, versionData):
+ """
+ Read version token.
+ """
+ self._current.popleft().success(versionData)
+
+
+ def cmd_ERROR(self):
+ """
+ An non-existent command has been sent.
+ """
+ log.err("Non-existent command sent.")
+ cmd = self._current.popleft()
+ cmd.fail(NoSuchCommand())
+
+
+ def cmd_CLIENT_ERROR(self, errText):
+ """
+ An invalid input as been sent.
+ """
+ log.err("Invalid input: %s" % (errText,))
+ cmd = self._current.popleft()
+ cmd.fail(ClientError(errText))
+
+
+ def cmd_SERVER_ERROR(self, errText):
+ """
+ An error has happened server-side.
+ """
+ log.err("Server error: %s" % (errText,))
+ cmd = self._current.popleft()
+ cmd.fail(ServerError(errText))
+
+
+ def cmd_DELETED(self):
+ """
+ A delete command has completed successfully.
+ """
+ self._current.popleft().success(True)
+
+
+ def cmd_OK(self):
+ """
+ The last command has been completed.
+ """
+ self._current.popleft().success(True)
+
+
+ def cmd_EXISTS(self):
+ """
+ A C{checkAndSet} update has failed.
+ """
+ self._current.popleft().success(False)
+
+
+ def lineReceived(self, line):
+ """
+ Receive line commands from the server.
+ """
+ self.resetTimeout()
+ token = line.split(" ", 1)[0]
+ # First manage standard commands without space
+ cmd = getattr(self, "cmd_%s" % (token,), None)
+ if cmd is not None:
+ args = line.split(" ", 1)[1:]
+ if args:
+ cmd(args[0])
+ else:
+ cmd()
+ else:
+ # Then manage commands with space in it
+ line = line.replace(" ", "_")
+ cmd = getattr(self, "cmd_%s" % (line,), None)
+ if cmd is not None:
+ cmd()
+ else:
+ # Increment/Decrement response
+ cmd = self._current.popleft()
+ val = int(line)
+ cmd.success(val)
+ if not self._current:
+ # No pending request, remove timeout
+ self.setTimeout(None)
+
+
+ def increment(self, key, val=1):
+ """
+ Increment the value of C{key} by given value (default to 1).
+ C{key} must be consistent with an int. Return the new value.
+
+ @param key: the key to modify.
+ @type key: C{str}
+
+ @param val: the value to increment.
+ @type val: C{int}
+
+ @return: a deferred with will be called back with the new value
+ associated with the key (after the increment).
+ @rtype: L{Deferred}
+ """
+ return self._incrdecr("incr", key, val)
+
+
+ def decrement(self, key, val=1):
+ """
+ Decrement the value of C{key} by given value (default to 1).
+ C{key} must be consistent with an int. Return the new value, coerced to
+ 0 if negative.
+
+ @param key: the key to modify.
+ @type key: C{str}
+
+ @param val: the value to decrement.
+ @type val: C{int}
+
+ @return: a deferred with will be called back with the new value
+ associated with the key (after the decrement).
+ @rtype: L{Deferred}
+ """
+ return self._incrdecr("decr", key, val)
+
+
+ def _incrdecr(self, cmd, key, val):
+ """
+ Internal wrapper for incr/decr.
+ """
+ if not isinstance(key, str):
+ return fail(ClientError(
+ "Invalid type for key: %s, expecting a string" % (type(key),)))
+ if len(key) > self.MAX_KEY_LENGTH:
+ return fail(ClientError("Key too long"))
+ fullcmd = "%s %s %d" % (cmd, key, int(val))
+ self.sendLine(fullcmd)
+ cmdObj = Command(cmd, key=key)
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def replace(self, key, val, flags=0, expireTime=0):
+ """
+ Replace the given C{key}. It must already exist in the server.
+
+ @param key: the key to replace.
+ @type key: C{str}
+
+ @param val: the new value associated with the key.
+ @type val: C{str}
+
+ @param flags: the flags to store with the key.
+ @type flags: C{int}
+
+ @param expireTime: if different from 0, the relative time in seconds
+ when the key will be deleted from the store.
+ @type expireTime: C{int}
+
+ @return: a deferred that will fire with C{True} if the operation has
+ succeeded, and C{False} with the key didn't previously exist.
+ @rtype: L{Deferred}
+ """
+ return self._set("replace", key, val, flags, expireTime, "")
+
+
+ def add(self, key, val, flags=0, expireTime=0):
+ """
+ Add the given C{key}. It must not exist in the server.
+
+ @param key: the key to add.
+ @type key: C{str}
+
+ @param val: the value associated with the key.
+ @type val: C{str}
+
+ @param flags: the flags to store with the key.
+ @type flags: C{int}
+
+ @param expireTime: if different from 0, the relative time in seconds
+ when the key will be deleted from the store.
+ @type expireTime: C{int}
+
+ @return: a deferred that will fire with C{True} if the operation has
+ succeeded, and C{False} with the key already exists.
+ @rtype: L{Deferred}
+ """
+ return self._set("add", key, val, flags, expireTime, "")
+
+
+ def set(self, key, val, flags=0, expireTime=0):
+ """
+ Set the given C{key}.
+
+ @param key: the key to set.
+ @type key: C{str}
+
+ @param val: the value associated with the key.
+ @type val: C{str}
+
+ @param flags: the flags to store with the key.
+ @type flags: C{int}
+
+ @param expireTime: if different from 0, the relative time in seconds
+ when the key will be deleted from the store.
+ @type expireTime: C{int}
+
+ @return: a deferred that will fire with C{True} if the operation has
+ succeeded.
+ @rtype: L{Deferred}
+ """
+ return self._set("set", key, val, flags, expireTime, "")
+
+
+ def checkAndSet(self, key, val, cas, flags=0, expireTime=0):
+ """
+ Change the content of C{key} only if the C{cas} value matches the
+ current one associated with the key. Use this to store a value which
+ hasn't been modified since last time you fetched it.
+
+ @param key: The key to set.
+ @type key: C{str}
+
+ @param val: The value associated with the key.
+ @type val: C{str}
+
+ @param cas: Unique 64-bit value returned by previous call of C{get}.
+ @type cas: C{str}
+
+ @param flags: The flags to store with the key.
+ @type flags: C{int}
+
+ @param expireTime: If different from 0, the relative time in seconds
+ when the key will be deleted from the store.
+ @type expireTime: C{int}
+
+ @return: A deferred that will fire with C{True} if the operation has
+ succeeded, C{False} otherwise.
+ @rtype: L{Deferred}
+ """
+ return self._set("cas", key, val, flags, expireTime, cas)
+
+
+ def _set(self, cmd, key, val, flags, expireTime, cas):
+ """
+ Internal wrapper for setting values.
+ """
+ if not isinstance(key, str):
+ return fail(ClientError(
+ "Invalid type for key: %s, expecting a string" % (type(key),)))
+ if len(key) > self.MAX_KEY_LENGTH:
+ return fail(ClientError("Key too long"))
+ if not isinstance(val, str):
+ return fail(ClientError(
+ "Invalid type for value: %s, expecting a string" %
+ (type(val),)))
+ if cas:
+ cas = " " + cas
+ length = len(val)
+ fullcmd = "%s %s %d %d %d%s" % (
+ cmd, key, flags, expireTime, length, cas)
+ self.sendLine(fullcmd)
+ self.sendLine(val)
+ cmdObj = Command(cmd, key=key, flags=flags, length=length)
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def append(self, key, val):
+ """
+ Append given data to the value of an existing key.
+
+ @param key: The key to modify.
+ @type key: C{str}
+
+ @param val: The value to append to the current value associated with
+ the key.
+ @type val: C{str}
+
+ @return: A deferred that will fire with C{True} if the operation has
+ succeeded, C{False} otherwise.
+ @rtype: L{Deferred}
+ """
+ # Even if flags and expTime values are ignored, we have to pass them
+ return self._set("append", key, val, 0, 0, "")
+
+
+ def prepend(self, key, val):
+ """
+ Prepend given data to the value of an existing key.
+
+ @param key: The key to modify.
+ @type key: C{str}
+
+ @param val: The value to prepend to the current value associated with
+ the key.
+ @type val: C{str}
+
+ @return: A deferred that will fire with C{True} if the operation has
+ succeeded, C{False} otherwise.
+ @rtype: L{Deferred}
+ """
+ # Even if flags and expTime values are ignored, we have to pass them
+ return self._set("prepend", key, val, 0, 0, "")
+
+
+ def get(self, key, withIdentifier=False):
+ """
+ Get the given C{key}. It doesn't support multiple keys. If
+ C{withIdentifier} is set to C{True}, the command issued is a C{gets},
+ that will return the current identifier associated with the value. This
+ identifier has to be used when issuing C{checkAndSet} update later,
+ using the corresponding method.
+
+ @param key: The key to retrieve.
+ @type key: C{str}
+
+ @param withIdentifier: If set to C{True}, retrieve the current
+ identifier along with the value and the flags.
+ @type withIdentifier: C{bool}
+
+ @return: A deferred that will fire with the tuple (flags, value) if
+ C{withIdentifier} is C{False}, or (flags, cas identifier, value)
+ if C{True}.
+ @rtype: L{Deferred}
+ """
+ if not isinstance(key, str):
+ return fail(ClientError(
+ "Invalid type for key: %s, expecting a string" % (type(key),)))
+ if len(key) > self.MAX_KEY_LENGTH:
+ return fail(ClientError("Key too long"))
+ if withIdentifier:
+ cmd = "gets"
+ else:
+ cmd = "get"
+ fullcmd = "%s %s" % (cmd, key)
+ self.sendLine(fullcmd)
+ cmdObj = Command(cmd, key=key, value=None, flags=0, cas="")
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def stats(self):
+ """
+ Get some stats from the server. It will be available as a dict.
+
+ @return: a deferred that will fire with a C{dict} of the available
+ statistics.
+ @rtype: L{Deferred}
+ """
+ self.sendLine("stats")
+ cmdObj = Command("stats", values={})
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def version(self):
+ """
+ Get the version of the server.
+
+ @return: a deferred that will fire with the string value of the
+ version.
+ @rtype: L{Deferred}
+ """
+ self.sendLine("version")
+ cmdObj = Command("version")
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def delete(self, key):
+ """
+ Delete an existing C{key}.
+
+ @param key: the key to delete.
+ @type key: C{str}
+
+ @return: a deferred that will be called back with C{True} if the key
+ was successfully deleted, or C{False} if not.
+ @rtype: L{Deferred}
+ """
+ if not isinstance(key, str):
+ return fail(ClientError(
+ "Invalid type for key: %s, expecting a string" % (type(key),)))
+ self.sendLine("delete %s" % key)
+ cmdObj = Command("delete", key=key)
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+ def flushAll(self):
+ """
+ Flush all cached values.
+
+ @return: a deferred that will be called back with C{True} when the
+ operation has succeeded.
+ @rtype: L{Deferred}
+ """
+ self.sendLine("flush_all")
+ cmdObj = Command("flush_all")
+ self._current.append(cmdObj)
+ return cmdObj._deferred
+
+
+
+__all__ = ["MemCacheProtocol", "DEFAULT_PORT", "NoSuchCommand", "ClientError",
+ "ServerError"]
+
</ins></span></pre></div>
<a id="CalendarServerbranchesunifiedcachetwistedcaldavtesttest_memcachepy"></a>
<div class="addfile"><h4>Added: CalendarServer/branches/unified-cache/twistedcaldav/test/test_memcache.py (0 => 2439)</h4>
<pre class="diff"><span>
<span class="info">--- CalendarServer/branches/unified-cache/twistedcaldav/test/test_memcache.py (rev 0)
+++ CalendarServer/branches/unified-cache/twistedcaldav/test/test_memcache.py 2008-05-21 23:11:19 UTC (rev 2439)
</span><span class="lines">@@ -0,0 +1,510 @@
</span><ins>+# Copyright (c) 2007 Twisted Matrix Laboratories.
+# See LICENSE for details.
+
+"""
+Test the memcache client protocol.
+"""
+
+from twisted.protocols.memcache import MemCacheProtocol, NoSuchCommand
+from twisted.protocols.memcache import ClientError, ServerError
+
+from twisted.trial.unittest import TestCase
+from twisted.test.proto_helpers import StringTransportWithDisconnection
+from twisted.internet.task import Clock
+from twisted.internet.defer import Deferred, gatherResults, TimeoutError
+
+
+
+class MemCacheTestCase(TestCase):
+ """
+ Test client protocol class L{MemCacheProtocol}.
+ """
+
+ def setUp(self):
+ """
+ Create a memcache client, connect it to a string protocol, and make it
+ use a deterministic clock.
+ """
+ self.proto = MemCacheProtocol()
+ self.clock = Clock()
+ self.proto.callLater = self.clock.callLater
+ self.transport = StringTransportWithDisconnection()
+ self.transport.protocol = self.proto
+ self.proto.makeConnection(self.transport)
+
+
+ def _test(self, d, send, recv, result):
+ """
+ Shortcut method for classic tests.
+
+ @param d: the resulting deferred from the memcache command.
+ @type d: C{Deferred}
+
+ @param send: the expected data to be sent.
+ @type send: C{str}
+
+ @param recv: the data to simulate as reception.
+ @type recv: C{str}
+
+ @param result: the expected result.
+ @type result: C{any}
+ """
+ def cb(res):
+ self.assertEquals(res, result)
+ self.assertEquals(self.transport.value(), send)
+ d.addCallback(cb)
+ self.proto.dataReceived(recv)
+ return d
+
+
+ def test_get(self):
+ """
+ L{MemCacheProtocol.get} should return a L{Deferred} which is
+ called back with the value and the flag associated with the given key
+ if the server returns a successful result.
+ """
+ return self._test(self.proto.get("foo"), "get foo\r\n",
+ "VALUE foo 0 3\r\nbar\r\nEND\r\n", (0, "bar"))
+
+
+ def test_emptyGet(self):
+ """
+ Test getting a non-available key: it should succeed but return C{None}
+ as value and C{0} as flag.
+ """
+ return self._test(self.proto.get("foo"), "get foo\r\n",
+ "END\r\n", (0, None))
+
+
+ def test_set(self):
+ """
+ L{MemCacheProtocol.set} should return a L{Deferred} which is
+ called back with C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.set("foo", "bar"),
+ "set foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_add(self):
+ """
+ L{MemCacheProtocol.add} should return a L{Deferred} which is
+ called back with C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.add("foo", "bar"),
+ "add foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_replace(self):
+ """
+ L{MemCacheProtocol.replace} should return a L{Deferred} which
+ is called back with C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.replace("foo", "bar"),
+ "replace foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_errorAdd(self):
+ """
+ Test an erroneous add: if a L{MemCacheProtocol.add} is called but the
+ key already exists on the server, it returns a B{NOT STORED} answer,
+ which should callback the resulting L{Deferred} with C{False}.
+ """
+ return self._test(self.proto.add("foo", "bar"),
+ "add foo 0 0 3\r\nbar\r\n", "NOT STORED\r\n", False)
+
+
+ def test_errorReplace(self):
+ """
+ Test an erroneous replace: if a L{MemCacheProtocol.replace} is called
+ but the key doesn't exist on the server, it returns a B{NOT STORED}
+ answer, which should callback the resulting L{Deferred} with C{False}.
+ """
+ return self._test(self.proto.replace("foo", "bar"),
+ "replace foo 0 0 3\r\nbar\r\n", "NOT STORED\r\n", False)
+
+
+ def test_delete(self):
+ """
+ L{MemCacheProtocol.delete} should return a L{Deferred} which is
+ called back with C{True} when the server notifies a success.
+ """
+ return self._test(self.proto.delete("bar"), "delete bar\r\n",
+ "DELETED\r\n", True)
+
+
+ def test_errorDelete(self):
+ """
+ Test a error during a delete: if key doesn't exist on the server, it
+ returns a B{NOT FOUND} answer which should callback the resulting
+ L{Deferred} with C{False}.
+ """
+ return self._test(self.proto.delete("bar"), "delete bar\r\n",
+ "NOT FOUND\r\n", False)
+
+
+ def test_increment(self):
+ """
+ Test incrementing a variable: L{MemCacheProtocol.increment} should
+ return a L{Deferred} which is called back with the incremented value of
+ the given key.
+ """
+ return self._test(self.proto.increment("foo"), "incr foo 1\r\n",
+ "4\r\n", 4)
+
+
+ def test_decrement(self):
+ """
+ Test decrementing a variable: L{MemCacheProtocol.decrement} should
+ return a L{Deferred} which is called back with the decremented value of
+ the given key.
+ """
+ return self._test(
+ self.proto.decrement("foo"), "decr foo 1\r\n", "5\r\n", 5)
+
+
+ def test_incrementVal(self):
+ """
+ L{MemCacheProtocol.increment} takes an optional argument C{value} which
+ should replace the default value of 1 when specified.
+ """
+ return self._test(self.proto.increment("foo", 8), "incr foo 8\r\n",
+ "4\r\n", 4)
+
+
+ def test_decrementVal(self):
+ """
+ L{MemCacheProtocol.decrement} takes an optional argument C{value} which
+ should replace the default value of 1 when specified.
+ """
+ return self._test(self.proto.decrement("foo", 3), "decr foo 3\r\n",
+ "5\r\n", 5)
+
+
+ def test_stats(self):
+ """
+ Test retrieving server statistics via the L{MemCacheProtocol.stats}
+ command: it should parse the data sent by the server and call back the
+ resulting L{Deferred} with a dictionary of the received statistics.
+ """
+ return self._test(self.proto.stats(), "stats\r\n",
+ "STAT foo bar\r\nSTAT egg spam\r\nEND\r\n",
+ {"foo": "bar", "egg": "spam"})
+
+
+ def test_version(self):
+ """
+ Test version retrieval via the L{MemCacheProtocol.version} command: it
+ should return a L{Deferred} which is called back with the version sent
+ by the server.
+ """
+ return self._test(self.proto.version(), "version\r\n",
+ "VERSION 1.1\r\n", "1.1")
+
+
+ def test_flushAll(self):
+ """
+ L{MemCacheProtocol.flushAll} should return a L{Deferred} which is
+ called back with C{True} if the server acknowledges success.
+ """
+ return self._test(self.proto.flushAll(), "flush_all\r\n",
+ "OK\r\n", True)
+
+
+ def test_invalidGetResponse(self):
+ """
+ If the value returned doesn't match the expected key of the current, we
+ should get an error in L{MemCacheProtocol.dataReceived}.
+ """
+ self.proto.get("foo")
+ s = "spamegg"
+ self.assertRaises(RuntimeError,
+ self.proto.dataReceived,
+ "VALUE bar 0 %s\r\n%s\r\nEND\r\n" % (len(s), s))
+
+
+ def test_timeOut(self):
+ """
+ Test the timeout on outgoing requests: when timeout is detected, all
+ current commands should fail with a L{TimeoutError}, and the
+ connection should be closed.
+ """
+ d1 = self.proto.get("foo")
+ d2 = self.proto.get("bar")
+ d3 = Deferred()
+ self.proto.connectionLost = d3.callback
+
+ self.clock.advance(self.proto.persistentTimeOut)
+ self.assertFailure(d1, TimeoutError)
+ self.assertFailure(d2, TimeoutError)
+ def checkMessage(error):
+ self.assertEquals(str(error), "Connection timeout")
+ d1.addCallback(checkMessage)
+ return gatherResults([d1, d2, d3])
+
+
+ def test_timeoutRemoved(self):
+ """
+ When a request gets a response, no pending timeout call should remain
+ around.
+ """
+ d = self.proto.get("foo")
+
+ self.clock.advance(self.proto.persistentTimeOut - 1)
+ self.proto.dataReceived("VALUE foo 0 3\r\nbar\r\nEND\r\n")
+
+ def check(result):
+ self.assertEquals(result, (0, "bar"))
+ self.assertEquals(len(self.clock.calls), 0)
+ d.addCallback(check)
+ return d
+
+
+ def test_timeOutRaw(self):
+ """
+ Test the timeout when raw mode was started: the timeout should not be
+ reset until all the data has been received, so we can have a
+ L{TimeoutError} when waiting for raw data.
+ """
+ d1 = self.proto.get("foo")
+ d2 = Deferred()
+ self.proto.connectionLost = d2.callback
+
+ self.proto.dataReceived("VALUE foo 0 10\r\n12345")
+ self.clock.advance(self.proto.persistentTimeOut)
+ self.assertFailure(d1, TimeoutError)
+ return gatherResults([d1, d2])
+
+
+ def test_timeOutStat(self):
+ """
+ Test the timeout when stat command has started: the timeout should not
+ be reset until the final B{END} is received.
+ """
+ d1 = self.proto.stats()
+ d2 = Deferred()
+ self.proto.connectionLost = d2.callback
+
+ self.proto.dataReceived("STAT foo bar\r\n")
+ self.clock.advance(self.proto.persistentTimeOut)
+ self.assertFailure(d1, TimeoutError)
+ return gatherResults([d1, d2])
+
+
+ def test_timeoutPipelining(self):
+ """
+ When two requests are sent, a timeout call should remain around for the
+ second request, and its timeout time should be correct.
+ """
+ d1 = self.proto.get("foo")
+ d2 = self.proto.get("bar")
+ d3 = Deferred()
+ self.proto.connectionLost = d3.callback
+
+ self.clock.advance(self.proto.persistentTimeOut - 1)
+ self.proto.dataReceived("VALUE foo 0 3\r\nbar\r\nEND\r\n")
+
+ def check(result):
+ self.assertEquals(result, (0, "bar"))
+ self.assertEquals(len(self.clock.calls), 1)
+ for i in range(self.proto.persistentTimeOut):
+ self.clock.advance(1)
+ return self.assertFailure(d2, TimeoutError).addCallback(checkTime)
+ def checkTime(ignored):
+ # Check that the timeout happened C{self.proto.persistentTimeOut}
+ # after the last response
+ self.assertEquals(self.clock.seconds(),
+ 2 * self.proto.persistentTimeOut - 1)
+ d1.addCallback(check)
+ return d1
+
+
+ def test_timeoutNotReset(self):
+ """
+ Check that timeout is not resetted for every command, but keep the
+ timeout from the first command without response.
+ """
+ d1 = self.proto.get("foo")
+ d3 = Deferred()
+ self.proto.connectionLost = d3.callback
+
+ self.clock.advance(self.proto.persistentTimeOut - 1)
+ d2 = self.proto.get("bar")
+ self.clock.advance(1)
+ self.assertFailure(d1, TimeoutError)
+ self.assertFailure(d2, TimeoutError)
+ return gatherResults([d1, d2, d3])
+
+
+ def test_tooLongKey(self):
+ """
+ Test that an error is raised when trying to use a too long key: the
+ called command should return a L{Deferred} which fail with a
+ L{ClientError}.
+ """
+ d1 = self.assertFailure(self.proto.set("a" * 500, "bar"), ClientError)
+ d2 = self.assertFailure(self.proto.increment("a" * 500), ClientError)
+ d3 = self.assertFailure(self.proto.get("a" * 500), ClientError)
+ d4 = self.assertFailure(self.proto.append("a" * 500, "bar"), ClientError)
+ d5 = self.assertFailure(self.proto.prepend("a" * 500, "bar"), ClientError)
+ return gatherResults([d1, d2, d3, d4, d5])
+
+
+ def test_invalidCommand(self):
+ """
+ When an unknown command is sent directly (not through public API), the
+ server answers with an B{ERROR} token, and the command should fail with
+ L{NoSuchCommand}.
+ """
+ d = self.proto._set("egg", "foo", "bar", 0, 0, "")
+ self.assertEquals(self.transport.value(), "egg foo 0 0 3\r\nbar\r\n")
+ self.assertFailure(d, NoSuchCommand)
+ self.proto.dataReceived("ERROR\r\n")
+ return d
+
+
+ def test_clientError(self):
+ """
+ Test the L{ClientError} error: when the server send a B{CLIENT_ERROR}
+ token, the originating command should fail with L{ClientError}, and the
+ error should contain the text sent by the server.
+ """
+ a = "eggspamm"
+ d = self.proto.set("foo", a)
+ self.assertEquals(self.transport.value(),
+ "set foo 0 0 8\r\neggspamm\r\n")
+ self.assertFailure(d, ClientError)
+ def check(err):
+ self.assertEquals(str(err), "We don't like egg and spam")
+ d.addCallback(check)
+ self.proto.dataReceived("CLIENT_ERROR We don't like egg and spam\r\n")
+ return d
+
+
+ def test_serverError(self):
+ """
+ Test the L{ServerError} error: when the server send a B{SERVER_ERROR}
+ token, the originating command should fail with L{ServerError}, and the
+ error should contain the text sent by the server.
+ """
+ a = "eggspamm"
+ d = self.proto.set("foo", a)
+ self.assertEquals(self.transport.value(),
+ "set foo 0 0 8\r\neggspamm\r\n")
+ self.assertFailure(d, ServerError)
+ def check(err):
+ self.assertEquals(str(err), "zomg")
+ d.addCallback(check)
+ self.proto.dataReceived("SERVER_ERROR zomg\r\n")
+ return d
+
+
+ def test_unicodeKey(self):
+ """
+ Using a non-string key as argument to commands should raise an error.
+ """
+ d1 = self.assertFailure(self.proto.set(u"foo", "bar"), ClientError)
+ d2 = self.assertFailure(self.proto.increment(u"egg"), ClientError)
+ d3 = self.assertFailure(self.proto.get(1), ClientError)
+ d4 = self.assertFailure(self.proto.delete(u"bar"), ClientError)
+ d5 = self.assertFailure(self.proto.append(u"foo", "bar"), ClientError)
+ d6 = self.assertFailure(self.proto.prepend(u"foo", "bar"), ClientError)
+ return gatherResults([d1, d2, d3, d4, d5, d6])
+
+
+ def test_unicodeValue(self):
+ """
+ Using a non-string value should raise an error.
+ """
+ return self.assertFailure(self.proto.set("foo", u"bar"), ClientError)
+
+
+ def test_pipelining(self):
+ """
+ Test that multiple requests can be sent subsequently to the server, and
+ that the protocol order the responses correctly and dispatch to the
+ corresponding client command.
+ """
+ d1 = self.proto.get("foo")
+ d1.addCallback(self.assertEquals, (0, "bar"))
+ d2 = self.proto.set("bar", "spamspamspam")
+ d2.addCallback(self.assertEquals, True)
+ d3 = self.proto.get("egg")
+ d3.addCallback(self.assertEquals, (0, "spam"))
+ self.assertEquals(self.transport.value(),
+ "get foo\r\nset bar 0 0 12\r\nspamspamspam\r\nget egg\r\n")
+ self.proto.dataReceived("VALUE foo 0 3\r\nbar\r\nEND\r\n"
+ "STORED\r\n"
+ "VALUE egg 0 4\r\nspam\r\nEND\r\n")
+ return gatherResults([d1, d2, d3])
+
+
+ def test_getInChunks(self):
+ """
+ If the value retrieved by a C{get} arrive in chunks, the protocol
+ should be able to reconstruct it and to produce the good value.
+ """
+ d = self.proto.get("foo")
+ d.addCallback(self.assertEquals, (0, "0123456789"))
+ self.assertEquals(self.transport.value(), "get foo\r\n")
+ self.proto.dataReceived("VALUE foo 0 10\r\n0123456")
+ self.proto.dataReceived("789")
+ self.proto.dataReceived("\r\nEND")
+ self.proto.dataReceived("\r\n")
+ return d
+
+
+ def test_append(self):
+ """
+ L{MemCacheProtocol.append} behaves like a L{MemCacheProtocol.set}
+ method: it should return a L{Deferred} which is called back with
+ C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.append("foo", "bar"),
+ "append foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_prepend(self):
+ """
+ L{MemCacheProtocol.prepend} behaves like a L{MemCacheProtocol.set}
+ method: it should return a L{Deferred} which is called back with
+ C{True} when the operation succeeds.
+ """
+ return self._test(self.proto.prepend("foo", "bar"),
+ "prepend foo 0 0 3\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_gets(self):
+ """
+ L{MemCacheProtocol.get} should handle an additional cas result when
+ C{withIdentifier} is C{True} and forward it in the resulting
+ L{Deferred}.
+ """
+ return self._test(self.proto.get("foo", True), "gets foo\r\n",
+ "VALUE foo 0 3 1234\r\nbar\r\nEND\r\n", (0, "1234", "bar"))
+
+
+ def test_emptyGets(self):
+ """
+ Test getting a non-available key with gets: it should succeed but
+ return C{None} as value, C{0} as flag and an empty cas value.
+ """
+ return self._test(self.proto.get("foo", True), "gets foo\r\n",
+ "END\r\n", (0, "", None))
+
+
+ def test_checkAndSet(self):
+ """
+ L{MemCacheProtocol.checkAndSet} passes an additional cas identifier that the
+ server should handle to check if the data has to be updated.
+ """
+ return self._test(self.proto.checkAndSet("foo", "bar", cas="1234"),
+ "cas foo 0 0 3 1234\r\nbar\r\n", "STORED\r\n", True)
+
+
+ def test_casUnknowKey(self):
+ """
+ When L{MemCacheProtocol.checkAndSet} response is C{EXISTS}, the resulting
+ L{Deferred} should fire with C{False}.
+ """
+ return self._test(self.proto.checkAndSet("foo", "bar", cas="1234"),
+ "cas foo 0 0 3 1234\r\nbar\r\n", "EXISTS\r\n", False)
</ins></span></pre>
</div>
</div>
</body>
</html>