upstream/ipython Files · IPython/zmq/tests/test_message_spec.py

load header with engine id when engine dies in TaskScheduler...

load header with engine id when engine dies in TaskScheduler This ensures that the metadata dict on the *Client* has the engine_uuid of the engine on which the task failed. It is identical to code elsewhere (Hub, Client) for identifying when engines die.

Matthias BUSSONNIER - - Load All Authors

File last commit:

r5390:c82649ea


                r6068:f8f19148

Download file

             test_message_spec.py
        
                    40 lines
            
             | 1017 B
            
                | text/x-python
            
             |
                PythonLexer
            
             / IPython / zmq / tests / test_message_spec.py
          
                    History
                
                 |
                  Source
                 | Raw
                 |Copy content
                 |Copy permalink

        Fernando Perez
    
Rework messaging to better conform to our spec....

              r2926
            
      """Test suite for our zeromq-based messaging specification.

      """

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

        Matthias BUSSONNIER
    
update copyright to 2011/20xx-2011...

              r5390
            
      #  Copyright (C) 2010-2011  The IPython Development Team

        Fernando Perez
    
Rework messaging to better conform to our spec....

              r2926
            
      #

      #  Distributed under the terms of the BSD License.  The full license is in

      #  the file COPYING.txt, distributed as part of this software.

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

      import sys

      import time

      import nose.tools as nt

      from ..blockingkernelmanager import BlockingKernelManager

      from IPython.utils import io

      def setup():

          global KM

          KM = BlockingKernelManager()

          KM.start_kernel()

          KM.start_channels()

          # Give the kernel a chance to come up.

          time.sleep(1)

      def teardown():

          io.rprint('Entering teardown...')  # dbg

          io.rprint('Stopping channels and kernel...')  # dbg

          KM.stop_channels()

          KM.kill_kernel()

      # Actual tests

      def test_execute():

        MinRK
    
cleanup channel names to match function not socket...

              r3974
            
          KM.shell_channel.execute(code='x=1')

          KM.shell_channel.execute(code='print 1')

        Fernando Perez
    
Rework messaging to better conform to our spec....

              r2926

	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

Fernando Perez Rework messaging to better conform to our spec....	r2926	"""Test suite for our zeromq-based messaging specification.
		"""
		#-----------------------------------------------------------------------------
Matthias BUSSONNIER update copyright to 2011/20xx-2011...	r5390	# Copyright (C) 2010-2011 The IPython Development Team
Fernando Perez Rework messaging to better conform to our spec....	r2926	#
		# Distributed under the terms of the BSD License. The full license is in
		# the file COPYING.txt, distributed as part of this software.
		#-----------------------------------------------------------------------------

		import sys
		import time

		import nose.tools as nt

		from ..blockingkernelmanager import BlockingKernelManager

		from IPython.utils import io

		def setup():
		global KM
		KM = BlockingKernelManager()

		KM.start_kernel()
		KM.start_channels()
		# Give the kernel a chance to come up.
		time.sleep(1)

		def teardown():
		io.rprint('Entering teardown...') # dbg
		io.rprint('Stopping channels and kernel...') # dbg
		KM.stop_channels()
		KM.kill_kernel()


		# Actual tests

		def test_execute():
MinRK cleanup channel names to match function not socket...	r3974	KM.shell_channel.execute(code='x=1')
		KM.shell_channel.execute(code='print 1')
Fernando Perez Rework messaging to better conform to our spec....	r2926