''' Tests for smacks.py Stream Management ''' import unittest import lib lib.setup_env() 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 self.con.sm = smacks.Smacks(self.con) def tearDown(self): # Unplug if needed if hasattr(self.dispatcher, '_owner'): self.dispatcher.PlugOut() def _simulate_connect(self): self.dispatcher.PlugIn(self.client) # client is owner self.con.sm.set_owner(self.client) self.dispatcher.sm = self.con.sm # Simulate that we have established a connection self.dispatcher.StreamInit() self.dispatcher.ProcessNonBlocking("") self.dispatcher.ProcessNonBlocking(" " " ") self.con.sm.negociate() self.dispatcher.ProcessNonBlocking("") assert(self.con.sm.enabled) def _simulate_resume(self): self.con.sm.resume_request() # Resuming acknowledging 5 stanzas self.dispatcher.ProcessNonBlocking("") assert(self.con.sm.resuming) def _send(self, send, r, stanza): for i in range(r): send(stanza) def test_messages(self): message = 'Helloo ' iq = ''' ''' presence = ''' 24 In love Kakashi Sensei :P db4b7c52e39ba28562c74542d5988d47f09108a3 ''' self._simulate_connect() uqueue = self.con.sm.uqueue self.assertEqual(self.con.sm.out_h, 0) self.assertEqual(self.con.sm.in_h, 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(self.con.sm.in_h, 10) self.assertEqual(self.con.sm.out_h, 0) m = protocol.Message() # The client sends 10 stanzas for i in range(10): m = protocol.Message(body=str(i)) self.dispatcher.send(m) # Client sends 10 stanzas and put them in the queue self.assertEqual(self.con.sm.out_h, 10) self.assertEqual(len(uqueue), 10) # The server acknowledges that it recieved 5 stanzas self.dispatcher.ProcessNonBlocking("") # 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]) l.pop(0) def test_resumption(self): self._simulate_connect() m = protocol.Message() # The client sends 5 stanzas for i in range(5): m = protocol.Message(body=str(i)) self.dispatcher.send(m) self._simulate_resume() # No stanzas left self.assertEqual(len(self.con.sm.uqueue), 0) if __name__ == '__main__': unittest.main()