upstream/ipython Files · IPython/lib/tests/test_backgroundjobs.py

Test of simpler way to define ast macros. (#14100)...

Test of simpler way to define ast macros. (#14100) Provide and easier way to generate magics and pre-post hooks This introduce a code base way of modifying the ast. This allow to use Template strings with the two special variable names names `__code__` and `__ret__` allowing to quickly write a magic, or hooks that modify the ast. This also introduce a `%code_wrap` cell magic to on the fly wrap code. It is this easy to for example modify IPython to say time each block of code, or retry them, or wrap them in try/except and analyse the error message, profile... Note that this is not new, but simply convenience function and utilities, especially around hygiene.

Matthias Bussonnier - - Load All Authors

File last commit:

r26723:f03bb325


                r28324:c3d7f161

Download file

             test_backgroundjobs.py
        
                    85 lines
            
             | 2.6 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / IPython / lib / tests / test_backgroundjobs.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      """Tests for pylab tools module.

      """

      #-----------------------------------------------------------------------------

      # Copyright (c) 2011, the IPython Development Team.

      #

      # Distributed under the terms of the Modified BSD License.

      #

      # The full license is in the file COPYING.txt, distributed with this software.

      #-----------------------------------------------------------------------------

      #-----------------------------------------------------------------------------

      # Imports

      #-----------------------------------------------------------------------------

      # Stdlib imports

      import time

      # Our own imports

      from IPython.lib import backgroundjobs as bg

      #-----------------------------------------------------------------------------

      # Globals and constants

      #-----------------------------------------------------------------------------

      t_short = 0.0001  # very short interval to wait on jobs

      #-----------------------------------------------------------------------------

      # Local utilities

      #-----------------------------------------------------------------------------

      def sleeper(interval=t_short, *a, **kw):

          args = dict(interval=interval,

                      other_args=a,

                      kw_args=kw)

          time.sleep(interval)

          return args

      def crasher(interval=t_short, *a, **kw):

          time.sleep(interval)

          raise Exception("Dead job with interval %s" % interval)

      #-----------------------------------------------------------------------------

      # Classes and functions

      #-----------------------------------------------------------------------------

      def test_result():

          """Test job submission and result retrieval"""

          jobs = bg.BackgroundJobManager()

          j = jobs.new(sleeper)

          j.join()

          assert j.result["interval"] == t_short

      def test_flush():

          """Test job control"""

          jobs = bg.BackgroundJobManager()

          j = jobs.new(sleeper)

          j.join()

          assert len(jobs.completed) == 1

          assert len(jobs.dead) == 0

          jobs.flush()

          assert len(jobs.completed) == 0

      def test_dead():

          """Test control of dead jobs"""

          jobs = bg.BackgroundJobManager()

          j = jobs.new(crasher)

          j.join()

          assert len(jobs.completed) == 0

          assert len(jobs.dead) == 1

          jobs.flush()    

          assert len(jobs.dead) == 0

      def test_longer():

          """Test control of longer-running jobs"""

          jobs = bg.BackgroundJobManager()

          # Sleep for long enough for the following two checks to still report the

          # job as running, but not so long that it makes the test suite noticeably

          # slower. 

          j = jobs.new(sleeper, 0.1)

          assert len(jobs.running) == 1

          assert len(jobs.completed) == 0

          j.join()

          assert len(jobs.running) == 0

          assert len(jobs.completed) == 1

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

				"""Tests for pylab tools module.
				"""
				#-----------------------------------------------------------------------------
				# Copyright (c) 2011, the IPython Development Team.
				#
				# Distributed under the terms of the Modified BSD License.
				#
				# The full license is in the file COPYING.txt, distributed with this software.
				#-----------------------------------------------------------------------------

				#-----------------------------------------------------------------------------
				# Imports
				#-----------------------------------------------------------------------------

				# Stdlib imports
				import time

				# Our own imports
				from IPython.lib import backgroundjobs as bg

				#-----------------------------------------------------------------------------
				# Globals and constants
				#-----------------------------------------------------------------------------
				t_short = 0.0001 # very short interval to wait on jobs

				#-----------------------------------------------------------------------------
				# Local utilities
				#-----------------------------------------------------------------------------
				def sleeper(interval=t_short, a, *kw):
				args = dict(interval=interval,
				other_args=a,
				kw_args=kw)
				time.sleep(interval)
				return args

				def crasher(interval=t_short, a, *kw):
				time.sleep(interval)
				raise Exception("Dead job with interval %s" % interval)

				#-----------------------------------------------------------------------------
				# Classes and functions
				#-----------------------------------------------------------------------------

				def test_result():
				"""Test job submission and result retrieval"""
				jobs = bg.BackgroundJobManager()
				j = jobs.new(sleeper)
				j.join()
				assert j.result["interval"] == t_short


				def test_flush():
				"""Test job control"""
				jobs = bg.BackgroundJobManager()
				j = jobs.new(sleeper)
				j.join()
				assert len(jobs.completed) == 1
				assert len(jobs.dead) == 0
				jobs.flush()
				assert len(jobs.completed) == 0


				def test_dead():
				"""Test control of dead jobs"""
				jobs = bg.BackgroundJobManager()
				j = jobs.new(crasher)
				j.join()
				assert len(jobs.completed) == 0
				assert len(jobs.dead) == 1
				jobs.flush()
				assert len(jobs.dead) == 0


				def test_longer():
				"""Test control of longer-running jobs"""
				jobs = bg.BackgroundJobManager()
				# Sleep for long enough for the following two checks to still report the
				# job as running, but not so long that it makes the test suite noticeably
				# slower.
				j = jobs.new(sleeper, 0.1)
				assert len(jobs.running) == 1
				assert len(jobs.completed) == 0
				j.join()
				assert len(jobs.running) == 0
				assert len(jobs.completed) == 1