htsworkflow/util/test/test_queuecommands.py

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