htsworkflow/util/test/test_queuecommands.py

   1 import os
   2 import logging
   3 import time
   4 import unittest
   5
   6
   7 from htsworkflow.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         # pity I had to add a 1 second sleep
  30         self.failUnless( end > 2.9 and end < 3.1,
  31                          "took %s seconds, exected ~3" % (end,))
  32
  33     def test_limited_run(self):
  34         """
  35         Run a limited number of jobs
  36         """
  37         cmds = ['/bin/sleep 1',
  38                 '/bin/sleep 2',
  39                 '/bin/sleep 3',]
  40
  41         q = QueueCommands(cmds, 2)
  42
  43         start = time.time()
  44         q.run()
  45         end = time.time()-start
  46         # pity I had to add a 1 second sleep
  47         self.failUnless( end > 5.9 and end < 6.1,
  48                          "took %s seconds, expected ~6" % (end,))
  49
  50 def suite():
  51     return unittest.makeSuite(testQueueCommands, 'test')
  52
  53 if __name__ == "__main__":
  54     unittest.main(defaultTest='suite')
  55
  56
  57
  58