[CalendarServer-changes] [2439] CalendarServer/branches/unified-cache/twistedcaldav

source_changes at macosforge.org source_changes at macosforge.org
Wed May 21 16:11:23 PDT 2008


Revision: 2439
          http://trac.macosforge.org/projects/calendarserver/changeset/2439
Author:   dreid at apple.com
Date:     2008-05-21 16:11:19 -0700 (Wed, 21 May 2008)

Log Message:
-----------
Add twisted memcached protocol implementation

Added Paths:
-----------
    CalendarServer/branches/unified-cache/twistedcaldav/memcache.py
    CalendarServer/branches/unified-cache/twistedcaldav/test/test_memcache.py

Added: CalendarServer/branches/unified-cache/twistedcaldav/memcache.py
===================================================================
--- 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)
@@ -0,0 +1,657 @@
+# -*- 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"]
+

Added: CalendarServer/branches/unified-cache/twistedcaldav/test/test_memcache.py
===================================================================
--- 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)
@@ -0,0 +1,510 @@
+# 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)

-------------- next part --------------
An HTML attachment was scrubbed...
URL: http://lists.macosforge.org/pipermail/calendarserver-changes/attachments/20080521/11116306/attachment-0001.htm 


More information about the calendarserver-changes mailing list