← Back to branch summary

~certify-web-dev/twisted/certify-trunk

« back to all changes in this revision

Viewing changes to twisted/words/test/test_jabbercomponent.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2007-01-17 14:52:35 UTC
  • mto: (2.2.3 sid)
  • mto: This revision was merged to the branch mainline in revision 15.
  • Revision ID: james.westby@ubuntu.com-20070117145235-7gaj253qxi5wiq16
Tags: upstream-2.5.0
Import upstream version 2.5.0

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/words/test/test_jabbercomponent.py
 
1
# Copyright (c) 2001-2005 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
 
 
5
import sys, os, sha
 
6
from twisted.trial import unittest
 
7
 
 
8
from twisted.words.protocols.jabber.component import ConnectComponentAuthenticator, ComponentInitiatingInitializer
 
9
from twisted.words.protocols import jabber
 
10
from twisted.words.protocols.jabber import xmlstream
 
11
 
 
12
class DummyTransport:
 
13
    def __init__(self, list):
 
14
        self.list = list
 
15
 
 
16
    def write(self, bytes):
 
17
        self.list.append(bytes)
 
18
 
 
19
class ComponentInitiatingInitializerTest(unittest.TestCase):
 
20
    def setUp(self):
 
21
        self.output = []
 
22
 
 
23
        self.authenticator = xmlstream.Authenticator()
 
24
        self.authenticator.password = 'secret'
 
25
        self.xmlstream = xmlstream.XmlStream(self.authenticator)
 
26
        self.xmlstream.namespace = 'test:component'
 
27
        self.xmlstream.send = self.output.append
 
28
        self.xmlstream.connectionMade()
 
29
        self.xmlstream.dataReceived(
 
30
                "<stream:stream xmlns='test:component' "
 
31
                "xmlns:stream='http://etherx.jabber.org/streams' "
 
32
                "from='example.com' id='12345' version='1.0'>")
 
33
        self.init = ComponentInitiatingInitializer(self.xmlstream)
 
34
 
 
35
    def testHandshake(self):
 
36
        """
 
37
        Test basic operations of component handshake.
 
38
        """
 
39
 
 
40
        d = self.init.initialize()
 
41
 
 
42
        # the initializer should have sent the handshake request
 
43
 
 
44
        handshake = self.output[0]
 
45
        self.assertEquals('handshake', handshake.name)
 
46
        self.assertEquals('test:component', handshake.uri)
 
47
        self.assertEquals(sha.new("%s%s" % ('12345', 'secret')).hexdigest(),
 
48
                          unicode(handshake))
 
49
 
 
50
        # successful authentication
 
51
 
 
52
        handshake.children = []
 
53
        self.xmlstream.dataReceived(handshake.toXml())
 
54
 
 
55
        return d
 
56
 
 
57
class ComponentAuthTest(unittest.TestCase):
 
58
    def authPassed(self, stream):
 
59
        self.authComplete = True
 
60
        
 
61
    def testAuth(self):
 
62
        self.authComplete = False
 
63
        outlist = []
 
64
 
 
65
        ca = ConnectComponentAuthenticator("cjid", "secret")
 
66
        xs = xmlstream.XmlStream(ca)
 
67
        xs.transport = DummyTransport(outlist)
 
68
 
 
69
        xs.addObserver(xmlstream.STREAM_AUTHD_EVENT,
 
70
                       self.authPassed)
 
71
 
 
72
        # Go...
 
73
        xs.connectionMade()
 
74
        xs.dataReceived("<stream:stream xmlns='jabber:component:accept' xmlns:stream='http://etherx.jabber.org/streams' from='cjid' id='12345'>")
 
75
 
 
76
        # Calculate what we expect the handshake value to be
 
77
        hv = sha.new("%s%s" % ("12345", "secret")).hexdigest()
 
78
 
 
79
        self.assertEquals(outlist[1], "<handshake>%s</handshake>" % (hv))
 
80
 
 
81
        xs.dataReceived("<handshake/>")
 
82
 
 
83
        self.assertEquals(self.authComplete, True)
 
84
        
 
85
 
 
86
class JabberServiceHarness(jabber.component.Service):
 
87
    def __init__(self):
 
88
        self.componentConnectedFlag = False
 
89
        self.componentDisconnectedFlag = False
 
90
        self.transportConnectedFlag = False
 
91
        
 
92
    def componentConnected(self, xmlstream):
 
93
        self.componentConnectedFlag = True
 
94
 
 
95
    def componentDisconnected(self):
 
96
        self.componentDisconnectedFlag = True
 
97
        
 
98
    def transportConnected(self, xmlstream):
 
99
        self.transportConnectedFlag = True
 
100
 
 
101
 
 
102
class TestJabberServiceManager(unittest.TestCase):
 
103
    def testSM(self):
 
104
        # Setup service manager and test harnes
 
105
        sm = jabber.component.ServiceManager("foo", "password")
 
106
        svc = JabberServiceHarness()
 
107
        svc.setServiceParent(sm)
 
108
 
 
109
        # Create a write list
 
110
        wlist = []
 
111
 
 
112
        # Setup a XmlStream
 
113
        xs = sm.getFactory().buildProtocol(None)
 
114
        xs.transport = self
 
115
        xs.transport.write = wlist.append
 
116
 
 
117
        # Indicate that it's connected
 
118
        xs.connectionMade()
 
119
 
 
120
        # Ensure the test service harness got notified
 
121
        self.assertEquals(True, svc.transportConnectedFlag)
 
122
 
 
123
        # Jump ahead and pretend like the stream got auth'd
 
124
        xs.dispatch(xs, xmlstream.STREAM_AUTHD_EVENT)
 
125
 
 
126
        # Ensure the test service harness got notified
 
127
        self.assertEquals(True, svc.componentConnectedFlag)
 
128
 
 
129
        # Pretend to drop the connection
 
130
        xs.connectionLost(None)
 
131
 
 
132
        # Ensure the test service harness got notified
 
133
        self.assertEquals(True, svc.componentDisconnectedFlag)