source: wokkel/test/test_ping.py @ 96:8e6130587088

Last change on this file since 96:8e6130587088 was 96:8e6130587088, checked in by Ralph Meijer <ralphm@…>, 10 years ago

Remove copyright dates from individual source files, only update LICENSE.

  • Property exe set to *
File size: 5.7 KB
Line 
1# Copyright (c) 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_onPingHandled(self):
137        """
138        The ping handler should mark the stanza as handled.
139        """
140        xml = """<iq from='test@example.com' to='example.com' type='get'>
141                   <ping xmlns='urn:xmpp:ping'/>
142                 </iq>"""
143        iq = parseXml(xml)
144        self.stub.send(iq)
145
146        self.assertTrue(iq.handled)
147
148
149    def test_interfaceIDisco(self):
150        """
151        The ping handler should provice Service Discovery information.
152        """
153        verify.verifyObject(iwokkel.IDisco, self.protocol)
154
155
156    def test_getDiscoInfo(self):
157        """
158        The ping namespace should be returned as a supported feature.
159        """
160        def cb(info):
161            discoInfo = disco.DiscoInfo()
162            for item in info:
163                discoInfo.append(item)
164            self.assertIn('urn:xmpp:ping', discoInfo.features)
165
166        d = defer.maybeDeferred(self.protocol.getDiscoInfo,
167                                JID('user@example.org/home'),
168                                JID('pubsub.example.org'),
169                                '')
170        d.addCallback(cb)
171        return d
172
173
174    def test_getDiscoInfoNode(self):
175        """
176        The ping namespace should not be returned for a node.
177        """
178        def cb(info):
179            discoInfo = disco.DiscoInfo()
180            for item in info:
181                discoInfo.append(item)
182            self.assertNotIn('urn:xmpp:ping', discoInfo.features)
183
184        d = defer.maybeDeferred(self.protocol.getDiscoInfo,
185                                JID('user@example.org/home'),
186                                JID('pubsub.example.org'),
187                                'test')
188        d.addCallback(cb)
189        return d
190
191
192    def test_getDiscoItems(self):
193        """
194        Items are not supported by this handler, so an empty list is expected.
195        """
196        def cb(items):
197            self.assertEquals(0, len(items))
198
199        d = defer.maybeDeferred(self.protocol.getDiscoItems,
200                                JID('user@example.org/home'),
201                                JID('pubsub.example.org'),
202                                '')
203        d.addCallback(cb)
204        return d
Note: See TracBrowser for help on using the repository browser.