/usr/share/pyshared/wokkel/test/test_client.py is in python-wokkel 0.7.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | # Copyright (c) Ralph Meijer.
# See LICENSE for details.
"""
Tests for L{wokkel.client}.
"""
from twisted.internet import defer
from twisted.trial import unittest
from twisted.words.protocols.jabber import xmlstream
from twisted.words.protocols.jabber.client import XMPPAuthenticator
from twisted.words.protocols.jabber.jid import JID
from twisted.words.protocols.jabber.xmlstream import STREAM_AUTHD_EVENT
from twisted.words.protocols.jabber.xmlstream import INIT_FAILED_EVENT
from twisted.words.protocols.jabber.xmlstream import XMPPHandler
from wokkel import client
class XMPPClientTest(unittest.TestCase):
"""
Tests for L{client.XMPPClient}.
"""
def setUp(self):
self.client = client.XMPPClient(JID('user@example.org'), 'secret')
def test_jid(self):
"""
Make sure the JID we pass is stored on the client.
"""
self.assertEquals(JID('user@example.org'), self.client.jid)
def test_jidWhenInitialized(self):
"""
Make sure that upon login, the JID is updated from the authenticator.
"""
xs = self.client.factory.buildProtocol(None)
self.client.factory.authenticator.jid = JID('user@example.org/test')
xs.dispatch(xs, xmlstream.STREAM_AUTHD_EVENT)
self.assertEquals(JID('user@example.org/test'), self.client.jid)
class DeferredClientFactoryTest(unittest.TestCase):
"""
Tests for L{client.DeferredClientFactory}.
"""
def setUp(self):
self.factory = client.DeferredClientFactory(JID('user@example.org'),
'secret')
def test_buildProtocol(self):
"""
The authenticator factory should be passed to its protocol and it
should instantiate the authenticator and save it.
L{xmlstream.XmlStream}s do that, but we also want to ensure it really
is one.
"""
xs = self.factory.buildProtocol(None)
self.assertIdentical(self.factory, xs.factory)
self.assertIsInstance(xs, xmlstream.XmlStream)
self.assertIsInstance(xs.authenticator, XMPPAuthenticator)
def test_deferredOnInitialized(self):
"""
Test the factory's deferred on stream initialization.
"""
xs = self.factory.buildProtocol(None)
xs.dispatch(xs, STREAM_AUTHD_EVENT)
return self.factory.deferred
def test_deferredOnNotInitialized(self):
"""
Test the factory's deferred on failed stream initialization.
"""
class TestException(Exception):
pass
xs = self.factory.buildProtocol(None)
xs.dispatch(TestException(), INIT_FAILED_EVENT)
self.assertFailure(self.factory.deferred, TestException)
return self.factory.deferred
def test_deferredOnConnectionFailure(self):
"""
Test the factory's deferred on connection faulure.
"""
class TestException(Exception):
pass
self.factory.buildProtocol(None)
self.factory.clientConnectionFailed(self, TestException())
self.assertFailure(self.factory.deferred, TestException)
return self.factory.deferred
def test_addHandler(self):
"""
Test the addition of a protocol handler.
"""
handler = XMPPHandler()
handler.setHandlerParent(self.factory.streamManager)
self.assertIn(handler, self.factory.streamManager)
self.assertIdentical(self.factory.streamManager, handler.parent)
def test_removeHandler(self):
"""
Test removal of a protocol handler.
"""
handler = XMPPHandler()
handler.setHandlerParent(self.factory.streamManager)
handler.disownHandlerParent(self.factory.streamManager)
self.assertNotIn(handler, self.factory.streamManager)
self.assertIdentical(None, handler.parent)
class ClientCreatorTest(unittest.TestCase):
"""
Tests for L{client.clientCreator}.
"""
def test_call(self):
"""
The factory is passed to an SRVConnector and a connection initiated.
"""
d1 = defer.Deferred()
factory = client.DeferredClientFactory(JID('user@example.org'),
'secret')
def cb(connector):
self.assertEqual('xmpp-client', connector.service)
self.assertEqual('example.org', connector.domain)
self.assertEqual(factory, connector.factory)
def connect(connector):
d1.callback(connector)
d1.addCallback(cb)
self.patch(client.SRVConnector, 'connect', connect)
d2 = client.clientCreator(factory)
self.assertEqual(factory.deferred, d2)
return d1
|