ftests.ftest_TasksDataManager.TestTasksDataManager Class Reference

Class TestTasksDataManager, contains functional tests of TasksDataManager module. More...

Inheritance diagram for ftests.ftest_TasksDataManager.TestTasksDataManager:

Collaboration diagram for ftests.ftest_TasksDataManager.TestTasksDataManager:

Public Member Functions
def	setUp (self)
def	connectionInit (self)
def	tearDown (self)
def	insert (self, taskId)
def	testOnNewTask (self)
def	testOnFetchTask (self)
def	testOnUpdateTask (self)
def	testOnDeleteTask (self)
def	testOnInsertEEResponse (self)
def	testOnFetchEEResponse (self)
def	testOnDeleteEEResponse (self)

Public Attributes
	servIndex
	connectionBuilder
	localConnection
	adminServerConnection
	config
	taskDataManager
	eventBuilder
	recvEvent
	event
	reply_event
	moduleName

Detailed Description

Class TestTasksDataManager, contains functional tests of TasksDataManager module.

Definition at line 28 of file ftest_TasksDataManager.py.

Member Function Documentation

connectionInit()

def ftests.ftest_TasksDataManager.TestTasksDataManager.connectionInit

(

self

)

Definition at line 49 of file ftest_TasksDataManager.py.

  def connectionInit(self):
    try:
      self.localConnection = self.connectionBuilder.build(transport.Consts.CLIENT_CONNECT, \
                        self.config.get(self.moduleName, self.taskDataManager.TASK_DATA_MANAGER_SERV_CONFIG_NAME))
      self.servIndex = self.servIndex + 1
    except ConfigParser.NoSectionError:
      logger.error(">>> TasksDataManager can't read config - Section Error")
    except ConfigParser.NoOptionError:
      logger.error(">>> TasksDataManager can't read config - Option Error")

insert()

def ftests.ftest_TasksDataManager.TestTasksDataManager.insert	(		self,
			taskId
	)

Definition at line 70 of file ftest_TasksDataManager.py.

  def insert(self, taskId):
    newTask = dtm.EventObjects.NewTask("bash", taskId)
    event = self.eventBuilder.build(EVENT.NEW_TASK, newTask)
    self.localConnection.send(event) 
    self.recvEvent = self.localConnection.recv()

Here is the caller graph for this function:

setUp()

def ftests.ftest_TasksDataManager.TestTasksDataManager.setUp

(

self

)

Definition at line 31 of file ftest_TasksDataManager.py.

  def setUp(self):
    self.servIndex = 1
    self.connectionBuilder = ConnectionBuilderLight()
    self.localConnection = None
    self.adminServerConnection = self.connectionBuilder.build(transport.Consts.SERVER_CONNECT,
                                                       BaseServerManager.ADMIN_CONNECT_ENDPOINT)
    self.config = ConfigParser.ConfigParser()
    self.config.read("./dtmd.ini")
    self.taskDataManager = TasksDataManager(self.config, self.connectionBuilder)
    self.taskDataManager.start()
    self.eventBuilder = EventBuilder()
    self.recvEvent = None
    self.event = None
    self.reply_event = None
    self.moduleName = self.taskDataManager.__class__.__name__
    self.connectionInit()

tearDown()

def ftests.ftest_TasksDataManager.TestTasksDataManager.tearDown

(

self

)

Definition at line 60 of file ftest_TasksDataManager.py.

  def tearDown(self):
    self.taskDataManager.exit_flag = True
    self.taskDataManager.join()
    for connection in self.taskDataManager.connections.values():
      connection.close()    
    self.adminServerConnection.close()
    self.localConnection.close()
    time.sleep(1)

testOnDeleteEEResponse()

def ftests.ftest_TasksDataManager.TestTasksDataManager.testOnDeleteEEResponse

(

self

)

Definition at line 146 of file ftest_TasksDataManager.py.

  def testOnDeleteEEResponse(self):
    taskId = 335
    deleteEEResponse = dtm.EventObjects.DeleteEEResponseData(taskId)
    event = self.eventBuilder.build(EVENT.DELETE_EE_DATA, deleteEEResponse)
    self.localConnection.send(event)
    self.recvEvent = self.localConnection.recv()
    self.assertTrue(self.recvEvent != None, ">>> ret reply_event is None")
    self.assertTrue(self.recvEvent.eventType == EVENT.DELETE_EE_DATA_RESPONSE,
                    ">>> reply_event event.eventType is wrong")

testOnDeleteTask()

def ftests.ftest_TasksDataManager.TestTasksDataManager.testOnDeleteTask

(

self

)

Definition at line 115 of file ftest_TasksDataManager.py.

  def testOnDeleteTask(self):
    taskId = 335
    self.insert(taskId)
    taskId = 335
    deleteTask = dtm.EventObjects.UpdateTask(taskId)
    event = self.eventBuilder.build(EVENT.DELETE_TASK_DATA, deleteTask)
    self.localConnection.send(event)
    self.recvEvent = self.localConnection.recv()
    self.assertTrue(self.recvEvent != None, ">>> ret reply_event is None")  

Here is the call graph for this function:

testOnFetchEEResponse()

def ftests.ftest_TasksDataManager.TestTasksDataManager.testOnFetchEEResponse

(

self

)

Definition at line 135 of file ftest_TasksDataManager.py.

  def testOnFetchEEResponse(self):
    taskId = 335
    fetchEEResponse = dtm.EventObjects.FetchEEResponseData(taskId)
    event = self.eventBuilder.build(EVENT.FETCH_EE_DATA, fetchEEResponse)
    self.localConnection.send(event)
    self.recvEvent = self.localConnection.recv()
    self.assertTrue(self.recvEvent != None, ">>> ret reply_event is None")
    self.assertTrue(self.recvEvent.eventType == EVENT.FETCH_EE_DATA_RESPONSE,
                    ">>> reply_event event.eventType is wrong")

testOnFetchTask()

def ftests.ftest_TasksDataManager.TestTasksDataManager.testOnFetchTask

(

self

)

Definition at line 84 of file ftest_TasksDataManager.py.

  def testOnFetchTask(self):
    taskId = 335
    self.insert(taskId)
    taskId = 200
    fetchTask = dtm.EventObjects.FetchTaskData(taskId)
    event = self.eventBuilder.build(EVENT.FETCH_TASK_DATA, fetchTask)
    self.localConnection.send(event)
    self.recvEvent = self.localConnection.recv()
    self.assertTrue(self.recvEvent != None, ">>> ret reply_event is None")
    self.assertTrue(self.recvEvent.eventType == EVENT.FETCH_TASK_DATA_RESPONSE,
                    ">>> reply_event event.eventType is wrong")
    self.assertTrue(self.recvEvent.eventObj != 1, ">>> reply.eventobject is None")
    taskId = 335
    fetchTask = dtm.EventObjects.FetchTaskData(taskId)
    event = self.eventBuilder.build(EVENT.FETCH_TASK_DATA, fetchTask)
    self.localConnection.send(event)
    self.recvEvent = self.localConnection.recv()
    self.assertTrue(self.recvEvent.eventObj != 1, ">>> reply.eventobject is None")

Here is the call graph for this function:

testOnInsertEEResponse()

def ftests.ftest_TasksDataManager.TestTasksDataManager.testOnInsertEEResponse

(

self

)

Definition at line 126 of file ftest_TasksDataManager.py.

  def testOnInsertEEResponse(self):
    taskId = 335
    insertEEResponse = dtm.EventObjects.EEResponseData(taskId)
    event = self.eventBuilder.build(EVENT.INSERT_EE_DATA, insertEEResponse)
    self.localConnection.send(event)
    self.recvEvent = self.localConnection.recv()
    self.assertTrue(self.recvEvent  != None, ">>> ret reply_event is None")

testOnNewTask()

def ftests.ftest_TasksDataManager.TestTasksDataManager.testOnNewTask

(

self

)

Definition at line 77 of file ftest_TasksDataManager.py.

  def testOnNewTask(self):
    taskId = 335
    self.insert(taskId)      
    self.assertTrue(self.recvEvent != None, ">>> ret reply_event is None")
    self.assertTrue(self.recvEvent.eventType == EVENT.NEW_TASK_RESPONSE, ">>> reply_event event.eventType is wrong")    

Here is the call graph for this function:

testOnUpdateTask()

def ftests.ftest_TasksDataManager.TestTasksDataManager.testOnUpdateTask

(

self

)

Definition at line 104 of file ftest_TasksDataManager.py.

  def testOnUpdateTask(self): 
    taskId = 335
    self.insert(taskId)
    taskId = 200
    updateTask = dtm.EventObjects.UpdateTask(taskId)
    event = self.eventBuilder.build(EVENT.UPDATE_TASK, updateTask)
    self.localConnection.send(event)
    self.recvEvent = self.localConnection.recv()
    self.assertTrue(self.recvEvent != None, ">>> ret reply_event is None")

Here is the call graph for this function:

Member Data Documentation

adminServerConnection

ftests.ftest_TasksDataManager.TestTasksDataManager.adminServerConnection

Definition at line 35 of file ftest_TasksDataManager.py.

config

ftests.ftest_TasksDataManager.TestTasksDataManager.config

Definition at line 37 of file ftest_TasksDataManager.py.

connectionBuilder

ftests.ftest_TasksDataManager.TestTasksDataManager.connectionBuilder

Definition at line 33 of file ftest_TasksDataManager.py.

event

ftests.ftest_TasksDataManager.TestTasksDataManager.event

Definition at line 43 of file ftest_TasksDataManager.py.

eventBuilder

ftests.ftest_TasksDataManager.TestTasksDataManager.eventBuilder

Definition at line 41 of file ftest_TasksDataManager.py.

localConnection

ftests.ftest_TasksDataManager.TestTasksDataManager.localConnection

Definition at line 34 of file ftest_TasksDataManager.py.

moduleName

ftests.ftest_TasksDataManager.TestTasksDataManager.moduleName

Definition at line 45 of file ftest_TasksDataManager.py.

recvEvent

ftests.ftest_TasksDataManager.TestTasksDataManager.recvEvent

Definition at line 42 of file ftest_TasksDataManager.py.

reply_event

ftests.ftest_TasksDataManager.TestTasksDataManager.reply_event

Definition at line 44 of file ftest_TasksDataManager.py.

servIndex

ftests.ftest_TasksDataManager.TestTasksDataManager.servIndex

Definition at line 32 of file ftest_TasksDataManager.py.

taskDataManager

ftests.ftest_TasksDataManager.TestTasksDataManager.taskDataManager

Definition at line 39 of file ftest_TasksDataManager.py.

---

The documentation for this class was generated from the following file:

• sources/hce/ftests/ftest_TasksDataManager.py