Inheritance diagram for tests.test_dtm_TasksStateUpdateService.TestTasksStateUpdateService:

Collaboration diagram for tests.test_dtm_TasksStateUpdateService.TestTasksStateUpdateService:

Public Member Functions
def	__init__ (self, args, kvargs)

def	setUp (self)

def	tearDown (self)

def	testUpdateStateNomal (self)

def	testUpdateStateError (self)

def	testCheckState (self)

Public Attributes
	eventBuilder

	inited

	adminServer

	taskManagerConnServer

	eeManagerServer

	tsus

	hceNode

	drceCommandConvertor

Static Public Attributes
string	CONFIG_SECTION = "TasksStateUpdateService"

string	SERVICE_BIND_IP = "127.0.0.1"

string	SERVICE_BIND_PORT = "5500"

Detailed Description

Definition at line 34 of file test_dtm_TasksStateUpdateService.py.

Constructor & Destructor Documentation

◆ init()

def tests.test_dtm_TasksStateUpdateService.TestTasksStateUpdateService.__init__	(	self,
		args,
		kvargs
	)

Definition at line 41 of file test_dtm_TasksStateUpdateService.py.

   def __init__(self, *args, **kvargs):
     unittest.TestCase.__init__(self, *args, **kvargs)
     self.eventBuilder = EventBuilder()
 

Member Function Documentation

◆ setUp()

def tests.test_dtm_TasksStateUpdateService.TestTasksStateUpdateService.setUp ( self )

Definition at line 45 of file test_dtm_TasksStateUpdateService.py.

   def setUp(self):
     if not hasattr(self, "inited"):
       self.inited = True
 
       config = ConfigParser.RawConfigParser()
       config.add_section(self.CONFIG_SECTION)
       config.set(self.CONFIG_SECTION, "clientTasksManager", "TasksManager")
       config.set(self.CONFIG_SECTION, "clientExecutionEnvironmentManager","EEManager")
       config.set(self.CONFIG_SECTION, "serverHost", self.SERVICE_BIND_IP)
       config.set(self.CONFIG_SECTION, "serverPort", self.SERVICE_BIND_PORT)
       
       #Dependent objects creation
       idGenerator = IDGenerator()
       connectionBuilderLight = ConnectionBuilderLight()
       connectionBuilder = ConnectionBuilder(idGenerator)
       #create AdminServer 
       self.adminServer = connectionBuilderLight.build(TRANSPORT_CONSTS.SERVER_CONNECT, BaseServerManager.ADMIN_CONNECT_ENDPOINT)
       self.taskManagerConnServer = connectionBuilderLight.build(TRANSPORT_CONSTS.SERVER_CONNECT, "TasksManager")
       self.eeManagerServer = connectionBuilderLight.build(TRANSPORT_CONSTS.SERVER_CONNECT, "EEManager")
       # time.sleep(1)
       self.tsus = TasksStateUpdateService(config)
       self.tsus.pollTimeout = 100
       # start the threaded server
       self.tsus.start()
       self.hceNode = connectionBuilder.build(TRANSPORT_CONSTS.DATA_CONNECT_TYPE, ConnectionParams(self.SERVICE_BIND_IP, self.SERVICE_BIND_PORT), TRANSPORT_CONSTS.CLIENT_CONNECT)
       self.drceCommandConvertor = CommandConvertor()
 

◆ tearDown()

def tests.test_dtm_TasksStateUpdateService.TestTasksStateUpdateService.tearDown ( self )

Definition at line 72 of file test_dtm_TasksStateUpdateService.py.

   def tearDown(self):
     self.hceNode.close()
     self.adminServer.close()
     self.eeManagerServer.close()
     self.tsus.exit_flag = True
     time.sleep(0.3)
     self.taskManagerConnServer.close()
     time.sleep(0.2)
     for conn in self.tsus.connections.itervalues():
       conn.close()
 

◆ testCheckState()

def tests.test_dtm_TasksStateUpdateService.TestTasksStateUpdateService.testCheckState ( self )

Definition at line 142 of file test_dtm_TasksStateUpdateService.py.

   def testCheckState(self):
     allAvailableTaskIds = ["1","2"]
     TasksStateUpdateService.FETCH_TASKS_IDS_INTERVAL = 1
 
     #test send CheckTaskState
     time.sleep(TasksStateUpdateService.FETCH_TASKS_IDS_INTERVAL)
     pollResult = self.taskManagerConnServer.poll(0.1)
     self.assertEqual(zmq.POLLIN, zmq.POLLIN & pollResult)
     event = self.taskManagerConnServer.recv()
     self.assertEqual(event.eventType, DTM_CONSTS.EVENT_TYPES.FETCH_AVAILABLE_TASK_IDS)
     res = AvailableTaskIds(allAvailableTaskIds)
     responseEvent = self.eventBuilder.build(DTM_CONSTS.EVENT_TYPES.AVAILABLE_TASK_IDS_RESPONSE, res)
     responseEvent.connect_identity = event.connect_identity
     self.taskManagerConnServer.send(responseEvent)
     time.sleep(0.1)
 
     #check the EEManager received the CheckTaskState
     pollResult = self.eeManagerServer.poll(0.1)
     self.assertEqual(zmq.POLLIN, zmq.POLLIN & pollResult)
     event = self.eeManagerServer.recv()
     self.assertEqual(event.eventType, DTM_CONSTS.EVENT_TYPES.CHECK_TASK_STATE)
     self.assertTrue(isinstance(event.eventObj, CheckTaskState))
     firstId = event.eventObj.id
     self.assertIn(firstId, allAvailableTaskIds)
 
     #test send CheckTaskState again
     time.sleep(TasksStateUpdateService.FETCH_TASKS_IDS_INTERVAL)
     pollResult = self.taskManagerConnServer.poll(0.1)
     self.assertEqual(zmq.POLLIN, zmq.POLLIN & pollResult)
     event = self.taskManagerConnServer.recv()
     self.assertEqual(event.eventType, DTM_CONSTS.EVENT_TYPES.FETCH_AVAILABLE_TASK_IDS)
     res = AvailableTaskIds(allAvailableTaskIds)
     responseEvent = self.eventBuilder.build(DTM_CONSTS.EVENT_TYPES.AVAILABLE_TASK_IDS_RESPONSE, res)
     responseEvent.connect_identity = event.connect_identity
     self.taskManagerConnServer.send(responseEvent)
     time.sleep(0.1)
 
     pollResult = self.eeManagerServer.poll(0.1)
     self.assertEqual(zmq.POLLIN, zmq.POLLIN & pollResult)
     event = self.eeManagerServer.recv()
     self.assertEqual(event.eventType, DTM_CONSTS.EVENT_TYPES.CHECK_TASK_STATE)
     self.assertTrue(isinstance(event.eventObj, CheckTaskState))
     secondId = event.eventObj.id
     self.assertIn(secondId, allAvailableTaskIds)
     #test the duplicate task avoiding
     self.assertNotEqual(firstId, secondId)
 
     #test send CheckTaskState again, all task sent, then it clear the sent records
     time.sleep(TasksStateUpdateService.FETCH_TASKS_IDS_INTERVAL)
     pollResult = self.taskManagerConnServer.poll(0.1)
     self.assertEqual(zmq.POLLIN, zmq.POLLIN & pollResult)
     event = self.taskManagerConnServer.recv()
     self.assertEqual(event.eventType, DTM_CONSTS.EVENT_TYPES.FETCH_AVAILABLE_TASK_IDS)
     res = AvailableTaskIds(allAvailableTaskIds)
     responseEvent = self.eventBuilder.build(DTM_CONSTS.EVENT_TYPES.AVAILABLE_TASK_IDS_RESPONSE, res)
     responseEvent.connect_identity = event.connect_identity
     self.taskManagerConnServer.send(responseEvent)
     time.sleep(0.1)
 
     #check the EEManager received the CheckTaskState
     pollResult = self.eeManagerServer.poll(0.1)
     self.assertEqual(zmq.POLLIN, zmq.POLLIN & pollResult)
     event = self.eeManagerServer.recv()
     self.assertEqual(event.eventType, DTM_CONSTS.EVENT_TYPES.CHECK_TASK_STATE)
     self.assertTrue(isinstance(event.eventObj, CheckTaskState))
     self.assertIn(event.eventObj.id, allAvailableTaskIds)
 

◆ testUpdateStateError()

def tests.test_dtm_TasksStateUpdateService.TestTasksStateUpdateService.testUpdateStateError ( self )

Definition at line 115 of file test_dtm_TasksStateUpdateService.py.

   def testUpdateStateError(self):
     req = Request("1")
     req.add_data('''[{
       "error_code": 3, 
       "error_message": "error message for test", 
       "id": "1", 
       "type": 0,
       "host": "", 
       "port": 0, 
       "state": 0, 
       "pid": 1186, 
       "stdout": "",
       "stderror": "",
       "exit_status": 0, 
       "files": [], 
       "node":"node for test",
       "time": 0
     }]
     ''')
     self.hceNode.send(req)
     time.sleep(0.3)
     pollResult = self.taskManagerConnServer.poll()
     #TasksStateService don't send message to TaskManager
     if pollResult & zmq.POLLIN == zmq.POLLIN:
       event = self.taskManagerConnServer.recv()
       self.assertNotEqual(event.eventType, DTM_CONSTS.EVENT_TYPES.UPDATE_TASK_FIELDS)
     

◆ testUpdateStateNomal()

def tests.test_dtm_TasksStateUpdateService.TestTasksStateUpdateService.testUpdateStateNomal ( self )

Definition at line 83 of file test_dtm_TasksStateUpdateService.py.

   def testUpdateStateNomal(self):
     req = Request("1")
     req.add_data('''[{
       "error_code": 0, 
       "error_message": "", 
       "id": "123", 
       "type": 0,
       "host": "", 
       "port": 0, 
       "state": 0, 
       "pid": 1186, 
       "stdout": "",
       "stderror": "",
       "exit_status": 0, 
       "files": [], 
       "node":"node for test",
       "time": 0
     }]
     ''')
     self.hceNode.send(req)
     time.sleep(0.3)
     #TasksStateUpdateService send UpdateTaskFields message to TaskManager
     pollResult = self.taskManagerConnServer.poll()
     self.assertEqual(zmq.POLLIN, pollResult & zmq.POLLIN)
     event = self.taskManagerConnServer.recv()
     self.assertEqual(event.eventType, DTM_CONSTS.EVENT_TYPES.UPDATE_TASK_FIELDS)
     self.assertTrue(isinstance(event.eventObj, UpdateTaskFields))
     self.assertEqual(event.eventObj.id, "123")
     self.assertEqual(event.eventObj.fields["state"], 0)
     self.assertEqual(event.eventObj.fields["pId"], 1186)
     self.assertEqual(event.eventObj.fields["nodeName"], "node for test")
 