source: wokkel/test/test_muc.py @ 108:50e3c368f2f3

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

add basic errors and disco stuff, still need iq and message parts re #24

File size: 4.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
61    def test_discoServerSupport(self):
62        """Test for disco support from a server.
63        """
64        test_srv = 'shakespeare.lit'
65
66        def cb(query):
67            # check namespace
68            self.failUnless(query.uri==disco.NS_INFO, 'Wrong namespace')
69           
70
71        d = self.protocol.disco(test_srv)
72        d.addCallback(cb)
73
74        iq = self.stub.output[-1]
75       
76        # send back a response
77        response = toResponse(iq, 'result')
78        response.addElement('query', disco.NS_INFO)
79        # need to add information to response
80        response.query.addChild(disco.DiscoFeature(muc.NS))
81        response.query.addChild(disco.DiscoIdentity(category='conference',
82                                                    name='Macbeth Chat Service',
83                                                    type='text'))
84       
85        self.stub.send(response)
86        return d
87       
88
89       
90    def test_joinRoom(self):
91        """Test joining a room
92        """
93        test_room = 'test'
94        test_srv  = 'conference.example.org'
95        test_nick = 'Nick'
96
97        def cb(room):
98            self.assertEquals(test_room, room.name)
99
100        d = self.protocol.join(test_srv, test_room, test_nick)
101        d.addCallback(cb)
102
103        prs = self.stub.output[-1]
104        self.failUnless(prs.name=='presence', "Need to be presence")
105        self.failUnless(getattr(prs, 'x', None), 'No muc x element')
106
107        # send back user presence, they joined       
108        response = muc.UserPresence(frm=test_room+'@'+test_srv+'/'+test_nick)
109        self.stub.send(response)
110        return d
111
112
113    def test_joinRoomForbidden(self):
114        """Test joining a room and getting an error
115        """
116        test_room = 'test'
117        test_srv  = 'conference.example.org'
118        test_nick = 'Nick'
119
120        # p = muc.BasicPresenc(to=)
121
122        def cb(error):
123            self.failUnless(isinstance(error.value,muc.PresenceError), 'Wrong type')
124            self.failUnless(error.value['type']=='error', 'Not an error returned')
125           
126           
127        d = self.protocol.join(test_srv, test_room, test_nick)
128        d.addBoth(cb)
129
130        prs = self.stub.output[-1]
131        self.failUnless(prs.name=='presence', "Need to be presence")
132        self.failUnless(getattr(prs, 'x', None), 'No muc x element')
133        # send back user presence, they joined
134       
135        response = muc.PresenceError(error=muc.MUCError('auth',
136                                                        'forbidden'
137                                                        ),
138                                     frm=test_room+'@'+test_srv+'/'+test_nick)
139        self.stub.send(response)
140        return d       
Note: See TracBrowser for help on using the repository browser.