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

Merge pull request from ellisonbg/kernelid...

Merge pull request from ellisonbg/kernelid General cleanup of kernel manager code. This does some general cleanup of MultiKernelManager to better reflect how it is actually being used. Sometimes there is a need to create kernel_id's elsewhere in code. This minor change allows a kernel_id to be created outside of the MultiKernelManager and passed in as a kwarg. An exception is raised if the id is already used. Tasks: - [x] Fix cleanup of ipc files. - [x] Allow kernel_id to be passed to MultiKernelManager.start_kernel. - [x] Add ipc support to MultiKernelManager. - [x] Add more tests for MultiKernelManager. - [x] Rename sub channel to iopub channel everywhere. - [x] Use consistent naming for all channel classes in zmq, inprocess and qt. - [x] Move BlockingChannelMixin to zmq.blockingkernelmanager. - [x] Create ABC for KernelManager. - [x] Make the InProcessKernelManager a Configurable. - [x] Cleanup docstrings in ABCs. - [x] Add tests for KernelManager. - [x] Check over MultiKernelManager. - [x] Make KernelManager,kill_kernel private in ABC and implementations. - [x] Find bug that is causing the kernel manager tests to hang unless the shell channel linger is set to 0. - [x] Decide about critical logging in ipkernel. - [x] Debug lack of stderr redirect in tests.

Brian E. Granger - - Load All Authors

File last commit:

r9151:a8f2ab2a


                r9155:ef8974d5

Download file

             test_kernelmanager.py
        
                    45 lines
            
             | 1.5 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / IPython / zmq / tests / test_kernelmanager.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      """Tests for the notebook kernel and session manager"""

      from subprocess import PIPE

      import time

      from unittest import TestCase

      from IPython.config.loader import Config

      from IPython.zmq.kernelmanager import KernelManager

      class TestKernelManager(TestCase):

          def _get_tcp_km(self):

              return KernelManager()

          def _get_ipc_km(self):

              c = Config()

              c.KernelManager.transport = 'ipc'

              c.KernelManager.ip = 'test'

              km = KernelManager(config=c)

              return km

          def _run_lifecycle(self, km):

              km.start_kernel(stdout=PIPE, stderr=PIPE)

              km.start_channels(shell=True, iopub=False, stdin=False, hb=False)

              km.restart_kernel()

              # We need a delay here to give the restarting kernel a chance to

              # restart. Otherwise, the interrupt will kill it, causing the test

              # suite to hang. The reason it *hangs* is that the shutdown

              # message for the restart sometimes hasn't been sent to the kernel.

              # Because linger is oo on the shell channel, the context can't

              # close until the message is sent to the kernel, which is not dead.

              time.sleep(1.0)

              km.interrupt_kernel()

              self.assertTrue(isinstance(km, KernelManager))

              km.shutdown_kernel()

              km.shell_channel.stop()

          def test_tcp_lifecycle(self):

              km = self._get_tcp_km()

              self._run_lifecycle(km)

          def testipc_lifecycle(self):

              km = self._get_ipc_km()

              self._run_lifecycle(km)

	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 the notebook kernel and session manager"""

				from subprocess import PIPE
				import time
				from unittest import TestCase

				from IPython.config.loader import Config
				from IPython.zmq.kernelmanager import KernelManager

				class TestKernelManager(TestCase):

				def _get_tcp_km(self):
				return KernelManager()

				def _get_ipc_km(self):
				c = Config()
				c.KernelManager.transport = 'ipc'
				c.KernelManager.ip = 'test'
				km = KernelManager(config=c)
				return km

				def _run_lifecycle(self, km):
				km.start_kernel(stdout=PIPE, stderr=PIPE)
				km.start_channels(shell=True, iopub=False, stdin=False, hb=False)
				km.restart_kernel()
				# We need a delay here to give the restarting kernel a chance to
				# restart. Otherwise, the interrupt will kill it, causing the test
				# suite to hang. The reason it hangs is that the shutdown
				# message for the restart sometimes hasn't been sent to the kernel.
				# Because linger is oo on the shell channel, the context can't
				# close until the message is sent to the kernel, which is not dead.
				time.sleep(1.0)
				km.interrupt_kernel()
				self.assertTrue(isinstance(km, KernelManager))
				km.shutdown_kernel()
				km.shell_channel.stop()

				def test_tcp_lifecycle(self):
				km = self._get_tcp_km()
				self._run_lifecycle(km)

				def testipc_lifecycle(self):
				km = self._get_ipc_km()
				self._run_lifecycle(km)