Inheritance diagram for tests.test_drce_CommandExecutor.TestCommandExecutor:

Collaboration diagram for tests.test_drce_CommandExecutor.TestCommandExecutor:

Public Member Functions
def	setUp (self)

def	teEst_execute_request (self)

def	test_let_out_timeout_exception (self)

def	test_catch_convert_exception_msg_format (self)

def	test_catch_convert_exception_cmd_format (self)

Public Attributes
	connect_mock

	cmd_executor

Detailed Description

Definition at line 25 of file test_drce_CommandExecutor.py.

Member Function Documentation

◆ setUp()

def tests.test_drce_CommandExecutor.TestCommandExecutor.setUp ( self )

Definition at line 28 of file test_drce_CommandExecutor.py.

     def setUp(self):
         self.connect_mock = MagicMock(spec=Connection)        
         self.cmd_executor = CommandExecutor(self.connect_mock, CommandConvertor())
 
 

◆ teEst_execute_request()

def tests.test_drce_CommandExecutor.TestCommandExecutor.teEst_execute_request ( self )

Definition at line 33 of file test_drce_CommandExecutor.py.

     def teEst_execute_request(self):
         response = Response([
           "1",
         """  
         {
           "error_code":0,
           "error_message" :"msg",
           "time":10,
           "state":1,
           "pid":101,
           "data":[
           {"files":[{"name":"f1", "data":"data1", "action":12}],
              "stdout":"out",
              "stderror":"error",
              "exit_status":5,
              "node":"n1",
              "time":100
           }
           ]
         }         
         """ ])
         
         connect_mock_cfg = {"send.return_value": None,
                                              "recv.return_value": response}        
         self.connect_mock.configure_mock(**connect_mock_cfg)
              
         #check
         connect_expect_calls = [call.send(ANY), call.recv(ANY)]
         
         task_request = TaskCheckRequest("task_id", consts.EXTEND_STATUS_INFO)
         task_response = self.cmd_executor.execute(task_request)
           
         self.connect_mock.assert_has_calls(connect_expect_calls)
         
         test_err_str = "json parsing is failed"
         self.assertEqual(task_response.error_code, 0, test_err_str)
         self.assertEqual(task_response.error_msg, "msg", test_err_str )
         self.assertEqual(task_response.state, 1, test_err_str)
         self.assertEqual(len(task_response.data), 1, test_err_str)
         self.assertEqual(task_response.pid, 101, test_err_str)
         
       
           

Here is the call graph for this function:

◆ test_catch_convert_exception_cmd_format()

def tests.test_drce_CommandExecutor.TestCommandExecutor.test_catch_convert_exception_cmd_format ( self )

Definition at line 96 of file test_drce_CommandExecutor.py.

     def test_catch_convert_exception_cmd_format(self):
         self.connect_mock.send.return_value = None
         self.connect_mock.recv.side_effect = CommandConvertorError("Boom")
         
         task_request = TaskCheckRequest("task_id", consts.EXTEND_STATUS_INFO)
 
         with self.assertRaises(CommandExecutorErr):
             list(self.cmd_executor.execute(task_request))
         

Here is the call graph for this function:

◆ test_catch_convert_exception_msg_format()

def tests.test_drce_CommandExecutor.TestCommandExecutor.test_catch_convert_exception_msg_format ( self )

Definition at line 86 of file test_drce_CommandExecutor.py.

     def test_catch_convert_exception_msg_format(self):
         self.connect_mock.send.return_value = None
         self.connect_mock.recv.side_effect = ResponseFormatErr("Boom")
         
         task_request = TaskCheckRequest("task_id", consts.EXTEND_STATUS_INFO)
 
         with self.assertRaises(CommandExecutorErr):
             list(self.cmd_executor.execute(task_request))
          
 

Here is the call graph for this function:

◆ test_let_out_timeout_exception()

def tests.test_drce_CommandExecutor.TestCommandExecutor.test_let_out_timeout_exception ( self )

Definition at line 76 of file test_drce_CommandExecutor.py.

     def test_let_out_timeout_exception(self):
         self.connect_mock.send.return_value = None
         self.connect_mock.recv.side_effect = ConnectionTimeout("Boom")
         task_request = TaskCheckRequest("task_id", consts.EXTEND_STATUS_INFO)
         
         with self.assertRaises(ConnectionTimeout):
             list(self.cmd_executor.execute(task_request))
             
             
             