htsworkflow/util/test/test_queuecommands.py

   1 import os
   2 import logging
   3 import time
   4 import unittest
   5
   6 from htsworkflow.util.queuecommands import QueueCommands
   7
   8 class testQueueCommands(unittest.TestCase):
   9     def setUp(self):
  10         logging.basicConfig(level=logging.DEBUG,
  11                             format='%(asctime)s %(name)-8s %(message)s')
  12
  13
  14
  15     def test_unlimited_run_slow(self):
  16         """
  17         Run everything at once
  18         """
  19         cmds = ['/bin/sleep 0',
  20                 '/bin/sleep 1',
  21                 '/bin/sleep 2',]
  22
  23         q = QueueCommands(cmds)
  24         start = time.time()
  25         q.run()
  26         end = time.time()-start
  27         # we should only take the length of the longest sleep
  28         # pity I had to add a 1 second sleep
  29         self.failUnless( end > 2.9 and end < 3.1,
  30                          "took %s seconds, exected ~3" % (end,))
  31
  32     def test_limited_run_slow(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         # pity I had to add a 1 second sleep
  46         self.failUnless( end > 5.9 and end < 6.1,
  47                          "took %s seconds, expected ~6" % (end,))
  48
  49 def suite():
  50     return unittest.makeSuite(testQueueCommands, 'test')
  51
  52 if __name__ == "__main__":
  53     unittest.main(defaultTest='suite')
  54
  55
  56
  57