##// END OF EJS Templates
Added commit accidently removed...
Added commit accidently removed...

File last commit:

r10329:e506739c
r11403:22992e68
Show More
test_multikernelmanager.py
85 lines | 2.8 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
update imports with new layout
r10284 from IPython.kernel 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):
Brian Granger
Refactoring kernel restarting.
r10282 c = Config()
km = MultiKernelManager(config=c)
return km
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116
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)
Brian E. Granger
Made is_alive a method of KernelManager and MultiKernelManager....
r10275 self.assertTrue(km.is_alive(kid))
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)
Brian E. Granger
Made is_alive a method of KernelManager and MultiKernelManager....
r10275 self.assertTrue(km.is_alive(kid))
Brian E. Granger
Removing return value of restart_kernel....
r9113 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
cleanup boundaries of MultiKernelManager and KernelRestarter classes...
r10295 stream = km.connect_iopub(kid)
MinRK
include stream creation in notebook KM tests
r9158 stream.close()
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 self.assertTrue('shell_port' in cinfo)
MinRK
cleanup boundaries of MultiKernelManager and KernelRestarter classes...
r10295 stream = km.connect_shell(kid)
MinRK
include stream creation in notebook KM tests
r9158 stream.close()
Brian E. Granger
Adding tested ipc support to MultiKernelManager.
r9116 self.assertTrue('hb_port' in cinfo)
MinRK
cleanup boundaries of MultiKernelManager and KernelRestarter classes...
r10295 stream = km.connect_hb(kid)
MinRK
include stream creation in notebook KM tests
r9158 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