src/Python/Test/Test_Python_Local_SH.py

   1 #  Copyright (C) 2007-2011  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-' + config.EXEC_TEST_NAME
  46     p[ARGUMENTS]  = ["copied-seta.sh", "copied-setb.sh", "orig-result.txt"];
  47     p[NAME] = 'Test_Python_Local_SH'
  48     p[WORKDIR] = config.TEST_LOCAL_SH_WORK_DIR
  49     p[INFILE] = [('seta.sh', 'copied-seta.sh'), ('setb.sh', 'copied-setb.sh'),
  50                    (config.EXEC_TEST_FULL_PATH, 'copied-' + config.EXEC_TEST_NAME)]
  51     p[OUTFILE] = [('result.txt', 'orig-result.txt')]
  52     job.setParametre(p)
  53     # ... and its environment
  54     e = {}
  55     job.setEnvironnement(e)
  56     print job
  57
  58     # Get the catalog
  59     c = BatchManagerCatalog.getInstance()
  60
  61     # Create a BatchManager of type Local_SSH on localhost
  62     bm = c('SH')('localhost')
  63
  64     # Submit the job to the BatchManager
  65     jobid = bm.submitJob(job)
  66     print jobid
  67
  68     # Query the job
  69     jobid.queryJob()
  70
  71     # Wait for the end of the job
  72     state = bm.waitForJobEnd(jobid, config.TEST_LOCAL_SH_TIMEOUT);
  73
  74     if state != FINISHED and state != FAILED:
  75         print "Error: Job not finished after timeout"
  76         return 1;
  77
  78     print "Job", jobid, "is done"
  79
  80     # test the result file
  81     exp = "c = 12"
  82     f = open('result.txt')
  83     res = f.read().strip()
  84     print "result found : %s, expected : %s" % (res, exp)
  85
  86     if (res == exp):
  87         return 0
  88     else:
  89         return 1
  90
  91 if __name__ == "__main__":
  92     retcode = work()
  93     sys.exit(retcode)