1 | # Copyright (c) 2003-2008 Ralph Meijer |
---|
2 | # See LICENSE for details. |
---|
3 | |
---|
4 | """ |
---|
5 | Unit test helpers. |
---|
6 | """ |
---|
7 | |
---|
8 | from twisted.internet import defer |
---|
9 | from twisted.words.xish import xpath |
---|
10 | from twisted.words.xish.utility import EventDispatcher |
---|
11 | |
---|
12 | from wokkel.generic import parseXml |
---|
13 | |
---|
14 | class XmlStreamStub(object): |
---|
15 | """ |
---|
16 | Stub for testing objects that communicate through XML Streams. |
---|
17 | |
---|
18 | Instances of this stub hold an object in L{xmlstream} that acts like an |
---|
19 | L{XmlStream<twisted.words.xish.xmlstream.XmlStream} after connection stream |
---|
20 | initialization. Stanzas can be sent through it by calling its C{send} |
---|
21 | method with an object implementing |
---|
22 | L{IElement<twisted.words.xish.domish.IElement>} as its first argument. |
---|
23 | These appear in sequence in the L{output} instance variable of the stub. |
---|
24 | |
---|
25 | For the reverse direction, stanzas passed to L{send} of the stub, will be |
---|
26 | dispatched in the stubbed XmlStream as if it was received over the wire, so |
---|
27 | that registered observers will be called. |
---|
28 | |
---|
29 | Example: |
---|
30 | |
---|
31 | >>> stub = XmlStreamStub() |
---|
32 | >>> stub.xmlstream.send(domish.Element((None, 'presence'))) |
---|
33 | >>> stub.output[-1].toXml() |
---|
34 | u'<presence/>' |
---|
35 | >>> def cb(stanza): |
---|
36 | ... print "Got: %r" stanza.toXml() |
---|
37 | >>> stub.xmlstream.addObserver('/presence') |
---|
38 | >>> stub.send(domish.Element((None, 'presence'))) |
---|
39 | Got: u'<presence/>' |
---|
40 | |
---|
41 | @ivar xmlstream: Stubbed XML Stream. |
---|
42 | @type xmlstream: L{EventDispatcher} |
---|
43 | @ivar output: List of stanzas sent to the XML Stream. |
---|
44 | @type output: L{list} |
---|
45 | """ |
---|
46 | |
---|
47 | def __init__(self): |
---|
48 | self.output = [] |
---|
49 | self.xmlstream = EventDispatcher() |
---|
50 | self.xmlstream.send = self.output.append |
---|
51 | |
---|
52 | def send(self, obj): |
---|
53 | """ |
---|
54 | Pass an element to the XML Stream as if received. |
---|
55 | |
---|
56 | @param obj: Element to be dispatched to C{self.xmlstream}. |
---|
57 | @type obj: object implementing |
---|
58 | L{IElement<twisted.words.xish.domish.IElement>}. |
---|
59 | """ |
---|
60 | self.xmlstream.dispatch(obj) |
---|
61 | |
---|
62 | |
---|
63 | class TestableRequestHandlerMixin(object): |
---|
64 | """ |
---|
65 | Mixin for testing XMPPHandlers that process iq requests. |
---|
66 | |
---|
67 | Handlers that use L{wokkel.subprotocols.IQHandlerMixin} define a |
---|
68 | C{iqHandlers} attribute that lists the handlers to be called for iq |
---|
69 | requests. This mixin provides L{handleRequest} to mimic the handler |
---|
70 | processing for easier testing. |
---|
71 | """ |
---|
72 | |
---|
73 | def handleRequest(self, xml): |
---|
74 | """ |
---|
75 | Find a handler and call it directly. |
---|
76 | |
---|
77 | @param xml: XML stanza that may yield a handler being called. |
---|
78 | @type xml: C{str}. |
---|
79 | @return: Deferred that fires with the result of a handler for this |
---|
80 | stanza. If no handler was found, the deferred has its errback |
---|
81 | called with a C{NotImplementedError} exception. |
---|
82 | """ |
---|
83 | handler = None |
---|
84 | iq = parseXml(xml) |
---|
85 | for queryString, method in self.service.iqHandlers.iteritems(): |
---|
86 | if xpath.internQuery(queryString).matches(iq): |
---|
87 | handler = getattr(self.service, method) |
---|
88 | |
---|
89 | if handler: |
---|
90 | d = defer.maybeDeferred(handler, iq) |
---|
91 | else: |
---|
92 | d = defer.fail(NotImplementedError()) |
---|
93 | |
---|
94 | return d |
---|