Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(86)

Unified Diff: third_party/twisted_8_1/twisted/words/test/test_jabbercomponent.py

Issue 12261012: Remove third_party/twisted_8_1 (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/tools/build
Patch Set: Created 7 years, 10 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: third_party/twisted_8_1/twisted/words/test/test_jabbercomponent.py
diff --git a/third_party/twisted_8_1/twisted/words/test/test_jabbercomponent.py b/third_party/twisted_8_1/twisted/words/test/test_jabbercomponent.py
deleted file mode 100644
index ce0f17700e5fcba298490df88912f6eb8114df4e..0000000000000000000000000000000000000000
--- a/third_party/twisted_8_1/twisted/words/test/test_jabbercomponent.py
+++ /dev/null
@@ -1,137 +0,0 @@
-# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
-# See LICENSE for details.
-
-"""
-Tests for L{twisted.words.protocols.jabber.component}
-"""
-
-import sha
-from twisted.trial import unittest
-
-from twisted.words.protocols.jabber import component
-from twisted.words.protocols import jabber
-from twisted.words.protocols.jabber import xmlstream
-
-class DummyTransport:
- def __init__(self, list):
- self.list = list
-
- def write(self, bytes):
- self.list.append(bytes)
-
-class ComponentInitiatingInitializerTest(unittest.TestCase):
- def setUp(self):
- self.output = []
-
- self.authenticator = xmlstream.Authenticator()
- self.authenticator.password = 'secret'
- self.xmlstream = xmlstream.XmlStream(self.authenticator)
- self.xmlstream.namespace = 'test:component'
- self.xmlstream.send = self.output.append
- self.xmlstream.connectionMade()
- self.xmlstream.dataReceived(
- "<stream:stream xmlns='test:component' "
- "xmlns:stream='http://etherx.jabber.org/streams' "
- "from='example.com' id='12345' version='1.0'>")
- self.xmlstream.sid = '12345'
- self.init = component.ComponentInitiatingInitializer(self.xmlstream)
-
- def testHandshake(self):
- """
- Test basic operations of component handshake.
- """
-
- d = self.init.initialize()
-
- # the initializer should have sent the handshake request
-
- handshake = self.output[-1]
- self.assertEquals('handshake', handshake.name)
- self.assertEquals('test:component', handshake.uri)
- self.assertEquals(sha.new("%s%s" % ('12345', 'secret')).hexdigest(),
- unicode(handshake))
-
- # successful authentication
-
- handshake.children = []
- self.xmlstream.dataReceived(handshake.toXml())
-
- return d
-
-class ComponentAuthTest(unittest.TestCase):
- def authPassed(self, stream):
- self.authComplete = True
-
- def testAuth(self):
- self.authComplete = False
- outlist = []
-
- ca = component.ConnectComponentAuthenticator("cjid", "secret")
- xs = xmlstream.XmlStream(ca)
- xs.transport = DummyTransport(outlist)
-
- xs.addObserver(xmlstream.STREAM_AUTHD_EVENT,
- self.authPassed)
-
- # Go...
- xs.connectionMade()
- xs.dataReceived("<stream:stream xmlns='jabber:component:accept' xmlns:stream='http://etherx.jabber.org/streams' from='cjid' id='12345'>")
-
- # Calculate what we expect the handshake value to be
- hv = sha.new("%s%s" % ("12345", "secret")).hexdigest()
-
- self.assertEquals(outlist[1], "<handshake>%s</handshake>" % (hv))
-
- xs.dataReceived("<handshake/>")
-
- self.assertEquals(self.authComplete, True)
-
-
-class JabberServiceHarness(jabber.component.Service):
- def __init__(self):
- self.componentConnectedFlag = False
- self.componentDisconnectedFlag = False
- self.transportConnectedFlag = False
-
- def componentConnected(self, xmlstream):
- self.componentConnectedFlag = True
-
- def componentDisconnected(self):
- self.componentDisconnectedFlag = True
-
- def transportConnected(self, xmlstream):
- self.transportConnectedFlag = True
-
-
-class TestJabberServiceManager(unittest.TestCase):
- def testSM(self):
- # Setup service manager and test harnes
- sm = jabber.component.ServiceManager("foo", "password")
- svc = JabberServiceHarness()
- svc.setServiceParent(sm)
-
- # Create a write list
- wlist = []
-
- # Setup a XmlStream
- xs = sm.getFactory().buildProtocol(None)
- xs.transport = self
- xs.transport.write = wlist.append
-
- # Indicate that it's connected
- xs.connectionMade()
-
- # Ensure the test service harness got notified
- self.assertEquals(True, svc.transportConnectedFlag)
-
- # Jump ahead and pretend like the stream got auth'd
- xs.dispatch(xs, xmlstream.STREAM_AUTHD_EVENT)
-
- # Ensure the test service harness got notified
- self.assertEquals(True, svc.componentConnectedFlag)
-
- # Pretend to drop the connection
- xs.connectionLost(None)
-
- # Ensure the test service harness got notified
- self.assertEquals(True, svc.componentDisconnectedFlag)

Powered by Google App Engine
This is Rietveld 408576698