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