Source Code for Module EDTestCaseEDJob

  1  # coding: utf8 
  2  # 
  3  #    Project: The EDNA Kernel 
  4  #             http://www.edna-site.org 
  5  # 
  6  #    File: "$Id: EDTestCaseParallelExecute.py 2548 2010-12-01 09:14:01Z kieffer $" 
  7  # 
  8  #    Copyright (C) 2011-2011 European Synchrotron Radiation Facility 
  9  #                            Grenoble, France 
 10  # 
 11  #    Principal authors: Jérôme Kieffer (Jerome.Kieffer@esrf.eu) 
 12  #                       Olof Svensson (svensson@esrf.fr)  
 13  # 
 14  #    This program is free software: you can redistribute it and/or modify 
 15  #    it under the terms of the GNU Lesser General Public License as published 
 16  #    by the Free Software Foundation, either version 3 of the License, or 
 17  #    (at your option) any later version. 
 18  # 
 19  #    This program is distributed in the hope that it will be useful, 
 20  #    but WITHOUT ANY WARRANTY; without even the implied warranty of 
 21  #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
 22  #    GNU Lesser General Public License for more details. 
 23  # 
 24  #    You should have received a copy of the GNU General Public License 
 25  #    and the GNU Lesser General Public License  along with this program.   
 26  #    If not, see <http://www.gnu.org/licenses/>. 
 27  # 
 28   
 29  __authors__ = ["Jérôme Kieffer", "Olof Svensson"] 
 30  __contact__ = "Jerome.Kieffer@esrf.eu" 
 31  __license__ = "LGPLv3+" 
 32  __copyright__ = "European Synchrotron Radiation Facility, Grenoble, France" 
 33   
 34  """This is the EDJob interface for launching jobs""" 
 35   
 36  import  time 
 37  from EDVerbose                           import EDVerbose 
 38  from EDTestCase                          import EDTestCase 
 39  from EDAssert                            import EDAssert 
 40  from EDJob                               import EDJob 
 41   
 42   
 43   
 44 -class EDTestCaseEDJob(EDTestCase): 
 45      """ 
 46      Unit & execution test for the EDParallelExecute class 
 47      """ 
 48      strPluginName = "EDPluginTestPluginFactory" 
 49      strXmlInput = """<?xml version="1.0" ?> 
 50  <XSDataString> 
 51      <value>Test string value.</value> 
 52  </XSDataString>""" 
 53 -    def __init__(self, _strTestName=None): 
 54          EDTestCase.__init__(self, "EDTestCaseEDJob") 
 55  #        self.__edpluginPE = EDParallelExecute("EDPluginTestPluginFactory", fakeXML) 
 56   
 57   
 58 -    def unitTestInitialState(self): 
 59          """ 
 60          check the status after a job creation 
 61          """ 
 62          EDVerbose.DEBUG("EDTestCaseEDJob.unitTestInitialState") 
 63          edJob = EDJob(self.strPluginName) 
 64          strJobId = edJob.getJobId() 
 65          EDVerbose.DEBUG("EDJobId is: %s" % strJobId) 
 66          EDAssert.equal(2, len(strJobId.split("-")), "JobID is composed of 2 parts") 
 67          EDAssert.equal(True, strJobId.split("-")[1].isdigit(), "JobID's second part is an integer") 
 68          EDAssert.equal("uninitialized", edJob.getStatus(), "Initial stat is ''uninitialized''") 
 69   
 70   
 71 -    def unitTestSetGetData(self): 
 72          """ 
 73          check the status after a job creation 
 74          """ 
 75          EDVerbose.DEBUG("EDTestCaseEDJob.unitTestSetGetData") 
 76          edJob = EDJob(self.strPluginName) 
 77          edJob.setDataInput(self.strXmlInput) 
 78          EDAssert.equal(self.strXmlInput, edJob.getDataInput().strip(), "Data Input is correctly set") 
 79          EDAssert.equal("uninitialized", edJob.getStatus(), "Job %s is still ''uninitialized''" % edJob.getJobId()) 
 80   
 81 -    def unitTestExecute(self): 
 82          """ 
 83          check the execution of a job (without callback) 
 84          """ 
 85          EDVerbose.DEBUG("EDTestCaseEDJob.unitTestExecute") 
 86          edJob = EDJob(self.strPluginName) 
 87          strJobId = edJob.getJobId() 
 88          edJob.setDataInput(self.strXmlInput) 
 89          ref = edJob.execute() 
 90          EDAssert.equal(strJobId, ref, "JobId has not changed") 
 91          strStatus = edJob.getStatus() 
 92          EDVerbose.WARNING("Job %s in State %s" % (strJobId, strStatus)) 
 93   
 94          while strStatus in ["running", "uninitialized"]: 
 95              EDVerbose.WARNING("Job %s in state %s" % (strJobId, strStatus)) 
 96              time.sleep(0.01) 
 97              strStatus = edJob.getStatus() 
 98   
 99          xsdOut = edJob.getDataOutput() 
100          while xsdOut is None: 
101              EDVerbose.WARNING("No Output data, still waiting for output data to arrive, %s" % edJob.getStatus()) 
102              time.sleep(0.01) 
103              xsdOut = edJob.getDataOutput() 
104          strOutput = xsdOut.strip() 
105          strStatus = edJob.getStatus() 
106          while strStatus == "running": 
107              EDVerbose.WARNING("Job %s is still in state %s" % (strJobId, strStatus)) 
108              time.sleep(0.01) 
109              strStatus = edJob.getStatus() 
110   
111          EDAssert.equal(strOutput, self.strXmlInput, "Output is OK") 
112          EDAssert.equal("success", edJob.getStatus(), "Job %s is finished with ''success''" % edJob.getJobId()) 
113   
114   
115 -    def callBack(self, _strJobId): 
116          """ 
117          Example of Call Back function ...  
118          """ 
119          myJob = EDJob.getJobFromID(_strJobId) 
120          strOutput = myJob.getDataOutput().strip() 
121          EDAssert.equal(strOutput, self.strXmlInput, "From Callback: Output is OK") 
122          EDAssert.equal("success", myJob.getStatus(), "From Callback: Job %s is finished with ''success''" % _strJobId) 
123   
124   
125 -    def unitTestExecuteCallback(self): 
126          """ 
127          check the execution of a job (without callback) 
128          """ 
129          EDVerbose.DEBUG("EDTestCaseEDJob.unitTestExecuteCallback") 
130          edJob = EDJob(self.strPluginName) 
131          edJob.connectCallBack(self.callBack) 
132          edJob.setDataInput(self.strXmlInput) 
133          strJobId = edJob.execute() 
134          strStatus = edJob.getStatus() 
135          EDVerbose.DEBUG("Job %s in State ''%s''" % (strJobId, strStatus)) 
136   
137 -    def unitTestExecuteCallbackSuccess(self): 
138          """ 
139          check the execution of a job (without callback) 
140          """ 
141          EDVerbose.DEBUG("EDTestCaseEDJob.unitTestExecuteCallbackSuccess") 
142          edJob = EDJob(self.strPluginName) 
143          edJob.connectSUCCESS(self.callBack) 
144          edJob.setDataInput(self.strXmlInput) 
145          strJobId = edJob.execute() 
146          strStatus = edJob.getStatus() 
147          EDVerbose.DEBUG("Job %s in State ''%s''" % (strJobId, strStatus)) 
148   
149   
150 -    def process(self): 
151          self.addTestMethod(self.unitTestInitialState) 
152          self.addTestMethod(self.unitTestSetGetData) 
153          self.addTestMethod(self.unitTestExecuteCallback) 
154          self.addTestMethod(self.unitTestExecuteCallbackSuccess) 
155          self.addTestMethod(self.unitTestExecute) 
156   
157   
158  if __name__ == '__main__': 
159   
160      edTestCaseEDJob = EDTestCaseEDJob("EDTestCaseEDJob") 
161      edTestCaseEDJob.execute() 
162