Back to index

rabbitmq-server  2.8.4
ssl_lifecycle.py
Go to the documentation of this file.
00001 import unittest
00002 import os
00003 
00004 import stomp
00005 import base
00006 
00007 ssl_key_file = os.path.abspath("test/certs/client/key.pem")
00008 ssl_cert_file = os.path.abspath("test/certs/client/cert.pem")
00009 ssl_ca_certs = os.path.abspath("test/certs/testca/cacert.pem")
00010 
00011 class TestSslClient(unittest.TestCase):
00012 
00013     def __ssl_connect(self):
00014         conn = stomp.Connection(user="guest", passcode="guest",
00015                                 host_and_ports = [ ('localhost', 61614) ],
00016                                 use_ssl = True, ssl_key_file = ssl_key_file,
00017                                 ssl_cert_file = ssl_cert_file,
00018                                 ssl_ca_certs = ssl_ca_certs)
00019 
00020         conn.start()
00021         conn.connect()
00022         return conn
00023 
00024     def __ssl_auth_connect(self):
00025         conn = stomp.Connection(host_and_ports = [ ('localhost', 61614) ],
00026                                 use_ssl = True, ssl_key_file = ssl_key_file,
00027                                 ssl_cert_file = ssl_cert_file,
00028                                 ssl_ca_certs = ssl_ca_certs)
00029         conn.start()
00030         conn.connect()
00031         return conn
00032 
00033     def test_ssl_connect(self):
00034         conn = self.__ssl_connect()
00035         conn.stop()
00036 
00037     def test_ssl_auth_connect(self):
00038         conn = self.__ssl_auth_connect()
00039         conn.stop()
00040 
00041     def test_ssl_send_receive(self):
00042         conn = self.__ssl_connect()
00043         self.__test_conn(conn)
00044 
00045     def test_ssl_auth_send_receive(self):
00046         conn = self.__ssl_auth_connect()
00047         self.__test_conn(conn)
00048 
00049     def __test_conn(self, conn):
00050         try:
00051             listener = base.WaitableListener()
00052 
00053             conn.set_listener('', listener)
00054 
00055             d = "/topic/ssl.test"
00056             conn.subscribe(destination=d, receipt="sub")
00057 
00058             self.assertTrue(listener.await(1))
00059 
00060             self.assertEquals("sub",
00061                               listener.receipts[0]['headers']['receipt-id'])
00062 
00063             listener.reset(1)
00064             conn.send("Hello SSL!", destination=d)
00065 
00066             self.assertTrue(listener.await())
00067 
00068             self.assertEquals("Hello SSL!", listener.messages[0]['message'])
00069         finally:
00070             conn.disconnect()