Inheritance diagram for tests.test_transport_Connection.TestConnection:

Collaboration diagram for tests.test_transport_Connection.TestConnection:

Public Member Functions
def	setUp (self)

def	test_send (self)

def	test_recv_message (self)

def	test_recv_message_timeout_exception (self)

def	test_alive_socket_check (self)

def	test_disconnect (self)

def	create_bound_pair (self, type1=zmq.PAIR, type2=zmq.PAIR, interface='tcp://127.0.0.1')

Public Attributes
	context

	connect_pair

	poller

	connection

Detailed Description

Definition at line 14 of file test_transport_Connection.py.

Member Function Documentation

◆ create_bound_pair()

def tests.test_transport_Connection.TestConnection.create_bound_pair	(	self,
		type1 = `zmq.PAIR`,
		type2 = `zmq.PAIR`,
		interface = `'tcp://127.0.0.1'`
	)

Create a bound socket pair using a random port.

Definition at line 58 of file test_transport_Connection.py.

     def create_bound_pair(self, type1=zmq.PAIR, type2=zmq.PAIR, interface='tcp://127.0.0.1'):        """Create a bound socket pair using a random port."""
         s1 = self.context.socket(type1)
         s1.setsockopt(zmq.LINGER, 0)
         port = s1.bind_to_random_port(interface)
         s2 = self.context.socket(type2)
         s2.setsockopt(zmq.LINGER, 0)
         s2.connect('%s:%s' % (interface, port))
         return s2, s1
 

◆ setUp()

def tests.test_transport_Connection.TestConnection.setUp ( self )

Definition at line 17 of file test_transport_Connection.py.

     def setUp(self):        
         self.context = zmq.Context()
         self.connect_pair = self.create_bound_pair()
         self.poller = zmq.Poller()
         self.connection = Connection(self.connect_pair[0], self.poller)
 
 

◆ test_alive_socket_check()

def tests.test_transport_Connection.TestConnection.test_alive_socket_check ( self )

Definition at line 49 of file test_transport_Connection.py.

     def test_alive_socket_check(self):
         self.assertTrue(self.connection.is_closed() == False, "sock must be valid")

◆ test_disconnect()

def tests.test_transport_Connection.TestConnection.test_disconnect ( self )

Definition at line 53 of file test_transport_Connection.py.

     def test_disconnect(self):
         self.connection.close()
         self.assertTrue(self.connection.is_closed() == True, "close doesn't work")
 
 

◆ test_recv_message()

def tests.test_transport_Connection.TestConnection.test_recv_message ( self )

Definition at line 35 of file test_transport_Connection.py.

     def test_recv_message(self):
         message = ["sock_identity", "232", "New message"]
         self.connect_pair[1].send_multipart(message)
         response = self.connection.recv(1000)
                         
         self.assertEqual(message[0], response.get_uid(), "connection recv is failed")
         self.assertEqual(message[1], response.get_body(), "connection recv is failed")
         
                         

◆ test_recv_message_timeout_exception()

def tests.test_transport_Connection.TestConnection.test_recv_message_timeout_exception ( self )

Definition at line 44 of file test_transport_Connection.py.

     def test_recv_message_timeout_exception(self):
         with self.assertRaises(ConnectionTimeout):
             list(self.connection.recv(250))
 
 

◆ test_send()

def tests.test_transport_Connection.TestConnection.test_send ( self )

Definition at line 24 of file test_transport_Connection.py.

     def test_send(self):
         request = Request("1")
         request.add_data("Hello people!")
         expect_msg = [b"1", b"Hello people!"]
         
         self.connection.send(request)
         recv_message = self.connect_pair[1].recv_multipart()
                 
         self.assertEqual(expect_msg, recv_message, "connection send doesn't work")
 
 