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
9 raise unittest.FailTest('Should have raised an exception.')
10
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
27
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 """
41 self.sent=[]
42 self.responses=list(responses)
43 self.connected = None
44 self.transport = FakeTransport(self)
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])
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
65 responses = self.responses.pop(0)
66 assert not responses
67 self.sent.append(op)
68
70 assert self.responses, 'Ran out of responses'
71 responses = self.responses.pop(0)
72 return responses
73
77
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
94 """TCP connection has opened"""
95 self.connected = 1
96
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
108
110 def createClient(factory):
111 factory.doStart()
112
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