src/Python/Test/Test_Python_Local_SH.py

   1 #  Copyright (C) 2007-2008  CEA/DEN, EDF R&D, OPEN CASCADE
   2 #
   3 #  Copyright (C) 2003-2007  OPEN CASCADE, EADS/CCR, LIP6, CEA/DEN,
   4 #  CEDRAT, EDF R&D, LEG, PRINCIPIA R&D, BUREAU VERITAS
   5 #
   6 #  This library is free software; you can redistribute it and/or
   7 #  modify it under the terms of the GNU Lesser General Public
   8 #  License as published by the Free Software Foundation; either
   9 #  version 2.1 of the License.
  10 #
  11 #  This library is distributed in the hope that it will be useful,
  12 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 #  Lesser General Public License for more details.
  15 #
  16 #  You should have received a copy of the GNU Lesser General Public
  17 #  License along with this library; if not, write to the Free Software
  18 #  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  19 #
  20 #  See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  21 #
  22
  23 import config
  24 import os
  25 import sys
  26 import time
  27
  28 # Import libBatch library
  29 from libBatch_Swig import *
  30
  31 def work():
  32     print "*******************************************************************************************"
  33     print "This script tests the local batch submission based on SH. No specific configuration is"
  34     print "needed for this test."
  35     print "*******************************************************************************************"
  36
  37     # eventually remove any previous result
  38     if (os.path.exists('result.txt')):
  39         os.remove('result.txt')
  40
  41     # Define the job...
  42     job = Job()
  43     # ... and its parameters ...
  44     p = {}
  45     p['EXECUTABLE'] = './copied-test-script.sh'
  46     p['NAME'] = 'Test_Local_SH'
  47     p['WORKDIR'] = '/tmp'
  48     p['INFILE'] = [('seta.sh', 'copied-seta.sh'), ('setb.sh', 'copied-setb.sh'),
  49                    ('test-script.sh', 'copied-test-script.sh')]
  50     p['OUTFILE'] = [('result.txt', 'orig-result.txt')]
  51     job.setParametre(p)
  52     # ... and its environment
  53     e = {}
  54     job.setEnvironnement(e)
  55     print job
  56
  57     # Get the catalog
  58     c = BatchManagerCatalog.getInstance()
  59
  60     # Create a BatchManager of type Local_SSH on localhost
  61     bm = c('SH')('localhost')
  62
  63     # Submit the job to the BatchManager
  64     jobid = bm.submitJob(job)
  65     print jobid
  66
  67     # Query the job
  68     jobid.queryJob()
  69
  70     # Wait for the end of the job
  71     state = 'Unknown'
  72     while state != 'Done':
  73         time.sleep(0.1)
  74         jinfo = jobid.queryJob()
  75         try:
  76             state = jinfo.getParametre()['STATE']
  77         except KeyError:
  78             pass
  79
  80     print "Job", jobid, "is done"
  81
  82     # wait for 2 more seconds for the copy of output files and the cleanup
  83     # (there's no cleaner way to do that yet)
  84     time.sleep(2)
  85
  86     # test the result file
  87     exp = "c = 12"
  88     f = open('result.txt')
  89     res = f.read().strip()
  90     print "result found : %s, expected : %s" % (res, exp)
  91
  92     if (res == exp):
  93         return 0
  94     else:
  95         return 1
  96
  97 if __name__ == "__main__":
  98     retcode = work()
  99     sys.exit(retcode)