import unittest
from slixmpp.test import SlixTest
from slixmpp import JID
class TestMIXPAM(SlixTest):
def setUp(self):
self.stream_start(plugins=['xep_0405'])
def tearDown(self):
self.stream_close()
def testGetRosterEmpty(self):
"""Test requesting an empty annotated roster"""
fut = self.xmpp.wrap(self.xmpp['xep_0405'].get_mix_roster())
self.wait_()
self.send("""
""")
self.recv("""
""")
result = self.run_coro(fut)
def testGetRoster(self):
"""Test requesting an annotated roster"""
fut = self.xmpp.wrap(self.xmpp['xep_0405'].get_mix_roster())
self.wait_()
self.send("""
""")
self.recv("""
-
""")
self.wait_()
contacts, channels = fut.result()
self.assertEqual(len(contacts), 2)
self.assertEqual(contacts[0]['jid'], 'romeo@example.net')
self.assertEqual(contacts[1]['jid'], 'juliet@example.net')
self.assertEqual(len(channels), 1)
self.assertEqual(channels[0]['jid'], 'balcony@example.net')
self.assertEqual(
channels[0]['channel']['participant-id'],
'123456'
)
def testClientJoin(self):
"""Test a client join"""
fut = self.xmpp.wrap(self.xmpp['xep_0405'].join_channel(
JID('coven@mix.shakespeare.example'),
'toto',
))
self.send("""
toto
""")
self.recv("""
""")
self.wait_()
self.assertEqual(fut.result(), set())
def testClientJoinNotAllNodes(self):
"""Test a client join where one of the nodes is rejected"""
fut = self.xmpp.wrap(self.xmpp['xep_0405'].join_channel(
JID('coven@mix.shakespeare.example'),
'toto',
))
self.send("""
toto
""")
self.recv("""
""")
self.wait_()
self.assertEqual(fut.result(), {'urn:xmpp:mix:nodes:info'})
def testClientLeave(self):
"""Test a client leave"""
fut = self.xmpp.wrap(self.xmpp['xep_0405'].leave_channel(
JID('coven@mix.shakespeare.example'),
))
self.send("""
""")
self.recv("""
""")
self.assertEqual(fut.done(), True)
self.assertEqual(fut.exception(), None)
suite = unittest.TestLoader().loadTestsFromTestCase(TestMIXPAM)