test_multikernelmanager.py
86 lines
| 2.7 KiB
| text/x-python
|
PythonLexer
|
r4344 | """Tests for the notebook kernel and session manager.""" | ||
|
r4343 | |||
|
r9132 | from subprocess import PIPE | ||
|
r9131 | import time | ||
|
r4343 | from unittest import TestCase | ||
|
r9156 | from IPython.testing import decorators as dec | ||
|
r9116 | from IPython.config.loader import Config | ||
|
r21098 | from ..localinterfaces import localhost | ||
|
r20951 | from jupyter_client import KernelManager | ||
from jupyter_client.multikernelmanager import MultiKernelManager | ||||
|
r4343 | |||
class TestKernelManager(TestCase): | ||||
|
r9116 | def _get_tcp_km(self): | ||
|
r10282 | c = Config() | ||
km = MultiKernelManager(config=c) | ||||
return km | ||||
|
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): | ||||
|
r9132 | kid = km.start_kernel(stdout=PIPE, stderr=PIPE) | ||
|
r10275 | self.assertTrue(km.is_alive(kid)) | ||
|
r7876 | self.assertTrue(kid in km) | ||
|
r9112 | self.assertTrue(kid in km.list_kernel_ids()) | ||
|
r7874 | self.assertEqual(len(km),1) | ||
|
r12416 | km.restart_kernel(kid, now=True) | ||
|
r10275 | self.assertTrue(km.is_alive(kid)) | ||
|
r9113 | self.assertTrue(kid in km.list_kernel_ids()) | ||
|
r9112 | km.interrupt_kernel(kid) | ||
|
r9116 | k = km.get_kernel(kid) | ||
self.assertTrue(isinstance(k, KernelManager)) | ||||
|
r12416 | km.shutdown_kernel(kid, now=True) | ||
|
r7876 | self.assertTrue(not kid in km) | ||
|
r4343 | |||
|
r9124 | def _run_cinfo(self, km, transport, ip): | ||
|
r9132 | kid = km.start_kernel(stdout=PIPE, stderr=PIPE) | ||
|
r9116 | k = km.get_kernel(kid) | ||
cinfo = km.get_connection_info(kid) | ||||
|
r9124 | self.assertEqual(transport, cinfo['transport']) | ||
self.assertEqual(ip, cinfo['ip']) | ||||
|
r9116 | self.assertTrue('stdin_port' in cinfo) | ||
self.assertTrue('iopub_port' in cinfo) | ||||
|
r10295 | stream = km.connect_iopub(kid) | ||
|
r9158 | stream.close() | ||
|
r9116 | self.assertTrue('shell_port' in cinfo) | ||
|
r10295 | stream = km.connect_shell(kid) | ||
|
r9158 | stream.close() | ||
|
r9116 | self.assertTrue('hb_port' in cinfo) | ||
|
r10295 | stream = km.connect_hb(kid) | ||
|
r9158 | stream.close() | ||
|
r12416 | km.shutdown_kernel(kid, now=True) | ||
|
r4343 | |||
|
r9131 | def test_tcp_lifecycle(self): | ||
|
r9124 | km = self._get_tcp_km() | ||
self._run_lifecycle(km) | ||||
|
r20949 | |||
|
r20670 | def test_shutdown_all(self): | ||
km = self._get_tcp_km() | ||||
kid = km.start_kernel(stdout=PIPE, stderr=PIPE) | ||||
self.assertIn(kid, km) | ||||
km.shutdown_all() | ||||
self.assertNotIn(kid, km) | ||||
# shutdown again is okay, because we have no kernels | ||||
km.shutdown_all() | ||||
|
r20949 | |||
|
r9124 | def test_tcp_cinfo(self): | ||
km = self._get_tcp_km() | ||||
|
r12591 | self._run_cinfo(km, 'tcp', localhost()) | ||
|
r9124 | |||
|
r9518 | @dec.skip_win32 | ||
|
r9131 | def test_ipc_lifecycle(self): | ||
|
r9124 | km = self._get_ipc_km() | ||
self._run_lifecycle(km) | ||||
|
r20949 | |||
|
r9518 | @dec.skip_win32 | ||
|
r9116 | def test_ipc_cinfo(self): | ||
km = self._get_ipc_km() | ||||
|
r9124 | self._run_cinfo(km, 'ipc', 'test') | ||