apt_p2p_Khashmir/krpc.py

   1 ## Copyright 2002-2003 Andrew Loewenstern, All Rights Reserved
   2 # see LICENSE.txt for license information
   3
   4 """The KRPC communication protocol implementation.
   5
   6 @var KRPC_TIMEOUT: the number of seconds after which requests timeout
   7 @var UDP_PACKET_LIMIT: the maximum number of bytes that can be sent in a
   8     UDP packet without fragmentation
   9
  10 @var KRPC_ERROR: the code for a generic error
  11 @var KRPC_ERROR_SERVER_ERROR: the code for a server error
  12 @var KRPC_ERROR_MALFORMED_PACKET: the code for a malformed packet error
  13 @var KRPC_ERROR_METHOD_UNKNOWN: the code for a method unknown error
  14 @var KRPC_ERROR_MALFORMED_REQUEST: the code for a malformed request error
  15 @var KRPC_ERROR_INVALID_TOKEN: the code for an invalid token error
  16 @var KRPC_ERROR_RESPONSE_TOO_LONG: the code for a response too long error
  17
  18 @var KRPC_ERROR_INTERNAL: the code for an internal error
  19 @var KRPC_ERROR_RECEIVED_UNKNOWN: the code for an unknown message type error
  20 @var KRPC_ERROR_TIMEOUT: the code for a timeout error
  21 @var KRPC_ERROR_PROTOCOL_STOPPED: the code for a stopped protocol error
  22
  23 @var TID: the identifier for the transaction ID
  24 @var REQ: the identifier for a request packet
  25 @var RSP: the identifier for a response packet
  26 @var TYP: the identifier for the type of packet
  27 @var ARG: the identifier for the argument to the request
  28 @var ERR: the identifier for an error packet
  29
  30 @group Remote node error codes: KRPC_ERROR, KRPC_ERROR_SERVER_ERROR,
  31     KRPC_ERROR_MALFORMED_PACKET, KRPC_ERROR_METHOD_UNKNOWN,
  32     KRPC_ERROR_MALFORMED_REQUEST, KRPC_ERROR_INVALID_TOKEN,
  33     KRPC_ERROR_RESPONSE_TOO_LONG
  34 @group Local node error codes: KRPC_ERROR_INTERNAL, KRPC_ERROR_RECEIVED_UNKNOWN,
  35     KRPC_ERROR_TIMEOUT, KRPC_ERROR_PROTOCOL_STOPPED
  36 @group Command identifiers: TID, REQ, RSP, TYP, ARG, ERR
  37
  38 """
  39
  40 from bencode import bencode, bdecode
  41 from time import asctime
  42 from math import ceil
  43
  44 from twisted.internet.defer import Deferred
  45 from twisted.internet import protocol, reactor
  46 from twisted.python import log
  47 from twisted.trial import unittest
  48
  49 from khash import newID
  50
  51 KRPC_TIMEOUT = 20
  52 UDP_PACKET_LIMIT = 1472
  53
  54 # Remote node errors
  55 KRPC_ERROR = 200
  56 KRPC_ERROR_SERVER_ERROR = 201
  57 KRPC_ERROR_MALFORMED_PACKET = 202
  58 KRPC_ERROR_METHOD_UNKNOWN = 203
  59 KRPC_ERROR_MALFORMED_REQUEST = 204
  60 KRPC_ERROR_INVALID_TOKEN = 205
  61 KRPC_ERROR_RESPONSE_TOO_LONG = 206
  62
  63 # Local errors
  64 KRPC_ERROR_INTERNAL = 100
  65 KRPC_ERROR_RECEIVED_UNKNOWN = 101
  66 KRPC_ERROR_TIMEOUT = 102
  67 KRPC_ERROR_PROTOCOL_STOPPED = 103
  68
  69 # commands
  70 TID = 't'
  71 REQ = 'q'
  72 RSP = 'r'
  73 TYP = 'y'
  74 ARG = 'a'
  75 ERR = 'e'
  76
  77 class KrpcError(Exception):
  78     """An error occurred in the KRPC protocol."""
  79     pass
  80
  81 def verifyMessage(msg):
  82     """Check received message for corruption and errors.
  83
  84     @type msg: C{dictionary}
  85     @param msg: the dictionary of information received on the connection
  86     @raise KrpcError: if the message is corrupt
  87     """
  88
  89     if type(msg) != dict:
  90         raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "not a dictionary")
  91     if TYP not in msg:
  92         raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "no message type")
  93     if msg[TYP] == REQ:
  94         if REQ not in msg:
  95             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "request type not specified")
  96         if type(msg[REQ]) != str:
  97             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "request type is not a string")
  98         if ARG not in msg:
  99             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "no arguments for request")
 100         if type(msg[ARG]) != dict:
 101             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "arguments for request are not in a dictionary")
 102     elif msg[TYP] == RSP:
 103         if RSP not in msg:
 104             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "response not specified")
 105         if type(msg[RSP]) != dict:
 106             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "response is not a dictionary")
 107     elif msg[TYP] == ERR:
 108         if ERR not in msg:
 109             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "error not specified")
 110         if type(msg[ERR]) != list:
 111             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "error is not a list")
 112         if len(msg[ERR]) != 2:
 113             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "error is not a 2-element list")
 114         if type(msg[ERR][0]) not in (int, long):
 115             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "error number is not a number")
 116         if type(msg[ERR][1]) != str:
 117             raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "error string is not a string")
 118 #    else:
 119 #        raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "unknown message type")
 120     if TID not in msg:
 121         raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "no transaction ID specified")
 122     if type(msg[TID]) != str:
 123         raise KrpcError, (KRPC_ERROR_MALFORMED_PACKET, "transaction id is not a string")
 124
 125 class hostbroker(protocol.DatagramProtocol):
 126     """The factory for the KRPC protocol.
 127
 128     @type server: L{khashmir.Khashmir}
 129     @ivar server: the main Khashmir program
 130     @type stats: L{stats.StatsLogger}
 131     @ivar stats: the statistics logger to save transport info
 132     @type config: C{dictionary}
 133     @ivar config: the configuration parameters for the DHT
 134     @type connections: C{dictionary}
 135     @ivar connections: all the connections that have ever been made to the
 136         protocol, keys are IP address and port pairs, values are L{KRPC}
 137         protocols for the addresses
 138     @ivar protocol: the protocol to use to handle incoming connections
 139         (added externally)
 140     @type addr: (C{string}, C{int})
 141     @ivar addr: the IP address and port of this node
 142     """
 143
 144     def __init__(self, server, stats, config):
 145         """Initialize the factory.
 146
 147         @type server: L{khashmir.Khashmir}
 148         @param server: the main DHT program
 149         @type stats: L{stats.StatsLogger}
 150         @param stats: the statistics logger to save transport info
 151         @type config: C{dictionary}
 152         @param config: the configuration parameters for the DHT
 153         """
 154         self.server = server
 155         self.stats = stats
 156         self.config = config
 157         # this should be changed to storage that drops old entries
 158         self.connections = {}
 159
 160     def datagramReceived(self, datagram, addr):
 161         """Optionally create a new protocol object, and handle the new datagram.
 162
 163         @type datagram: C{string}
 164         @param datagram: the data received from the transport.
 165         @type addr: (C{string}, C{int})
 166         @param addr: source IP address and port of datagram.
 167         """
 168         c = self.connectionForAddr(addr)
 169         c.datagramReceived(datagram, addr)
 170         #if c.idle():
 171         #    del self.connections[addr]
 172
 173     def connectionForAddr(self, addr):
 174         """Get a protocol object for the source.
 175
 176         @type addr: (C{string}, C{int})
 177         @param addr: source IP address and port of datagram.
 178         """
 179         # Don't connect to ourself
 180         if addr == self.addr:
 181             raise KrcpError
 182
 183         # Create a new protocol object if necessary
 184         if not self.connections.has_key(addr):
 185             conn = self.protocol(addr, self.server, self.stats, self.transport, self.config['SPEW'])
 186             self.connections[addr] = conn
 187         else:
 188             conn = self.connections[addr]
 189         return conn
 190
 191     def makeConnection(self, transport):
 192         """Make a connection to a transport and save our address."""
 193         protocol.DatagramProtocol.makeConnection(self, transport)
 194         tup = transport.getHost()
 195         self.addr = (tup.host, tup.port)
 196
 197     def stopProtocol(self):
 198         """Stop all the open connections."""
 199         for conn in self.connections.values():
 200             conn.stop()
 201         protocol.DatagramProtocol.stopProtocol(self)
 202
 203 class KRPC:
 204     """The KRPC protocol implementation.
 205
 206     @ivar transport: the transport to use for the protocol
 207     @type factory: L{khashmir.Khashmir}
 208     @ivar factory: the main Khashmir program
 209     @type stats: L{stats.StatsLogger}
 210     @ivar stats: the statistics logger to save transport info
 211     @type addr: (C{string}, C{int})
 212     @ivar addr: the IP address and port of the source node
 213     @type noisy: C{boolean}
 214     @ivar noisy: whether to log additional details of the protocol
 215     @type tids: C{dictionary}
 216     @ivar tids: the transaction IDs outstanding for requests, keys are the
 217         transaction ID of the request, values are the deferreds to call with
 218         the results
 219     @type stopped: C{boolean}
 220     @ivar stopped: whether the protocol has been stopped
 221     """
 222
 223     def __init__(self, addr, server, stats, transport, spew = False):
 224         """Initialize the protocol.
 225
 226         @type addr: (C{string}, C{int})
 227         @param addr: the IP address and port of the source node
 228         @type server: L{khashmir.Khashmir}
 229         @param server: the main Khashmir program
 230         @type stats: L{stats.StatsLogger}
 231         @param stats: the statistics logger to save transport info
 232         @param transport: the transport to use for the protocol
 233         @type spew: C{boolean}
 234         @param spew: whether to log additional details of the protocol
 235             (optional, defaults to False)
 236         """
 237         self.transport = transport
 238         self.factory = server
 239         self.stats = stats
 240         self.addr = addr
 241         self.noisy = spew
 242         self.tids = {}
 243         self.stopped = False
 244
 245     def datagramReceived(self, data, addr):
 246         """Process the new datagram.
 247
 248         @type data: C{string}
 249         @param data: the data received from the transport.
 250         @type addr: (C{string}, C{int})
 251         @param addr: source IP address and port of datagram.
 252         """
 253         self.stats.receivedBytes(len(data))
 254         if self.stopped:
 255             if self.noisy:
 256                 log.msg("stopped, dropping message from %r: %s" % (addr, data))
 257
 258         # Bdecode the message
 259         try:
 260             msg = bdecode(data)
 261         except Exception, e:
 262             if self.noisy:
 263                 log.msg("krpc bdecode error: ")
 264                 log.err(e)
 265             return
 266
 267         # Make sure the remote node isn't trying anything funny
 268         try:
 269             verifyMessage(msg)
 270         except Exception, e:
 271             log.msg("krpc message verification error: ")
 272             log.err(e)
 273             return
 274
 275         if self.noisy:
 276             log.msg("%d received from %r: %s" % (self.factory.port, addr, msg))
 277
 278         # Process it based on its type
 279         if msg[TYP]  == REQ:
 280             ilen = len(data)
 281
 282             # Requests are handled by the factory
 283             f = getattr(self.factory ,"krpc_" + msg[REQ], None)
 284             msg[ARG]['_krpc_sender'] =  self.addr
 285             if f and callable(f):
 286                 self.stats.receivedAction(msg[REQ])
 287                 try:
 288                     ret = f(*(), **msg[ARG])
 289                 except KrpcError, e:
 290                     log.msg('Got a Krpc error while running: krpc_%s' % msg[REQ])
 291                     log.err(e)
 292                     self.stats.errorAction(msg[REQ])
 293                     olen = self._sendResponse(msg[REQ], addr, msg[TID], ERR,
 294                                               [e[0], e[1]])
 295                 except TypeError, e:
 296                     log.msg('Got a malformed request for: krpc_%s' % msg[REQ])
 297                     log.err(e)
 298                     self.stats.errorAction(msg[REQ])
 299                     olen = self._sendResponse(msg[REQ], addr, msg[TID], ERR,
 300                                               [KRPC_ERROR_MALFORMED_REQUEST, str(e)])
 301                 except Exception, e:
 302                     log.msg('Got an unknown error while running: krpc_%s' % msg[REQ])
 303                     log.err(e)
 304                     self.stats.errorAction(msg[REQ])
 305                     olen = self._sendResponse(msg[REQ], addr, msg[TID], ERR,
 306                                               [KRPC_ERROR_SERVER_ERROR, str(e)])
 307                 else:
 308                     olen = self._sendResponse(msg[REQ], addr, msg[TID], RSP, ret)
 309             else:
 310                 # Request for unknown method
 311                 log.msg("ERROR: don't know about method %s" % msg[REQ])
 312                 self.stats.receivedAction('unknown')
 313                 olen = self._sendResponse(msg[REQ], addr, msg[TID], ERR,
 314                                           [KRPC_ERROR_METHOD_UNKNOWN, "unknown method "+str(msg[REQ])])
 315             if self.noisy:
 316                 log.msg("%s >>> %s - %s %s %s" % (addr, self.factory.node.port,
 317                                                   ilen, msg[REQ], olen))
 318         elif msg[TYP] == RSP:
 319             # Responses get processed by their TID's deferred
 320             if self.tids.has_key(msg[TID]):
 321                 df = self.tids[msg[TID]]
 322                 #       callback
 323                 del(self.tids[msg[TID]])
 324                 df.callback({'rsp' : msg[RSP], '_krpc_sender': addr})
 325             else:
 326                 # no tid, this transaction timed out already...
 327                 if self.noisy:
 328                     log.msg('timeout: %r' % msg[RSP]['id'])
 329         elif msg[TYP] == ERR:
 330             # Errors get processed by their TID's deferred's errback
 331             if self.tids.has_key(msg[TID]):
 332                 df = self.tids[msg[TID]]
 333                 del(self.tids[msg[TID]])
 334                 # callback
 335                 df.errback(KrpcError(*msg[ERR]))
 336             else:
 337                 # day late and dollar short, just log it
 338                 log.msg("Got an error for an unknown request: %r" % (msg[ERR], ))
 339                 pass
 340         else:
 341             # Received an unknown message type
 342             if self.noisy:
 343                 log.msg("unknown message type: %r" % msg)
 344             if msg[TID] in self.tids:
 345                 df = self.tids[msg[TID]]
 346                 del(self.tids[msg[TID]])
 347                 # callback
 348                 df.errback(KrpcError(KRPC_ERROR_RECEIVED_UNKNOWN,
 349                                      "Received an unknown message type: %r" % msg[TYP]))
 350
 351     def _sendResponse(self, request, addr, tid, msgType, response):
 352         """Helper function for sending responses to nodes.
 353
 354         @param request: the name of the requested method
 355         @type addr: (C{string}, C{int})
 356         @param addr: source IP address and port of datagram.
 357         @param tid: the transaction ID of the request
 358         @param msgType: the type of message to respond with
 359         @param response: the arguments for the response
 360         """
 361         if not response:
 362             response = {}
 363
 364         try:
 365             # Create the response message
 366             msg = {TID : tid, TYP : msgType, msgType : response}
 367
 368             if self.noisy:
 369                 log.msg("%d responding to %r: %s" % (self.factory.port, addr, msg))
 370
 371             out = bencode(msg)
 372
 373             # Make sure its not too long
 374             if len(out) > UDP_PACKET_LIMIT:
 375                 # Can we remove some values to shorten it?
 376                 if 'values' in response:
 377                     # Save the original list of values
 378                     orig_values = response['values']
 379                     len_orig_values = len(bencode(orig_values))
 380
 381                     # Caclulate the maximum value length possible
 382                     max_len_values = len_orig_values - (len(out) - UDP_PACKET_LIMIT)
 383                     assert max_len_values > 0
 384
 385                     # Start with a calculation of how many values should be included
 386                     # (assumes all values are the same length)
 387                     per_value = (float(len_orig_values) - 2.0) / float(len(orig_values))
 388                     num_values = len(orig_values) - int(ceil(float(len(out) - UDP_PACKET_LIMIT) / per_value))
 389
 390                     # Do a linear search for the actual maximum number possible
 391                     bencoded_values = len(bencode(orig_values[:num_values]))
 392                     while bencoded_values < max_len_values and num_values + 1 < len(orig_values):
 393                         bencoded_values += len(bencode(orig_values[num_values]))
 394                         num_values += 1
 395                     while bencoded_values > max_len_values and num_values > 0:
 396                         num_values -= 1
 397                         bencoded_values -= len(bencode(orig_values[num_values]))
 398                     assert num_values > 0
 399
 400                     # Encode the result
 401                     response['values'] = orig_values[:num_values]
 402                     out = bencode(msg)
 403                     assert len(out) < UDP_PACKET_LIMIT
 404                     log.msg('Shortened a long packet from %d to %d values, new packet length: %d' %
 405                             (len(orig_values), num_values, len(out)))
 406                 else:
 407                     # Too long a response, send an error
 408                     log.msg('Could not send response, too long: %d bytes' % len(out))
 409                     self.stats.errorAction(request)
 410                     msg = {TID : tid, TYP : ERR, ERR : [KRPC_ERROR_RESPONSE_TOO_LONG, "response was %d bytes" % len(out)]}
 411                     out = bencode(msg)
 412
 413         except Exception, e:
 414             # Unknown error, send an error message
 415             self.stats.errorAction(request)
 416             msg = {TID : tid, TYP : ERR, ERR : [KRPC_ERROR_SERVER_ERROR, "unknown error sending response: %s" % str(e)]}
 417             out = bencode(msg)
 418
 419         self.stats.sentBytes(len(out))
 420         self.transport.write(out, addr)
 421         return len(out)
 422
 423     def sendRequest(self, method, args):
 424         """Send a request to the remote node.
 425
 426         @type method: C{string}
 427         @param method: the methiod name to call on the remote node
 428         @param args: the arguments to send to the remote node's method
 429         """
 430         if self.stopped:
 431             raise KrpcError, (KRPC_ERROR_PROTOCOL_STOPPED, "cannot send, connection has been stopped")
 432
 433         # Create the request message
 434         msg = {TID : newID(), TYP : REQ,  REQ : method, ARG : args}
 435         if self.noisy:
 436             log.msg("%d sending to %r: %s" % (self.factory.port, self.addr, msg))
 437         data = bencode(msg)
 438
 439         # Create the deferred and save it with the TID
 440         d = Deferred()
 441         self.tids[msg[TID]] = d
 442
 443         # Save the conclusion of the action
 444         d.addCallbacks(self.stats.responseAction, self.stats.failedAction,
 445                        callbackArgs = (method, ), errbackArgs = (method, ))
 446
 447         # Schedule a later timeout call
 448         def timeOut(tids = self.tids, id = msg[TID], method = method, addr = self.addr):
 449             """Call the deferred's errback if a timeout occurs."""
 450             if tids.has_key(id):
 451                 df = tids[id]
 452                 del(tids[id])
 453                 df.errback(KrpcError(KRPC_ERROR_TIMEOUT, "timeout waiting for '%s' from %r" % (method, addr)))
 454         later = reactor.callLater(KRPC_TIMEOUT, timeOut)
 455
 456         # Cancel the timeout call if a response is received
 457         def dropTimeOut(dict, later_call = later):
 458             """Cancel the timeout call when a response is received."""
 459             if later_call.active():
 460                 later_call.cancel()
 461             return dict
 462         d.addBoth(dropTimeOut)
 463
 464         # Save some stats
 465         self.stats.sentAction(method)
 466         self.stats.sentBytes(len(data))
 467
 468         self.transport.write(data, self.addr)
 469         return d
 470
 471     def stop(self):
 472         """Timeout all pending requests."""
 473         for df in self.tids.values():
 474             df.errback(KrpcError(KRPC_ERROR_PROTOCOL_STOPPED,
 475                                  'connection has been stopped while waiting for response'))
 476         self.tids = {}
 477         self.stopped = True
 478
 479 #{ For testing the KRPC protocol
 480 def connectionForAddr(host, port):
 481     return host
 482
 483 class Receiver(protocol.Factory):
 484     protocol = KRPC
 485     def __init__(self):
 486         self.buf = []
 487     def krpc_store(self, msg, _krpc_sender):
 488         self.buf += [msg]
 489         return {}
 490     def krpc_echo(self, msg, _krpc_sender):
 491         return {'msg': msg}
 492     def krpc_values(self, length, num, _krpc_sender):
 493         return {'values': ['1'*length]*num}
 494
 495 def make(port):
 496     af = Receiver()
 497     a = hostbroker(af, {'SPEW': False})
 498     a.protocol = KRPC
 499     p = reactor.listenUDP(port, a)
 500     return af, a, p
 501
 502 class KRPCTests(unittest.TestCase):
 503     timeout = 2
 504
 505     def setUp(self):
 506         self.af, self.a, self.ap = make(1180)
 507         self.bf, self.b, self.bp = make(1181)
 508
 509     def tearDown(self):
 510         self.ap.stopListening()
 511         self.bp.stopListening()
 512
 513     def bufEquals(self, result, value):
 514         self.failUnlessEqual(self.bf.buf, value)
 515
 516     def testSimpleMessage(self):
 517         d = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('store', {'msg' : "This is a test."})
 518         d.addCallback(self.bufEquals, ["This is a test."])
 519         return d
 520
 521     def testMessageBlast(self):
 522         for i in range(100):
 523             d = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('store', {'msg' : "This is a test."})
 524         d.addCallback(self.bufEquals, ["This is a test."] * 100)
 525         return d
 526
 527     def testEcho(self):
 528         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is a test."})
 529         df.addCallback(self.gotMsg, "This is a test.")
 530         return df
 531
 532     def gotMsg(self, dict, should_be):
 533         _krpc_sender = dict['_krpc_sender']
 534         msg = dict['rsp']
 535         self.failUnlessEqual(msg['msg'], should_be)
 536
 537     def testManyEcho(self):
 538         for i in xrange(100):
 539             df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is a test."})
 540             df.addCallback(self.gotMsg, "This is a test.")
 541         return df
 542
 543     def testMultiEcho(self):
 544         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is a test."})
 545         df.addCallback(self.gotMsg, "This is a test.")
 546
 547         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is another test."})
 548         df.addCallback(self.gotMsg, "This is another test.")
 549
 550         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is yet another test."})
 551         df.addCallback(self.gotMsg, "This is yet another test.")
 552
 553         return df
 554
 555     def testEchoReset(self):
 556         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is a test."})
 557         df.addCallback(self.gotMsg, "This is a test.")
 558
 559         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is another test."})
 560         df.addCallback(self.gotMsg, "This is another test.")
 561         df.addCallback(self.echoReset)
 562         return df
 563
 564     def echoReset(self, dict):
 565         del(self.a.connections[('127.0.0.1', 1181)])
 566         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is yet another test."})
 567         df.addCallback(self.gotMsg, "This is yet another test.")
 568         return df
 569
 570     def testUnknownMeth(self):
 571         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('blahblah', {'msg' : "This is a test."})
 572         df.addBoth(self.gotErr, KRPC_ERROR_METHOD_UNKNOWN)
 573         return df
 574
 575     def testMalformedRequest(self):
 576         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('echo', {'msg' : "This is a test.", 'foo': 'bar'})
 577         df.addBoth(self.gotErr, KRPC_ERROR_MALFORMED_REQUEST)
 578         return df
 579
 580     def gotErr(self, err, should_be):
 581         self.failUnlessEqual(err.value[0], should_be)
 582
 583     def testLongPackets(self):
 584         df = self.a.connectionForAddr(('127.0.0.1', 1181)).sendRequest('values', {'length' : 1, 'num': 2000})
 585         df.addCallback(self.gotLongRsp)
 586         return df
 587
 588     def gotLongRsp(self, dict):
 589         # Not quite accurate, but good enough
 590         self.failUnless(len(bencode(dict))-10 < UDP_PACKET_LIMIT)
 591