##// END OF EJS Templates
keep track of the sigint timer, and cancel it on a new prompt
keep track of the sigint timer, and cancel it on a new prompt

File last commit:

r9518:a0863906
r10383:251778c7
Show More
test_multikernelmanager.py
81 lines | 2.7 KiB | text/x-python | PythonLexer
/ IPython / kernel / tests / test_multikernelmanager.py
Brian E. Granger
Refactoring the notebook app to support the new config system.
r4344 """Tests for the notebook kernel and session manager."""
Brian E. Granger
Refactored htmlnotebook session and kernel manager....
r4343
Brian E. Granger
Fixing last few things with the test suite for kernel managers.
r9132 from subprocess import PIPE
Brian E. Granger
Fixing bug in kernelmanager tests that was causing tests to hang....
r9131 import time
Brian E. Granger
Refactored htmlnotebook session and kernel manager....
r4343 from unittest import TestCase
MinRK
skip ipc tests on Windows...
r9156 from IPython.testing import decorators as dec
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 from IPython.config.loader import Config
W. Trevor King
frontend.html.notebook: Use utils.localinterfaces.LOCALHOST
r9252 from IPython.utils.localinterfaces import LOCALHOST
MinRK
move zmq.KernelManagers into IPython.kernel
r9370 from IPython.kernel.kernelmanager import KernelManager
MinRK
move multikernelmanager to IPython.kernel
r9371 from IPython.kernel.multikernelmanager import MultiKernelManager
Brian E. Granger
Refactored htmlnotebook session and kernel manager....
r4343
class TestKernelManager(TestCase):
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 def _get_tcp_km(self):
return MultiKernelManager()
def _get_ipc_km(self):
c = Config()
c.KernelManager.transport = 'ipc'
c.KernelManager.ip = 'test'
km = MultiKernelManager(config=c)
return km
def _run_lifecycle(self, km):
Brian E. Granger
Fixing last few things with the test suite for kernel managers.
r9132 kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
Bradley M. Froehle
s/assert_/assertTrue/
r7876 self.assertTrue(kid in km)
Brian E. Granger
General cleanup of kernelmanger.MultiKernelManager.
r9112 self.assertTrue(kid in km.list_kernel_ids())
Bradley M. Froehle
s/assertEquals/assertEqual/
r7874 self.assertEqual(len(km),1)
Brian E. Granger
Removing return value of restart_kernel....
r9113 km.restart_kernel(kid)
self.assertTrue(kid in km.list_kernel_ids())
Brian E. Granger
Fixing bug in kernelmanager tests that was causing tests to hang....
r9131 # 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)
Brian E. Granger
General cleanup of kernelmanger.MultiKernelManager.
r9112 km.interrupt_kernel(kid)
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 k = km.get_kernel(kid)
self.assertTrue(isinstance(k, KernelManager))
km.shutdown_kernel(kid)
Bradley M. Froehle
s/assert_/assertTrue/
r7876 self.assertTrue(not kid in km)
Brian E. Granger
Refactored htmlnotebook session and kernel manager....
r4343
Brian Granger
Cleaning up MultiKernelManager test and adding linger=0 to shell....
r9124 def _run_cinfo(self, km, transport, ip):
Brian E. Granger
Fixing last few things with the test suite for kernel managers.
r9132 kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 k = km.get_kernel(kid)
cinfo = km.get_connection_info(kid)
Brian Granger
Cleaning up MultiKernelManager test and adding linger=0 to shell....
r9124 self.assertEqual(transport, cinfo['transport'])
self.assertEqual(ip, cinfo['ip'])
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 self.assertTrue('stdin_port' in cinfo)
self.assertTrue('iopub_port' in cinfo)
MinRK
include stream creation in notebook KM tests
r9158 stream = km.create_iopub_stream(kid)
stream.close()
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 self.assertTrue('shell_port' in cinfo)
MinRK
include stream creation in notebook KM tests
r9158 stream = km.create_shell_stream(kid)
stream.close()
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 self.assertTrue('hb_port' in cinfo)
MinRK
include stream creation in notebook KM tests
r9158 stream = km.create_hb_stream(kid)
stream.close()
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 km.shutdown_kernel(kid)
Brian E. Granger
Refactored htmlnotebook session and kernel manager....
r4343
Brian E. Granger
Fixing bug in kernelmanager tests that was causing tests to hang....
r9131 def test_tcp_lifecycle(self):
Brian Granger
Cleaning up MultiKernelManager test and adding linger=0 to shell....
r9124 km = self._get_tcp_km()
self._run_lifecycle(km)
def test_tcp_cinfo(self):
km = self._get_tcp_km()
W. Trevor King
frontend.html.notebook: Use utils.localinterfaces.LOCALHOST
r9252 self._run_cinfo(km, 'tcp', LOCALHOST)
Brian Granger
Cleaning up MultiKernelManager test and adding linger=0 to shell....
r9124
MinRK
skip remaining ipc test on Windows...
r9518 @dec.skip_win32
Brian E. Granger
Fixing bug in kernelmanager tests that was causing tests to hang....
r9131 def test_ipc_lifecycle(self):
Brian Granger
Cleaning up MultiKernelManager test and adding linger=0 to shell....
r9124 km = self._get_ipc_km()
self._run_lifecycle(km)
MinRK
skip remaining ipc test on Windows...
r9518 @dec.skip_win32
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 def test_ipc_cinfo(self):
km = self._get_ipc_km()
Brian Granger
Cleaning up MultiKernelManager test and adding linger=0 to shell....
r9124 self._run_cinfo(km, 'ipc', 'test')
Brian E. Granger
Fixing bug in kernelmanager tests that was causing tests to hang....
r9131