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, 3)
  24         start = time.time()
  25         q.run()
  26         end = time.time()-start
  27         # we should only take the length of the longest sleep
  28         self.failUnless( end > 1.9 and end < 2.1,
  29                          "took %s seconds, exected ~2" % (end,))
  30
  31     def test_limited_run_slow(self):
  32         """
  33         Run a limited number of jobs
  34         """
  35         cmds = ['/bin/sleep 1',
  36                 '/bin/sleep 2',
  37                 '/bin/sleep 3',]
  38
  39         q = QueueCommands(cmds, 2)
  40
  41         start = time.time()
  42         q.run()
  43         end = time.time()-start
  44         self.failUnless( end > 3.9 and end < 4.1,
  45                          "took %s seconds, expected ~4" % (end,))
  46
  47 def suite():
  48     return unittest.makeSuite(testQueueCommands, 'test')
  49
  50 if __name__ == "__main__":
  51     unittest.main(defaultTest='suite')
  52
  53
  54
  55