test/integration/test_xmpp_client_nb.py

   1 '''
   2 Testing script for NonBlockingClient class (src/common/xmpp/client_nb.py)
   3
   4 It actually connects to a xmpp server.
   5 '''
   6
   7 import unittest
   8
   9 import lib
  10 lib.setup_env()
  11
  12 from xmpp_mocks import MockConnection, IdleQueueThread
  13 from mock import Mock
  14 from common.xmpp import client_nb
  15
  16 #import logging
  17 #log = logging.getLogger('gajim')
  18 #log.setLevel(logging.DEBUG)
  19
  20 # (XMPP server hostname, c2s port). Script will connect to the machine.
  21 xmpp_server_port = ('gajim.org', 5222)
  22
  23 # [username, password, passphrase]. Script will authenticate to server above
  24 credentials = ['unittest', 'testtest', 'res']
  25
  26 class TestNonBlockingClient(unittest.TestCase):
  27     '''
  28     Test Cases class for NonBlockingClient.
  29     '''
  30     def setUp(self):
  31         ''' IdleQueue thread is run and dummy connection is created. '''
  32         self.idlequeue_thread = IdleQueueThread()
  33         self.connection = MockConnection() # for dummy callbacks
  34         self.idlequeue_thread.start()
  35
  36     def tearDown(self):
  37         ''' IdleQueue thread is stopped. '''
  38         self.idlequeue_thread.stop_thread()
  39         self.idlequeue_thread.join()
  40
  41         self.client = None
  42
  43     def open_stream(self, server_port, wrong_pass=False):
  44         '''
  45         Method opening the XMPP connection. It returns when <stream:features>
  46         is received from server.
  47
  48         :param server_port: tuple of (hostname, port) for where the client should
  49         connect.
  50         '''
  51
  52         class TempConnection():
  53             def get_password(self, cb, mechanism):
  54                 if wrong_pass:
  55                     cb('wrong pass')
  56                 else:
  57                     cb(credentials[1])
  58             def on_connect_failure(self):
  59                 pass
  60
  61         self.client = client_nb.NonBlockingClient(
  62                 domain=server_port[0],
  63                 idlequeue=self.idlequeue_thread.iq,
  64                 caller=Mock(realClass=TempConnection))
  65
  66         self.client.connect(
  67                 hostname=server_port[0],
  68                 port=server_port[1],
  69                 on_connect=lambda *args: self.connection.on_connect(True, *args),
  70                 on_connect_failure=lambda *args: self.connection.on_connect(
  71                         False, *args))
  72
  73         self.assert_(self.connection.wait(),
  74                 msg='waiting for callback from client constructor')
  75
  76         # if on_connect was called, client has to be connected and vice versa
  77         if self.connection.connect_succeeded:
  78             self.assert_(self.client.get_connect_type())
  79         else:
  80             self.assert_(not self.client.get_connect_type())
  81
  82     def client_auth(self, username, password, resource, sasl):
  83         '''
  84         Method authenticating connected client with supplied credentials. Returns
  85         when authentication is over.
  86
  87         :param sasl: whether to use sasl (sasl=1) or old (sasl=0) authentication
  88         :todo: to check and be more specific about when it returns
  89                 (bind, session..)
  90         '''
  91         self.client.auth(username, password, resource, sasl,
  92                         on_auth=self.connection.on_auth)
  93
  94         self.assert_(self.connection.wait(), msg='waiting for authentication')
  95
  96     def do_disconnect(self):
  97         '''
  98         Does disconnecting of connected client. Returns when TCP connection is
  99         closed.
 100         '''
 101         self.client.RegisterDisconnectHandler(self.connection.set_event)
 102         self.client.disconnect()
 103
 104         self.assertTrue(self.connection.wait(), msg='waiting for disconnecting')
 105
 106     def test_proper_connect_sasl(self):
 107         '''
 108         The ideal testcase - client is connected, authenticated with SASL and
 109         then disconnected.
 110         '''
 111         self.open_stream(xmpp_server_port)
 112
 113         # if client is not connected, lets raise the AssertionError
 114         self.assert_(self.client.get_connect_type())
 115         # client.disconnect() is already called from NBClient via
 116         # _on_connected_failure, no need to call it here
 117
 118         self.client_auth(credentials[0], credentials[1], credentials[2], sasl=1)
 119         self.assert_(self.connection.con)
 120         self.assert_(self.connection.auth=='sasl', msg='Unable to auth via SASL')
 121
 122         self.do_disconnect()
 123
 124     def test_proper_connect_oldauth(self):
 125         '''
 126         The ideal testcase - client is connected, authenticated with old auth and
 127         then disconnected.
 128         '''
 129         self.open_stream(xmpp_server_port)
 130         self.assert_(self.client.get_connect_type())
 131         self.client_auth(credentials[0], credentials[1], credentials[2], sasl=0)
 132         self.assert_(self.connection.con)
 133         features = self.client.Dispatcher.Stream.features
 134         if not features.getTag('auth'):
 135             print "Server doesn't support old authentication type, ignoring test"
 136         else:
 137             self.assert_(self.connection.auth=='old_auth',
 138                     msg='Unable to auth via old_auth')
 139         self.do_disconnect()
 140
 141     def test_connect_to_nonexisting_host(self):
 142         '''
 143         Connect to nonexisting host. DNS request for A records should return
 144         nothing.
 145         '''
 146         self.open_stream(('fdsfsdf.fdsf.fss', 5222))
 147         self.assert_(not self.client.get_connect_type())
 148
 149     def test_connect_to_wrong_port(self):
 150         '''
 151         Connect to nonexisting server. DNS request for A records should return an
 152         IP      but there shouldn't be XMPP server running on specified port.
 153         '''
 154         self.open_stream((xmpp_server_port[0], 31337))
 155         self.assert_(not self.client.get_connect_type())
 156
 157     def test_connect_with_wrong_creds(self):
 158         '''
 159         Connecting with invalid password.
 160         '''
 161         self.open_stream(xmpp_server_port, wrong_pass=True)
 162         self.assert_(self.client.get_connect_type())
 163         self.client_auth(credentials[0], 'wrong pass', credentials[2], sasl=1)
 164         self.assert_(self.connection.auth is None)
 165         self.do_disconnect()
 166
 167
 168 if __name__ == '__main__':
 169     unittest.main()