Inheritance diagram for tests.test_app_BaseServerManager.TestBaseServerManager:

Collaboration diagram for tests.test_app_BaseServerManager.TestBaseServerManager:

Public Member Functions
def	setUp (self)

def	tearDown (self)

def	test_correct_add_new_connection (self)

def	test_set__event_handler (self)

def	test_call_unhandled_event_handler (self)

def	test_send_event (self)

def	test_reply (self)

def	test_run_one_loop (self)

def	test_call_on_timeout_callback (self)

def	test_propagate_out_TransportInternalErr (self)

def	test_process_response_tcp_raw (self)

def	test_create_client_admin_connect (self)

def	test_create_server_connect (self)

def	test_get_client_admin_event (self)

def	test_admin_event_broken_command_type (self)

def	test_admin_event_broken_class_name (self)

def	test_init_stat_fields_admin_connect (self)

def	test_init_user_connection (self)

def	test_stats_recv_several_msg (self)

def	test_stats_update_invalid_stat_key (self)

def	test__admin_fetch_stat_data_all (self)

def	test__admin_fetch_stat_data_some_fields (self)

Public Attributes
	connectionBuilderLight

	adminServerFake

	sock_mock

	poller_mock

	poller_manager_mock

	connect_mock

	base_server_manager

	event_builder

	event

	connect_name

	adminState

Detailed Description

Definition at line 54 of file test_app_BaseServerManager.py.

Member Function Documentation

◆ setUp()

def tests.test_app_BaseServerManager.TestBaseServerManager.setUp ( self )

Definition at line 57 of file test_app_BaseServerManager.py.

     def setUp(self):
         self.connectionBuilderLight = ConnectionBuilderLight()
         self.adminServerFake = self.connectionBuilderLight.build(transport_const.SERVER_CONNECT, BaseServerManager.ADMIN_CONNECT_ENDPOINT)
         
         self.sock_mock = MagicMock(spec = zmq.Socket)
         self.poller_mock = MagicMock(spec = zmq.Poller)
         self.poller_manager_mock = MagicMock(spec = PollerManager)
         self.connect_mock = MagicMock(spec = ConnectionLight) 
         self.base_server_manager = BaseServerManager(self.poller_manager_mock, conectionLightBuilder=self.connectionBuilderLight)
         self.event_builder = EventBuilder()
         self.event = self.event_builder.build(EVENT_TYPES.NEW_TASK, "eventObj")                 
         self.connect_name = "Admin"
         self.event.connect_name = self.connect_name        
         self.adminState = AdminState("BaseServerManager", AdminState.STATE_SHUTDOWN)
         
         

◆ tearDown()

def tests.test_app_BaseServerManager.TestBaseServerManager.tearDown ( self )

Definition at line 73 of file test_app_BaseServerManager.py.

     def tearDown(self):
         self.adminServerFake.close()

◆ test__admin_fetch_stat_data_all()

def tests.test_app_BaseServerManager.TestBaseServerManager.test__admin_fetch_stat_data_all ( self )

Definition at line 289 of file test_app_BaseServerManager.py.

     def test__admin_fetch_stat_data_all(self):        
         connectLightBuilder_mock = MagicMock(ConnectionBuilderLight)
         self.base_server_manager = BaseServerManager(self.poller_manager_mock, conectionLightBuilder=connectLightBuilder_mock)
                         
         adminStatData = AdminStatData("BaseServerManager")        
         adminStatEvent = self.event_builder.build(EVENT_TYPES.ADMIN_FETCH_STAT_DATA, adminStatData)        
         connect_mock = MagicMock(ConnectionLight)
         connect_mock.recv.return_value = adminStatEvent
         
         self.poller_manager_mock.poll.return_value = [self.base_server_manager.ADMIN_CONNECT_CLIENT]        
         self.base_server_manager.addConnection(self.base_server_manager.ADMIN_CONNECT_CLIENT, connect_mock)
 
         expectAdminStatData = AdminStatData("BaseServerManager", self.base_server_manager.statFields)
         expect_event = self.event_builder.build(EVENT_TYPES.ADMIN_FETCH_STAT_DATA_RESPONSE, expectAdminStatData)
         expect_event.uid = adminStatEvent.uid                         
         match_event = Matcher(matchStatEvents, expect_event)
                 
         self.base_server_manager.poll()
         
         connect_mock.send.assert_called_once_with(match_event)
                                 
                 

◆ test__admin_fetch_stat_data_some_fields()

def tests.test_app_BaseServerManager.TestBaseServerManager.test__admin_fetch_stat_data_some_fields ( self )

Definition at line 311 of file test_app_BaseServerManager.py.

     def test__admin_fetch_stat_data_some_fields(self):        
         connectLightBuilder_mock = MagicMock(ConnectionBuilderLight)
         self.base_server_manager = BaseServerManager(self.poller_manager_mock, conectionLightBuilder=connectLightBuilder_mock)
                         
         adminStatData = AdminStatData("BaseServerManager", dict({"Admin_recv_cnt":0, "WrongName":0}))        
         adminStatEvent = self.event_builder.build(EVENT_TYPES.ADMIN_FETCH_STAT_DATA, adminStatData)        
         connect_mock = MagicMock(ConnectionLight)
         connect_mock.recv.return_value = adminStatEvent
         
         self.poller_manager_mock.poll.return_value = [self.base_server_manager.ADMIN_CONNECT_CLIENT]
         
         self.base_server_manager.addConnection(self.base_server_manager.ADMIN_CONNECT_CLIENT, connect_mock)
 
         expectAdminStatData = AdminStatData("BaseServerManager", dict({"Admin_recv_cnt":1, "WrongName":0}))
         expect_event = self.event_builder.build(EVENT_TYPES.ADMIN_FETCH_STAT_DATA_RESPONSE, expectAdminStatData)
         expect_event.uid = adminStatEvent.uid                         
         match_event = Matcher(matchStatEvents, expect_event)
                 
         self.base_server_manager.poll()
         
         connect_mock.send.assert_called_once_with(match_event)
                                 

◆ test_admin_event_broken_class_name()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_admin_event_broken_class_name ( self )

Definition at line 226 of file test_app_BaseServerManager.py.

     def test_admin_event_broken_class_name(self):
         connectLightBuilder_mock = MagicMock(ConnectionBuilderLight)
         self.base_server_manager = BaseServerManager(self.poller_manager_mock, conectionLightBuilder=connectLightBuilder_mock)
                         
         self.adminState.className = "AdminState.STATE_NOP"
         adminEvent = self.event_builder.build(EVENT_TYPES.ADMIN_STATE, self.adminState)        
         connect_mock = MagicMock(ConnectionLight)
         connect_mock.recv.return_value = adminEvent
         
         self.poller_manager_mock.poll.return_value = [self.base_server_manager.ADMIN_CONNECT_CLIENT]        
         self.base_server_manager.addConnection(self.base_server_manager.ADMIN_CONNECT_CLIENT, connect_mock)
         
         self.base_server_manager.poll()
         
         self.assertEqual(self.base_server_manager.exit_flag, False, "stopped")
         connect_mock.send.assert_called_once_with(ANY)
         
         

◆ test_admin_event_broken_command_type()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_admin_event_broken_command_type ( self )

Definition at line 208 of file test_app_BaseServerManager.py.

     def test_admin_event_broken_command_type(self):
         connectLightBuilder_mock = MagicMock(ConnectionBuilderLight)
         self.base_server_manager = BaseServerManager(self.poller_manager_mock, conectionLightBuilder=connectLightBuilder_mock)
                         
         self.adminState.command = AdminState.STATE_NOP
         adminEvent = self.event_builder.build(EVENT_TYPES.ADMIN_STATE, self.adminState)        
         connect_mock = MagicMock(ConnectionLight)
         connect_mock.recv.return_value = adminEvent
         
         self.poller_manager_mock.poll.return_value = [self.base_server_manager.ADMIN_CONNECT_CLIENT]        
         self.base_server_manager.addConnection(self.base_server_manager.ADMIN_CONNECT_CLIENT, connect_mock)
         
         self.base_server_manager.poll()
         
         self.assertEqual(self.base_server_manager.exit_flag, False, "stopped")
         connect_mock.send.assert_called_once_with(ANY)
         
         

◆ test_call_on_timeout_callback()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_call_on_timeout_callback ( self )

Definition at line 135 of file test_app_BaseServerManager.py.

     def test_call_on_timeout_callback(self):
         self.poller_manager_mock.poll.side_effect = ConnectionTimeout("boom")        
         on_timeout_mock = MagicMock()
         self.base_server_manager.on_poll_timeout = on_timeout_mock
         
         self.base_server_manager.poll()
         
         on_timeout_mock.assert_called_once_with()
         
                         

◆ test_call_unhandled_event_handler()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_call_unhandled_event_handler ( self )

Definition at line 93 of file test_app_BaseServerManager.py.

     def test_call_unhandled_event_handler(self):
         unhandled_mock = MagicMock()
         self.base_server_manager.on_unhandled_event = unhandled_mock
                                                                         
         self.base_server_manager.process(self.event)
         
         unhandled_mock.assert_called_once_with(self.event)

◆ test_correct_add_new_connection()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_correct_add_new_connection ( self )

Definition at line 77 of file test_app_BaseServerManager.py.

     def test_correct_add_new_connection(self):
         connect = ConnectionLight(self.sock_mock, transport_const.CLIENT_CONNECT)        
         self.base_server_manager.addConnection(self.connect_name, connect)        
                 
         self.assertEqual(self.base_server_manager.connections[self.connect_name], connect, "")

◆ test_create_client_admin_connect()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_create_client_admin_connect ( self )

Definition at line 176 of file test_app_BaseServerManager.py.

     def test_create_client_admin_connect(self):
         connectLightBuilder_mock = MagicMock(ConnectionBuilderLight)
         self.base_server_manager = BaseServerManager(self.poller_manager_mock, conectionLightBuilder=connectLightBuilder_mock)
         
         connectLightBuilder_mock.build.assert_called_once_with(ANY, ANY)

◆ test_create_server_connect()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_create_server_connect ( self )

Definition at line 183 of file test_app_BaseServerManager.py.

     def test_create_server_connect(self):
         connectLightBuilder_mock = MagicMock(ConnectionBuilderLight)
         adminConnect_mock = MagicMock(ConnectionLight)
         self.base_server_manager = BaseServerManager(self.poller_manager_mock, conectionLightBuilder=connectLightBuilder_mock,
                                                      admin_connection=adminConnect_mock)
         self.assertEqual(connectLightBuilder_mock.build.call_count, 0, "was called")
         
         

◆ test_get_client_admin_event()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_get_client_admin_event ( self )

Definition at line 191 of file test_app_BaseServerManager.py.

     def test_get_client_admin_event(self):
         connectLightBuilder_mock = MagicMock(ConnectionBuilderLight)
         self.base_server_manager = BaseServerManager(self.poller_manager_mock, conectionLightBuilder=connectLightBuilder_mock)
                         
         adminEvent = self.event_builder.build(EVENT_TYPES.ADMIN_STATE, self.adminState)        
         connect_mock = MagicMock(ConnectionLight)
         connect_mock.recv.return_value = adminEvent
         
         self.poller_manager_mock.poll.return_value = [self.base_server_manager.ADMIN_CONNECT_CLIENT]        
         self.base_server_manager.addConnection(self.base_server_manager.ADMIN_CONNECT_CLIENT, connect_mock)
         
         self.base_server_manager.poll()
         
         self.assertEqual(self.base_server_manager.exit_flag, True, "still running")
         connect_mock.send.assert_called_once_with(ANY)
         
                 

◆ test_init_stat_fields_admin_connect()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_init_stat_fields_admin_connect ( self )

Definition at line 244 of file test_app_BaseServerManager.py.

     def test_init_stat_fields_admin_connect(self):
         className = self.base_server_manager.__class__.__name__
         ready = AdminState(className, AdminState.STATE_READY)
         readyEvent = self.base_server_manager.eventBuilder.build(EVENT_TYPES.ADMIN_STATE_RESPONSE, ready)
 
         admin_connect_name = BaseServerManager.ADMIN_CONNECT_CLIENT
         self.assertTrue(self.base_server_manager.statFields[admin_connect_name + "_send_cnt"] == 1, "")
         self.assertTrue(self.base_server_manager.statFields[admin_connect_name + "_recv_cnt"] == 0, "")
         self.assertTrue(self.base_server_manager.statFields[admin_connect_name + "_send_bytes"] == sys.getsizeof(readyEvent), "")
         self.assertTrue(self.base_server_manager.statFields[admin_connect_name + "_recv_bytes"] == 0, "")
     
     

◆ test_init_user_connection()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_init_user_connection ( self )

Definition at line 256 of file test_app_BaseServerManager.py.

     def test_init_user_connection(self):
         fake_user_connect = None
         connect_name = "user_connect"
         self.base_server_manager.addConnection(connect_name, fake_user_connect)
         
         self.assertTrue(self.base_server_manager.statFields[connect_name + "_send_cnt"] == 0, "")
         self.assertTrue(self.base_server_manager.statFields[connect_name + "_recv_cnt"] == 0, "")
         self.assertTrue(self.base_server_manager.statFields[connect_name + "_send_bytes"] == 0, "")
         self.assertTrue(self.base_server_manager.statFields[connect_name + "_recv_bytes"] == 0, "")
         
 

◆ test_process_response_tcp_raw()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_process_response_tcp_raw ( self )

Definition at line 156 of file test_app_BaseServerManager.py.

     def test_process_response_tcp_raw(self):
         raw_server = "raw_server"
         response = Response(["sock_identity", "id", "body"])
         connect_mock = MagicMock(Connection)
         connect_mock.recv.return_value = response
         self.poller_manager_mock.poll.return_value = [raw_server]
         
         self.base_server_manager.addConnection(raw_server, connect_mock)
         event_processor_mock = MagicMock()
         
         expect_event = self.base_server_manager.eventBuilder.build(EVENT_TYPES.SERVER_TCP_RAW, response)
         match_event = Matcher(match_tcp_raw_events, expect_event)
         
         self.base_server_manager.setEventHandler(EVENT_TYPES.SERVER_TCP_RAW, event_processor_mock)
                         
         self.base_server_manager.poll()
         
         event_processor_mock.assert_called_once_with(match_event)
         
         

◆ test_propagate_out_TransportInternalErr()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_propagate_out_TransportInternalErr ( self )

Definition at line 145 of file test_app_BaseServerManager.py.

     def test_propagate_out_TransportInternalErr(self):
         self.poller_manager_mock.poll.return_value = [self.connect_name]
         self.connect_mock.recv.side_effect = TransportInternalErr("boom")
         
         self.base_server_manager.addConnection(self.connect_name, self.connect_mock)
         
         with self.assertRaises(TransportInternalErr):
             list(self.base_server_manager.poll())

◆ test_reply()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_reply ( self )

Definition at line 109 of file test_app_BaseServerManager.py.

     def test_reply(self):
         self.base_server_manager.addConnection(self.connect_name, self.connect_mock)
         
         reply_event = self.event_builder.build(EVENT_TYPES.NEW_TASK, "new obj")            
         expect_event = reply_event
         expect_event.uid = self.event.uid
          
         match_event = Matcher(matchEvents, expect_event)
         self.base_server_manager.reply(self.event, reply_event)
         
         self.connect_mock.send.assert_called_once_with(match_event)
         
                     

◆ test_run_one_loop()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_run_one_loop ( self )

Definition at line 122 of file test_app_BaseServerManager.py.

     def test_run_one_loop(self):        
         self.poller_manager_mock.poll.return_value = [self.connect_name]
         self.connect_mock.recv.return_value = self.event
 
         handler_mock = MagicMock()        
         self.base_server_manager.setEventHandler(EVENT_TYPES.NEW_TASK, handler_mock)
         self.base_server_manager.addConnection(self.connect_name, self.connect_mock)
         
         self.base_server_manager.poll()
                 
         handler_mock.assert_called_once_with(self.event)
         
         

◆ test_send_event()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_send_event ( self )

Definition at line 102 of file test_app_BaseServerManager.py.

     def test_send_event(self):
         self.base_server_manager.addConnection(self.connect_name, self.connect_mock)        
         self.base_server_manager.send(self.connect_name,self. event)
         
         self.connect_mock.send.assert_called_once_with(self.event)

◆ test_set__event_handler()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_set__event_handler ( self )

Definition at line 84 of file test_app_BaseServerManager.py.

     def test_set__event_handler(self):        
         handler_mock = MagicMock()        
         self.base_server_manager.setEventHandler(EVENT_TYPES.NEW_TASK, handler_mock)
         
         self.base_server_manager.process(self.event)
         
         handler_mock.assert_called_once_with(self.event)

◆ test_stats_recv_several_msg()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_stats_recv_several_msg ( self )

Definition at line 267 of file test_app_BaseServerManager.py.

     def test_stats_recv_several_msg(self):
         msg_size = sys.getsizeof(self.event)
         fake_user_connect = None
         connect_name = "user_connect"
         self.event.connect_name = connect_name
         self.base_server_manager.addConnection(connect_name, fake_user_connect)
         
         self.base_server_manager.process(self.event)
         self.base_server_manager.process(self.event)
 
         self.assertTrue(self.base_server_manager.statFields[connect_name + "_send_cnt"] == 0, "")
         self.assertTrue(self.base_server_manager.statFields[connect_name + "_recv_cnt"] == 2, "")
         self.assertTrue(self.base_server_manager.statFields[connect_name + "_send_bytes"] == 0, "")
         self.assertTrue(self.base_server_manager.statFields[connect_name + "_recv_bytes"] == 2 * msg_size, "")
         
         

◆ test_stats_update_invalid_stat_key()

def tests.test_app_BaseServerManager.TestBaseServerManager.test_stats_update_invalid_stat_key ( self )

Definition at line 283 of file test_app_BaseServerManager.py.

     def test_stats_update_invalid_stat_key(self):
         self.event.connect_name = "bad_stat_name"
                 
         self.base_server_manager.process(self.event)