Source Code for Module solvcon.tests.test_batch

  1  # -*- coding: UTF-8 -*- 
  2   
  3  from unittest import TestCase 
  4   
  5 -class TestBatch(TestCase): 
  6 -    def test_init(self): 
  7          from ..case import BlockCase 
  8          from ..batch import Batch 
  9          case = BlockCase() 
 10   
 11 -    def test_script(self): 
 12          from ..case import BlockCase 
 13          from ..batch import Batch 
 14          case = BlockCase(rootdir='/tmp') 
 15          sbm = Batch(case, arnname='arn') 
 16          self.assertRaises(NotImplementedError, lambda: sbm.str_resource) 
 17          self.assertRaises(NotImplementedError, lambda: sbm.str_jobname) 
 18          self.assertRaises(NotImplementedError, lambda: sbm.str_output) 
 19          self.assertRaises(NotImplementedError, lambda: sbm.str_shell) 
 20          self.assertRaises(NotImplementedError, lambda: str(sbm)) 
 21   
 22 -class TestTorque(TestCase): 
 23      SCRIPT = '''#!/bin/sh 
 24   
 25  #PBS -N arn 
 26  #PBS -j oe 
 27  #PBS -S /bin/sh 
 28  echo "Customized paths for job:" 
 29  . $HOME/.bashrc_path 
 30  export PYTHONPATH=/tmp:$PYTHONPATH 
 31  echo "Run @`date`:" 
 32  cd /tmp/arn 
 33  time scg run arn --runlevel %d 
 34  echo "Finish @`date`."''' 
 35   
 36 -    def test_script(self): 
 37          import sys 
 38          from nose.plugins.skip import SkipTest 
 39          if sys.platform.startswith('win'): raise SkipTest 
 40          from ..case import BlockCase 
 41          from ..batch import batregy 
 42          case = BlockCase(rootdir='/tmp') 
 43          sbm = batregy.Torque(case, arnname='arn') 
 44          self.assertEqual(str(sbm), self.SCRIPT) 
 45   
 46 -    def test_tofile(self): 
 47          import sys 
 48          from nose.plugins.skip import SkipTest 
 49          if sys.platform.startswith('win'): raise SkipTest 
 50          import os, shutil 
 51          from ..case import BlockCase 
 52          from ..batch import batregy 
 53          case = BlockCase(rootdir='/tmp') 
 54          msg = [] 
 55          case.info = lambda m: msg.append(m) 
 56          sbm = batregy.Torque(case, arnname='arn') 
 57          fnlist = sbm.tofile() 
 58          for it in range(len(fnlist)): 
 59              self.assertEqual(fnlist[it], '/tmp/arn/arn.pbs%d'%it) 
 60              fn = fnlist[it] 
 61              f = open(fn) 
 62              self.assertEqual(f.read(), self.SCRIPT%it) 
 63              f.close() 
 64          shutil.rmtree('/tmp/arn') 
 65          self.assertFalse(os.path.exists('/tmp/arn')) 
 66   
 67 -    def test_with_jobdir(self): 
 68          import sys 
 69          from nose.plugins.skip import SkipTest 
 70          if sys.platform.startswith('win'): raise SkipTest 
 71          import os, shutil 
 72          from ..case import BlockCase 
 73          from ..batch import batregy 
 74          case = BlockCase(rootdir='/tmp') 
 75          msg = [] 
 76          case.info = lambda m: msg.append(m) 
 77          sbm = batregy.Torque(case, arnname='arn') 
 78          if not os.path.exists('/tmp/arn'): 
 79              os.makedirs('/tmp/arn') 
 80          fnlist = sbm.tofile() 
 81          for it in range(len(fnlist)): 
 82              self.assertEqual(fnlist[it], '/tmp/arn/arn.pbs%d'%it) 
 83          self.assertEqual(len(msg), 1) 
 84          self.assertEqual(msg[0], 'Job directory was there: /tmp/arn\n') 
 85          shutil.rmtree('/tmp/arn') 
 86          self.assertFalse(os.path.exists('/tmp/arn')) 
 87   
 88 -    def test_without_jobdir(self): 
 89          import sys 
 90          from nose.plugins.skip import SkipTest 
 91          if sys.platform.startswith('win'): raise SkipTest 
 92          import os, shutil 
 93          from ..case import BlockCase 
 94          from ..batch import batregy 
 95          case = BlockCase(rootdir='/tmp') 
 96          msg = [] 
 97          case.info = lambda m: msg.append(m) 
 98          sbm = batregy.Torque(case, arnname='arn') 
 99          if os.path.exists('/tmp/arn'): 
100              shutil.rmtree('/tmp/arn') 
101          fnlist = sbm.tofile() 
102          for it in range(len(fnlist)): 
103              self.assertEqual(fnlist[it], '/tmp/arn/arn.pbs%d'%it) 
104          self.assertEqual(len(msg), 0) 
105          shutil.rmtree('/tmp/arn') 
106          self.assertFalse(os.path.exists('/tmp/arn')) 
107   
108 -    def test_with_empty(self): 
109          import sys 
110          from nose.plugins.skip import SkipTest 
111          if sys.platform.startswith('win'): raise SkipTest 
112          import os, shutil 
113          from ..case import BlockCase 
114          from ..batch import batregy 
115          case = BlockCase(rootdir='/tmp', empty_jobdir=True) 
116          msg = [] 
117          case.info = lambda m: msg.append(m) 
118          sbm = batregy.Torque(case, arnname='arn') 
119          if not os.path.exists('/tmp/arn'): 
120              os.makedirs('/tmp/arn') 
121          fnlist = sbm.tofile() 
122          for it in range(len(fnlist)): 
123              self.assertEqual(fnlist[it], '/tmp/arn/arn.pbs%d'%it) 
124          self.assertEqual(len(msg), 2) 
125          self.assertEqual(msg[0], 'Job directory was there: /tmp/arn\n') 
126          self.assertEqual(msg[1], 'Delete all file in job directory.\n') 
127          shutil.rmtree('/tmp/arn') 
128          self.assertFalse(os.path.exists('/tmp/arn')) 
129