source: wokkel/test/test_ping.py @ 68:2c6d1f895583

Last change on this file since 68:2c6d1f895583 was 65:6936acbdb9bd, checked in by Ralph Meijer <ralphm@…>, 13 years ago

Add support for the XMPP Ping protocol.

Author: ralphm
Fixes #55.

This adds a whole bunch of examples around the ping protocol in different
settings for both the pinger and pingee: as a external server-side component,
as a standalone server with server-to-server connectivity and as a client.

Also updates the disco support to allow for non-deferred responses from
subprotocol handlers

File size: 5.4 KB
Line 
1# Copyright (c) 2003-2009 Ralph Meijer
2# See LICENSE for details.
3
4"""
5Tests for L{wokkel.ping}.
6"""
7
8from zope.interface import verify
9
10from twisted.internet import defer
11from twisted.trial import unittest
12from twisted.words.protocols.jabber.error import StanzaError
13from twisted.words.protocols.jabber.jid import JID
14from twisted.words.protocols.jabber.xmlstream import toResponse
15
16from wokkel import disco, iwokkel, ping
17from wokkel.generic import parseXml
18from wokkel.test.helpers import XmlStreamStub
19
20class PingClientProtocolTest(unittest.TestCase):
21    """
22    Tests for L{ping.PingClientProtocol}.
23    """
24
25    def setUp(self):
26        self.stub = XmlStreamStub()
27        self.protocol = ping.PingClientProtocol()
28        self.protocol.xmlstream = self.stub.xmlstream
29        self.protocol.connectionInitialized()
30
31
32    def test_ping(self):
33        """
34        Pinging a service should fire a deferred with None
35        """
36        def cb(result):
37            self.assertIdentical(None, result)
38
39        d = self.protocol.ping(JID("example.com"))
40        d.addCallback(cb)
41
42        iq = self.stub.output[-1]
43        self.assertEqual(u'example.com', iq.getAttribute(u'to'))
44        self.assertEqual(u'get', iq.getAttribute(u'type'))
45        self.assertEqual('urn:xmpp:ping', iq.ping.uri)
46
47        response = toResponse(iq, u'result')
48        self.stub.send(response)
49
50        return d
51
52
53    def test_pingWithSender(self):
54        """
55        Pinging a service with a sender address should include that address.
56        """
57        d = self.protocol.ping(JID("example.com"),
58                               sender=JID('user@example.com'))
59
60        iq = self.stub.output[-1]
61        self.assertEqual(u'user@example.com', iq.getAttribute(u'from'))
62
63        response = toResponse(iq, u'result')
64        self.stub.send(response)
65
66        return d
67
68
69    def test_pingNotSupported(self):
70        """
71        Pinging a service should fire a deferred with None if not supported.
72        """
73        def cb(result):
74            self.assertIdentical(None, result)
75
76        d = self.protocol.ping(JID("example.com"))
77        d.addCallback(cb)
78
79        iq = self.stub.output[-1]
80
81        exc = StanzaError('service-unavailable')
82        response = exc.toResponse(iq)
83        self.stub.send(response)
84
85        return d
86
87
88    def test_pingStanzaError(self):
89        """
90        Pinging a service should errback a deferred on other (stanza) errors.
91        """
92        def cb(exc):
93            self.assertEquals('item-not-found', exc.condition)
94
95        d = self.protocol.ping(JID("example.com"))
96        self.assertFailure(d, StanzaError)
97        d.addCallback(cb)
98
99        iq = self.stub.output[-1]
100
101        exc = StanzaError('item-not-found')
102        response = exc.toResponse(iq)
103        self.stub.send(response)
104
105        return d
106
107
108
109class PingHandlerTest(unittest.TestCase):
110    """
111    Tests for L{ping.PingHandler}.
112    """
113
114    def setUp(self):
115        self.stub = XmlStreamStub()
116        self.protocol = ping.PingHandler()
117        self.protocol.xmlstream = self.stub.xmlstream
118        self.protocol.connectionInitialized()
119
120
121    def test_onPing(self):
122        """
123        A ping should have a simple result response.
124        """
125        xml = """<iq from='test@example.com' to='example.com' type='get'>
126                   <ping xmlns='urn:xmpp:ping'/>
127                 </iq>"""
128        self.stub.send(parseXml(xml))
129
130        response = self.stub.output[-1]
131        self.assertEquals('example.com', response.getAttribute('from'))
132        self.assertEquals('test@example.com', response.getAttribute('to'))
133        self.assertEquals('result', response.getAttribute('type'))
134
135
136    def test_interfaceIDisco(self):
137        """
138        The ping handler should provice Service Discovery information.
139        """
140        verify.verifyObject(iwokkel.IDisco, self.protocol)
141
142
143    def test_getDiscoInfo(self):
144        """
145        The ping namespace should be returned as a supported feature.
146        """
147        def cb(info):
148            discoInfo = disco.DiscoInfo()
149            for item in info:
150                discoInfo.append(item)
151            self.assertIn('urn:xmpp:ping', discoInfo.features)
152
153        d = defer.maybeDeferred(self.protocol.getDiscoInfo,
154                                JID('user@example.org/home'),
155                                JID('pubsub.example.org'),
156                                '')
157        d.addCallback(cb)
158        return d
159
160
161    def test_getDiscoInfoNode(self):
162        """
163        The ping namespace should not be returned for a node.
164        """
165        def cb(info):
166            discoInfo = disco.DiscoInfo()
167            for item in info:
168                discoInfo.append(item)
169            self.assertNotIn('urn:xmpp:ping', discoInfo.features)
170
171        d = defer.maybeDeferred(self.protocol.getDiscoInfo,
172                                JID('user@example.org/home'),
173                                JID('pubsub.example.org'),
174                                'test')
175        d.addCallback(cb)
176        return d
177
178
179    def test_getDiscoItems(self):
180        """
181        Items are not supported by this handler, so an empty list is expected.
182        """
183        def cb(items):
184            self.assertEquals(0, len(items))
185
186        d = defer.maybeDeferred(self.protocol.getDiscoItems,
187                                JID('user@example.org/home'),
188                                JID('pubsub.example.org'),
189                                '')
190        d.addCallback(cb)
191        return d
Note: See TracBrowser for help on using the repository browser.