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
10 raise unittest.FailTest('Should have raised an exception.')
11
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
28
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 """
46 self.sent=[]
47 self.responses=list(responses)
48 self.connected = None
49 self.transport = FakeTransport(self)
50
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
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
81 responses = self.responses.pop(0)
82 assert not responses
83 self.sent.append(op)
84
86 assert self.responses, 'Ran out of responses'
87 responses = self.responses.pop(0)
88 return responses
89
93
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
110 """TCP connection has opened"""
111 self.connected = 1
112
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
124
126 def createClient(factory):
127 factory.doStart()
128
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