Back to index

rabbitmq-server  2.8.4
connect_options.py
Go to the documentation of this file.
00001 import unittest
00002 import stomp
00003 import base
00004 import test_util
00005 
00006 class TestConnectOptions(base.BaseTest):
00007 
00008     def test_implicit_connect(self):
00009         ''' Implicit connect with receipt on first command '''
00010         self.conn.disconnect()
00011         test_util.enable_implicit_connect()
00012         listener = base.WaitableListener()
00013         new_conn = stomp.Connection()
00014         new_conn.set_listener('', listener)
00015 
00016         new_conn.start() # not going to issue connect
00017         new_conn.subscribe(destination="/topic/implicit", id='sub_implicit', receipt='implicit')
00018 
00019         try:
00020             self.assertTrue(listener.await(5))
00021             self.assertEquals(1, len(listener.receipts),
00022                               'Missing receipt. Likely not connected')
00023             self.assertEquals('implicit', listener.receipts[0]['headers']['receipt-id'])
00024         finally:
00025             new_conn.disconnect()
00026             test_util.disable_implicit_connect()
00027 
00028     def test_default_user(self):
00029         ''' Default user connection '''
00030         self.conn.disconnect()
00031         test_util.enable_default_user()
00032         listener = base.WaitableListener()
00033         new_conn = stomp.Connection()
00034         new_conn.set_listener('', listener)
00035         new_conn.start()
00036         new_conn.connect()
00037         try:
00038             self.assertFalse(listener.await(3)) # no error back
00039             self.assertTrue(new_conn.is_connected())
00040         finally:
00041             new_conn.disconnect()
00042             test_util.disable_default_user()