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