| Index: third_party/twisted_8_1/twisted/words/test/test_jabberclient.py
|
| diff --git a/third_party/twisted_8_1/twisted/words/test/test_jabberclient.py b/third_party/twisted_8_1/twisted/words/test/test_jabberclient.py
|
| deleted file mode 100644
|
| index 3da3788bf9ea0451861429a31a12e7de99537316..0000000000000000000000000000000000000000
|
| --- a/third_party/twisted_8_1/twisted/words/test/test_jabberclient.py
|
| +++ /dev/null
|
| @@ -1,272 +0,0 @@
|
| -# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
|
| -# See LICENSE for details.
|
| -
|
| -"""
|
| -Tests for L{twisted.words.protocols.jabber.client}
|
| -"""
|
| -
|
| -import sha
|
| -from twisted.trial import unittest
|
| -from twisted.words.protocols.jabber import client, error, jid, xmlstream
|
| -from twisted.words.protocols.jabber.sasl import SASLInitiatingInitializer
|
| -
|
| -class CheckVersionInitializerTest(unittest.TestCase):
|
| - def setUp(self):
|
| - a = xmlstream.Authenticator()
|
| - xs = xmlstream.XmlStream(a)
|
| - self.init = client.CheckVersionInitializer(xs)
|
| -
|
| - def testSupported(self):
|
| - """
|
| - Test supported version number 1.0
|
| - """
|
| - self.init.xmlstream.version = (1, 0)
|
| - self.init.initialize()
|
| -
|
| - def testNotSupported(self):
|
| - """
|
| - Test unsupported version number 0.0, and check exception.
|
| - """
|
| - self.init.xmlstream.version = (0, 0)
|
| - exc = self.assertRaises(error.StreamError, self.init.initialize)
|
| - self.assertEquals('unsupported-version', exc.condition)
|
| -
|
| -
|
| -class InitiatingInitializerHarness(object):
|
| - def setUp(self):
|
| - self.output = []
|
| - self.authenticator = xmlstream.ConnectAuthenticator('example.org')
|
| - self.xmlstream = xmlstream.XmlStream(self.authenticator)
|
| - self.xmlstream.send = self.output.append
|
| - self.xmlstream.connectionMade()
|
| - self.xmlstream.dataReceived("<stream:stream xmlns='jabber:client' "
|
| - "xmlns:stream='http://etherx.jabber.org/streams' "
|
| - "from='example.com' id='12345' version='1.0'>")
|
| -
|
| -
|
| -class IQAuthInitializerTest(InitiatingInitializerHarness, unittest.TestCase):
|
| - def setUp(self):
|
| - super(IQAuthInitializerTest, self).setUp()
|
| - self.init = client.IQAuthInitializer(self.xmlstream)
|
| - self.authenticator.jid = jid.JID('user@example.com/resource')
|
| - self.authenticator.password = 'secret'
|
| -
|
| - def testBasic(self):
|
| - """
|
| - Test basic operations with plain text authentication.
|
| -
|
| - Set up a stream, and act as if authentication succeeds.
|
| - """
|
| - d = self.init.initialize()
|
| -
|
| - # The initializer should have sent query to find out auth methods.
|
| -
|
| - # Examine query.
|
| - iq = self.output[-1]
|
| - self.assertEquals('iq', iq.name)
|
| - self.assertEquals('get', iq['type'])
|
| - self.assertEquals(('jabber:iq:auth', 'query'),
|
| - (iq.children[0].uri, iq.children[0].name))
|
| -
|
| - # Send server response
|
| - iq['type'] = 'result'
|
| - iq.query.addElement('username')
|
| - iq.query.addElement('password')
|
| - iq.query.addElement('resource')
|
| - self.xmlstream.dataReceived(iq.toXml())
|
| -
|
| - # Upon receiving the response, the initializer can start authentication
|
| -
|
| - iq = self.output[-1]
|
| - self.assertEquals('iq', iq.name)
|
| - self.assertEquals('set', iq['type'])
|
| - self.assertEquals(('jabber:iq:auth', 'query'),
|
| - (iq.children[0].uri, iq.children[0].name))
|
| - self.assertEquals('user', unicode(iq.query.username))
|
| - self.assertEquals('secret', unicode(iq.query.password))
|
| - self.assertEquals('resource', unicode(iq.query.resource))
|
| -
|
| - # Send server response
|
| - self.xmlstream.dataReceived("<iq type='result' id='%s'/>" % iq['id'])
|
| - return d
|
| -
|
| - def testDigest(self):
|
| - """
|
| - Test digest authentication.
|
| -
|
| - Set up a stream, and act as if authentication succeeds.
|
| - """
|
| - d = self.init.initialize()
|
| -
|
| - # The initializer should have sent query to find out auth methods.
|
| -
|
| - iq = self.output[-1]
|
| -
|
| - # Send server response
|
| - iq['type'] = 'result'
|
| - iq.query.addElement('username')
|
| - iq.query.addElement('digest')
|
| - iq.query.addElement('resource')
|
| - self.xmlstream.dataReceived(iq.toXml())
|
| -
|
| - # Upon receiving the response, the initializer can start authentication
|
| -
|
| - iq = self.output[-1]
|
| - self.assertEquals('iq', iq.name)
|
| - self.assertEquals('set', iq['type'])
|
| - self.assertEquals(('jabber:iq:auth', 'query'),
|
| - (iq.children[0].uri, iq.children[0].name))
|
| - self.assertEquals('user', unicode(iq.query.username))
|
| - self.assertEquals(sha.new('12345secret').hexdigest(),
|
| - unicode(iq.query.digest))
|
| - self.assertEquals('resource', unicode(iq.query.resource))
|
| -
|
| - # Send server response
|
| - self.xmlstream.dataReceived("<iq type='result' id='%s'/>" % iq['id'])
|
| - return d
|
| -
|
| - def testFailRequestFields(self):
|
| - """
|
| - Test failure of request for fields.
|
| - """
|
| - d = self.init.initialize()
|
| - iq = self.output[-1]
|
| - response = error.StanzaError('not-authorized').toResponse(iq)
|
| - self.xmlstream.dataReceived(response.toXml())
|
| - self.assertFailure(d, error.StanzaError)
|
| - return d
|
| -
|
| - def testFailAuth(self):
|
| - """
|
| - Test failure of request for fields.
|
| - """
|
| - d = self.init.initialize()
|
| - iq = self.output[-1]
|
| - iq['type'] = 'result'
|
| - iq.query.addElement('username')
|
| - iq.query.addElement('password')
|
| - iq.query.addElement('resource')
|
| - self.xmlstream.dataReceived(iq.toXml())
|
| - iq = self.output[-1]
|
| - response = error.StanzaError('not-authorized').toResponse(iq)
|
| - self.xmlstream.dataReceived(response.toXml())
|
| - self.assertFailure(d, error.StanzaError)
|
| - return d
|
| -
|
| -
|
| -class BindInitializerTest(InitiatingInitializerHarness, unittest.TestCase):
|
| - def setUp(self):
|
| - super(BindInitializerTest, self).setUp()
|
| - self.init = client.BindInitializer(self.xmlstream)
|
| - self.authenticator.jid = jid.JID('user@example.com/resource')
|
| -
|
| - def testBasic(self):
|
| - """
|
| - Test basic operations.
|
| -
|
| - Set up a stream, and act as if resource binding succeeds.
|
| - """
|
| - def cb(result):
|
| - self.assertEquals(jid.JID('user@example.com/other resource'),
|
| - self.authenticator.jid)
|
| -
|
| - d = self.init.start().addCallback(cb)
|
| - iq = self.output[-1]
|
| - self.assertEquals('iq', iq.name)
|
| - self.assertEquals('set', iq['type'])
|
| - self.assertEquals(('urn:ietf:params:xml:ns:xmpp-bind', 'bind'),
|
| - (iq.children[0].uri, iq.children[0].name))
|
| - iq['type'] = 'result'
|
| - iq.bind.children = []
|
| - iq.bind.addElement('jid', content='user@example.com/other resource')
|
| - self.xmlstream.dataReceived(iq.toXml())
|
| - return d
|
| -
|
| - def testFailure(self):
|
| - """
|
| - Test basic operations.
|
| -
|
| - Set up a stream, and act as if resource binding fails.
|
| - """
|
| - def cb(result):
|
| - self.assertEquals(jid.JID('user@example.com/resource'),
|
| - self.authenticator.jid)
|
| -
|
| - d = self.init.start()
|
| - id = self.output[-1]['id']
|
| - self.xmlstream.dataReceived("<iq type='error' id='%s'/>" % id)
|
| - self.assertFailure(d, error.StanzaError)
|
| - return d
|
| -
|
| -
|
| -class SessionInitializerTest(InitiatingInitializerHarness, unittest.TestCase):
|
| -
|
| - def setUp(self):
|
| - super(SessionInitializerTest, self).setUp()
|
| - self.init = client.SessionInitializer(self.xmlstream)
|
| -
|
| - def testSuccess(self):
|
| - """
|
| - Test basic operations.
|
| -
|
| - Set up a stream, and act as if resource binding succeeds.
|
| - """
|
| - d = self.init.start()
|
| - iq = self.output[-1]
|
| - self.assertEquals('iq', iq.name)
|
| - self.assertEquals('set', iq['type'])
|
| - self.assertEquals(('urn:ietf:params:xml:ns:xmpp-session', 'session'),
|
| - (iq.children[0].uri, iq.children[0].name))
|
| - self.xmlstream.dataReceived("<iq type='result' id='%s'/>" % iq['id'])
|
| - return d
|
| -
|
| - def testFailure(self):
|
| - """
|
| - Test basic operations.
|
| -
|
| - Set up a stream, and act as if session establishment succeeds.
|
| - """
|
| - d = self.init.start()
|
| - id = self.output[-1]['id']
|
| - self.xmlstream.dataReceived("<iq type='error' id='%s'/>" % id)
|
| - self.assertFailure(d, error.StanzaError)
|
| - return d
|
| -
|
| -
|
| -class XMPPAuthenticatorTest(unittest.TestCase):
|
| - """
|
| - Test for both XMPPAuthenticator and XMPPClientFactory.
|
| - """
|
| - def testBasic(self):
|
| - """
|
| - Test basic operations.
|
| -
|
| - Setup an XMPPClientFactory, which sets up an XMPPAuthenticator, and let
|
| - it produce a protocol instance. Then inspect the instance variables of
|
| - the authenticator and XML stream objects.
|
| - """
|
| - self.client_jid = jid.JID('user@example.com/resource')
|
| -
|
| - # Get an XmlStream instance. Note that it gets initialized with the
|
| - # XMPPAuthenticator (that has its associateWithXmlStream called) that
|
| - # is in turn initialized with the arguments to the factory.
|
| - xs = client.XMPPClientFactory(self.client_jid,
|
| - 'secret').buildProtocol(None)
|
| -
|
| - # test authenticator's instance variables
|
| - self.assertEqual('example.com', xs.authenticator.otherHost)
|
| - self.assertEqual(self.client_jid, xs.authenticator.jid)
|
| - self.assertEqual('secret', xs.authenticator.password)
|
| -
|
| - # test list of initializers
|
| - version, tls, sasl, bind, session = xs.initializers
|
| -
|
| - self.assert_(isinstance(tls, xmlstream.TLSInitiatingInitializer))
|
| - self.assert_(isinstance(sasl, SASLInitiatingInitializer))
|
| - self.assert_(isinstance(bind, client.BindInitializer))
|
| - self.assert_(isinstance(session, client.SessionInitializer))
|
| -
|
| - self.assertFalse(tls.required)
|
| - self.assertTrue(sasl.required)
|
| - self.assertFalse(bind.required)
|
| - self.assertFalse(session.required)
|
|
|