gaworkflow/util/test/test_queuecommands.py

   1 import os
   2 import logging
   3 import time
   4 import unittest
   5
   6
   7 from gaworkflow.util.queuecommands import QueueCommands
   8
   9 class testQueueCommands(unittest.TestCase):
  10     def setUp(self):
  11         logging.basicConfig(level=logging.DEBUG,
  12                             format='%(asctime)s %(name)-8s %(message)s')
  13
  14
  15
  16     def test_unlimited_run(self):
  17         """
  18         Run everything at once
  19         """
  20         cmds = [['/bin/sleep', '0'],
  21                 ['/bin/sleep', '1'],
  22                 ['/bin/sleep', '2'],]
  23
  24         q = QueueCommands(cmds)
  25         start = time.time()
  26         q.run()
  27         end = time.time()-start
  28         # we should only take the length of the longest sleep
  29         self.failUnless( end > 1.9 and end < 2.1,
  30                          "took %s seconds, exected ~5" % (end,))
  31
  32     def test_limited_run(self):
  33         """
  34         Run a limited number of jobs
  35         """
  36         cmds = [['/bin/sleep', '1'],
  37                 ['/bin/sleep', '2'],
  38                 ['/bin/sleep', '3'],]
  39
  40         q = QueueCommands(cmds, 2)
  41
  42         start = time.time()
  43         q.run()
  44         end = time.time()-start
  45         self.failUnless( end > 3.9 and end < 4.1,
  46                          "took %s seconds, expected ~6" % (end,))
  47
  48 def suite():
  49     return unittest.makeSuite(testQueueCommands, 'test')
  50
  51 if __name__ == "__main__":
  52     unittest.main(defaultTest='suite')
  53
  54
  55
  56