Package ldaptor :: Module testutil
[hide private]
[frames] | no frames]

Source Code for Module ldaptor.testutil

  1  """Utilities for writing Twistedy unit tests and debugging.""" 
  2   
  3  from twisted.internet import defer 
  4  from twisted.python import failure 
  5  from twisted.trial import unittest 
  6  from twisted.test import proto_helpers 
  7  from ldaptor import config 
  8   
9 -def mustRaise(dummy):
10 raise unittest.FailTest('Should have raised an exception.')
11
12 -def calltrace():
13 """Print out all function calls. For debug use only.""" 14 def printfuncnames(frame, event, arg): 15 print "|%s: %s:%d:%s" % (event, 16 frame.f_code.co_filename, 17 frame.f_code.co_firstlineno, 18 frame.f_code.co_name)
19 import sys 20 sys.setprofile(printfuncnames) 21
22 -class FakeTransport:
23 - def __init__(self, proto):
24 self.proto = proto
25
26 - def loseConnection(self):
27 self.proto.connectionLost()
28
29 -class LDAPClientTestDriver:
30 """ 31 32 A test driver that looks somewhat like a real LDAPClient. 33 34 Pass in a list of lists of LDAPProtocolResponses. For each sent 35 LDAP message, the first item of said list is iterated through, and 36 all the items are sent as responses to the callback. The sent LDAP 37 messages are stored in self.sent, so you can assert that the sent 38 messages are what they are supposed to be. 39 40 It is also possible to include a Failure instance instead of a list 41 of LDAPProtocolResponses which will cause the errback to be called 42 with the failure. 43 44 """
45 - def __init__(self, *responses):
46 self.sent=[] 47 self.responses=list(responses) 48 self.connected = None 49 self.transport = FakeTransport(self)
50
51 - def send(self, op):
52 self.sent.append(op) 53 l = self._response() 54 assert len(l) == 1, \ 55 "got %d responses for a .send()" % len(l) 56 r = l[0] 57 if isinstance(r, failure.Failure): 58 return defer.fail(r) 59 else: 60 return defer.succeed(r)
61
62 - def send_multiResponse(self, op, handler, *args, **kwargs):
63 d = defer.Deferred() 64 self.sent.append(op) 65 responses = self._response() 66 while responses: 67 r = responses.pop(0) 68 if isinstance(r, failure.Failure): 69 d.errback(r) 70 break 71 ret = handler(r, *args, **kwargs) 72 if responses: 73 assert not ret, \ 74 "got %d responses still to give, but handler wants none (got %r)." % (len(responses), ret) 75 else: 76 assert ret, \ 77 "no more responses to give, but handler still wants more (got %r)." % ret 78 return d
79
80 - def send_noResponse(self, op):
81 responses = self.responses.pop(0) 82 assert not responses 83 self.sent.append(op)
84
85 - def _response(self):
86 assert self.responses, 'Ran out of responses' 87 responses = self.responses.pop(0) 88 return responses
89
90 - def assertNothingSent(self):
91 # just a bit more explicit 92 self.assertSent()
93
94 - def assertSent(self, *shouldBeSent):
95 shouldBeSent = list(shouldBeSent) 96 assert self.sent == shouldBeSent, \ 97 '%s expected to send %r but sent %r' % ( 98 self.__class__.__name__, 99 shouldBeSent, 100 self.sent) 101 sentStr = ''.join([str(x) for x in self.sent]) 102 shouldBeSentStr = ''.join([str(x) for x in shouldBeSent]) 103 assert sentStr == shouldBeSentStr, \ 104 '%s expected to send data %r but sent %r' % ( 105 self.__class__.__name__, 106 shouldBeSentStr, 107 sentStr)
108
109 - def connectionMade(self):
110 """TCP connection has opened""" 111 self.connected = 1
112
113 - def connectionLost(self, reason=None):
114 """Called when TCP connection has been lost""" 115 assert not self.responses, \ 116 "connectionLost called even when have responses left: %r" % self.responses 117 self.connected = 0
118
119 - def unbind(self):
120 assert self.connected 121 r='fake-unbind-by-LDAPClientTestDriver' 122 self.send_noResponse(r) 123 self.transport.loseConnection()
124
125 -def createServer(proto, *responses, **kw):
126 def createClient(factory): 127 factory.doStart() 128 #TODO factory.startedConnecting(c) 129 proto = factory.buildProtocol(addr=None) 130 proto.connectionMade()
131 cfg = config.loadConfig( 132 configFiles=[], 133 reload=True) 134 overrides = kw.setdefault('serviceLocationOverrides', {}) 135 overrides.setdefault('', createClient) 136 conf = config.LDAPConfig(**kw) 137 server = proto(conf) 138 server.protocol = lambda : LDAPClientTestDriver(*responses) 139 server.transport = proto_helpers.StringTransport() 140 server.connectionMade() 141 return server 142