1  """Tests for pylab tools module.
   2  """
   3  #-----------------------------------------------------------------------------
   4  # Copyright (c) 2011, the IPython Development Team.
   5  #
   6  # Distributed under the terms of the Modified BSD License.
   7  #
   8  # The full license is in the file COPYING.txt, distributed with this software.
   9  #-----------------------------------------------------------------------------
  10  
  11  #-----------------------------------------------------------------------------
  12  # Imports
  13  #-----------------------------------------------------------------------------
  14  
  15  # Stdlib imports
  16  import time
  17  
  18  # Our own imports
  19  from IPython.lib import backgroundjobs as bg
  20  
  21  #-----------------------------------------------------------------------------
  22  # Globals and constants
  23  #-----------------------------------------------------------------------------
  24  t_short = 0.0001  # very short interval to wait on jobs
  25  
  26  #-----------------------------------------------------------------------------
  27  # Local utilities
  28  #-----------------------------------------------------------------------------
  29  def sleeper(interval=t_short, *a, **kw):
  30      args = dict(interval=interval,
  31                  other_args=a,
  32                  kw_args=kw)
  33      time.sleep(interval)
  34      return args
  35  
  36  def crasher(interval=t_short, *a, **kw):
  37      time.sleep(interval)
  38      raise Exception("Dead job with interval %s" % interval)
  39  
  40  #-----------------------------------------------------------------------------
  41  # Classes and functions
  42  #-----------------------------------------------------------------------------
  43  
  44  def test_result():
  45      """Test job submission and result retrieval"""
  46      jobs = bg.BackgroundJobManager()
  47      j = jobs.new(sleeper)
  48      j.join()
  49      assert j.result["interval"] == t_short
  50  
  51  
  52  def test_flush():
  53      """Test job control"""
  54      jobs = bg.BackgroundJobManager()
  55      j = jobs.new(sleeper)
  56      j.join()
  57      assert len(jobs.completed) == 1
  58      assert len(jobs.dead) == 0
  59      jobs.flush()
  60      assert len(jobs.completed) == 0
  61      
  62  
  63  def test_dead():
  64      """Test control of dead jobs"""
  65      jobs = bg.BackgroundJobManager()
  66      j = jobs.new(crasher)
  67      j.join()
  68      assert len(jobs.completed) == 0
  69      assert len(jobs.dead) == 1
  70      jobs.flush()    
  71      assert len(jobs.dead) == 0
  72  
  73  
  74  def test_longer():
  75      """Test control of longer-running jobs"""
  76      jobs = bg.BackgroundJobManager()
  77      # Sleep for long enough for the following two checks to still report the
  78      # job as running, but not so long that it makes the test suite noticeably
  79      # slower. 
  80      j = jobs.new(sleeper, 0.1)
  81      assert len(jobs.running) == 1
  82      assert len(jobs.completed) == 0
  83      j.join()
  84      assert len(jobs.running) == 0
  85      assert len(jobs.completed) == 1
  86