6 from htsworkflow.util.queuecommands import QueueCommands
8 class testQueueCommands(unittest.TestCase):
10 logging.basicConfig(level=logging.DEBUG,
11 format='%(asctime)s %(name)-8s %(message)s')
15 def test_unlimited_run_slow(self):
17 Run everything at once
19 cmds = ['/bin/sleep 0',
23 q = QueueCommands(cmds)
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,))
32 def test_limited_run_slow(self):
34 Run a limited number of jobs
36 cmds = ['/bin/sleep 1',
40 q = QueueCommands(cmds, 2)
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,))
50 return unittest.makeSuite(testQueueCommands, 'test')
52 if __name__ == "__main__":
53 unittest.main(defaultTest='suite')