Commit e02088e9 authored by zimio's avatar zimio

added tests for

parent b8aa5185
Tests for Stream Management
import unittest
import lib
from mock import Mock
from common.xmpp import dispatcher_nb
from common.xmpp import protocol
from common.xmpp import smacks
class TestDispatcherNB(unittest.TestCase):
Test class for NonBlocking dispatcher. Tested dispatcher will be plugged
into a mock client
def setUp(self):
self.dispatcher = dispatcher_nb.XMPPDispatcher()
# Setup mock client
self.client = Mock()
self.client.__str__ = lambda: 'Mock' # FIXME: why do I need this one?
self.client._caller = Mock()
self.client.defaultNamespace = protocol.NS_CLIENT
self.client.Connection = Mock() # mock transport
self.con = self.client.Connection = smacks.Smacks(self.con)
def tearDown(self):
# Unplug if needed
if hasattr(self.dispatcher, '_owner'):
def _simulate_connect(self):
self.dispatcher.PlugIn(self.client) # client is owner =
# Simulate that we have established a connection
self.dispatcher.ProcessNonBlocking("<stream:stream xmlns:stream='' xmlns='jabber:client'>")
self.dispatcher.ProcessNonBlocking("<stream:features> <sm xmlns='urn:xmpp:sm:2'> <optional/> </sm> </stream:features>")
self.dispatcher.ProcessNonBlocking("<enabled xmlns='urn:xmpp:sm:2' id='some-long-sm-id' resume='true'/>")
def _simulate_resume(self):
# Resuming acknowledging 5 stanzas
self.dispatcher.ProcessNonBlocking("<resumed xmlns='urn:xmpp:sm:2' id='some-long-sm-id' h='5'/>")
def _send(self, send, r, stanza):
for i in range(r):
def test_messages(self):
message = '<message><body>Helloo </body></message>'
iq = '''<iq from='' to='' type='error' id='18'>
<query xmlns=''/>
<error code='403' type='auth'>
<forbidden xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/>
presence = '''<presence from='' to=''>
<c node="" ver="" ext="share-v1 voice-v1" xmlns=""/>
<x stamp="20110614T23:17:51" xmlns="jabber:x:delay"/>
<status>In love Kakashi Sensei :P</status>
<x xmlns="vcard-temp:x:update">
</presence> '''
uqueue =
self.assertEqual(, 0)
self.assertEqual(, 0)
# The server sends 10 stanzas
self._send(self.dispatcher.ProcessNonBlocking, 5, message)
self._send(self.dispatcher.ProcessNonBlocking, 4, iq)
self._send(self.dispatcher.ProcessNonBlocking, 1, presence)
# The client has recieved 10 stanzas and sent none
self.assertEqual(, 10)
self.assertEqual(, 0)
m = protocol.Message()
# The client sends 10 stanzas
for i in range(10):
m = protocol.Message(body=str(i))
# Client sends 10 stanzas and put them in the queue
self.assertEqual(, 10)
self.assertEqual(len(uqueue), 10)
# The server acknowledges that it recieved 5 stanzas
self.dispatcher.ProcessNonBlocking("<a xmlns='urn:xmpp:sm:2' h='5'/>")
# 5 stanzas are removed from the queue, only 5 stanzas are left
self.assertEqual(len(uqueue), 5)
# Check for the right order of stanzas in the queue
l = ['5', '6', '7', '8', '9']
for i in uqueue:
self.assertEqual(i.getBody(), l[0])
def test_resumption(self):
m = protocol.Message()
# The client sends 5 stanzas
for i in range(5):
m = protocol.Message(body=str(i))
# No stanzas left
self.assertEqual(len(, 0)
if __name__ == '__main__':
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment