source: wokkel/test/test_muc.py @ 109:ee2752dca3b2

wokkel-muc-client-support-24
Last change on this file since 109:ee2752dca3b2 was 109:ee2752dca3b2, checked in by Christopher Zorn <tofu@…>, 13 years ago

add room configuration, groupchat and some clean up, still needs documentation and a bit more stuff to add re #24

File size: 5.0 KB
Line 
1# Copyright (c) 2003-2008 Ralph Meijer
2# See LICENSE for details.
3
4"""
5Tests for L{wokkel.muc}
6"""
7
8from zope.interface import verify
9
10from twisted.trial import unittest
11from twisted.internet import defer
12from twisted.words.xish import domish, xpath
13from twisted.words.protocols.jabber import error
14from twisted.words.protocols.jabber.jid import JID
15
16from wokkel import data_form, iwokkel, muc, shim, disco
17from wokkel.generic import parseXml
18from wokkel.test.helpers import XmlStreamStub
19
20try:
21    from twisted.words.protocols.jabber.xmlstream import toResponse
22except ImportError:
23    from wokkel.compat import toResponse
24
25
26def calledAsync(fn):
27    """
28    Function wrapper that fires a deferred upon calling the given function.
29    """
30    d = defer.Deferred()
31
32    def func(*args, **kwargs):
33        try:
34            result = fn(*args, **kwargs)
35        except:
36            d.errback()
37        else:
38            d.callback(result)
39
40    return d, func
41
42
43class MucClientTest(unittest.TestCase):
44    timeout = 2
45
46    def setUp(self):
47        self.stub = XmlStreamStub()
48        self.protocol = muc.MUCClient()
49        self.protocol.xmlstream = self.stub.xmlstream
50        self.protocol.connectionInitialized()
51
52
53    def test_interface(self):
54        """
55        Do instances of L{muc.MUCClient} provide L{iwokkel.IMUCClient}?
56        """
57        verify.verifyObject(iwokkel.IMUCClient, self.protocol)
58
59
60    def test_presence(self):
61        """Test receiving room presence
62        """
63        p = muc.UserPresence()
64       
65        def userPresence(prs):
66            self.failUnless(len(prs.children)==1, 'Not enough children')
67            self.failUnless(getattr(prs,'x',None), 'No x element')
68                   
69       
70        d, self.protocol.receivedUserPresence = calledAsync(userPresence)
71        self.stub.send(p)
72        return d
73
74
75    def test_groupChat(self):
76        """Test receiving room presence
77        """
78        m = muc.GroupChat('test@test.com',body='test')
79       
80        def groupChat(elem):
81            self.failUnless(elem.name=='message','Wrong stanza')
82            self.failUnless(elem['type'] == 'groupchat', 'Wrong attribute')
83                           
84       
85        d, self.protocol.receivedGroupChat = calledAsync(groupChat)
86        self.stub.send(m)
87        return d
88
89
90    def test_discoServerSupport(self):
91        """Test for disco support from a server.
92        """
93        test_srv = 'shakespeare.lit'
94
95        def cb(query):
96            # check namespace
97            self.failUnless(query.uri==disco.NS_INFO, 'Wrong namespace')
98           
99
100        d = self.protocol.disco(test_srv)
101        d.addCallback(cb)
102
103        iq = self.stub.output[-1]
104       
105        # send back a response
106        response = toResponse(iq, 'result')
107        response.addElement('query', disco.NS_INFO)
108        # need to add information to response
109        response.query.addChild(disco.DiscoFeature(muc.NS))
110        response.query.addChild(disco.DiscoIdentity(category='conference',
111                                                    name='Macbeth Chat Service',
112                                                    type='text'))
113       
114        self.stub.send(response)
115        return d
116       
117
118       
119    def test_joinRoom(self):
120        """Test joining a room
121        """
122        test_room = 'test'
123        test_srv  = 'conference.example.org'
124        test_nick = 'Nick'
125
126        def cb(room):
127            self.assertEquals(test_room, room.name)
128
129        d = self.protocol.join(test_srv, test_room, test_nick)
130        d.addCallback(cb)
131
132        prs = self.stub.output[-1]
133        self.failUnless(prs.name=='presence', "Need to be presence")
134        self.failUnless(getattr(prs, 'x', None), 'No muc x element')
135
136        # send back user presence, they joined       
137        response = muc.UserPresence(frm=test_room+'@'+test_srv+'/'+test_nick)
138        self.stub.send(response)
139        return d
140
141
142    def test_joinRoomForbidden(self):
143        """Test joining a room and getting an error
144        """
145        test_room = 'test'
146        test_srv  = 'conference.example.org'
147        test_nick = 'Nick'
148
149        def cb(error):
150            self.failUnless(isinstance(error.value,muc.PresenceError), 'Wrong type')
151            self.failUnless(error.value['type']=='error', 'Not an error returned')
152           
153           
154        d = self.protocol.join(test_srv, test_room, test_nick)
155        d.addBoth(cb)
156
157        prs = self.stub.output[-1]
158        self.failUnless(prs.name=='presence', "Need to be presence")
159        self.failUnless(getattr(prs, 'x', None), 'No muc x element')
160        # send back user presence, they joined
161       
162        response = muc.PresenceError(error=muc.MUCError('auth',
163                                                        'forbidden'
164                                                        ),
165                                     frm=test_room+'@'+test_srv+'/'+test_nick)
166        self.stub.send(response)
167        return d       
168
169    def test_roomConfigure(self):
170
171        test_room = 'test'
172        test_srv  = 'conference.example.org'
173        test_nick = 'Nick'       
174
175        self.fail('Not Implemented')
176       
177
Note: See TracBrowser for help on using the repository browser.