source: wokkel/test/test_ping.py @ 199:bb63666ccec7

Last change on this file since 199:bb63666ccec7 was 199:bb63666ccec7, checked in by Ralph Meijer <ralphm@…>, 4 years ago

imported patch py3-ping.patch

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