##// END OF EJS Templates
Fixing last few things with the test suite for kernel managers.
Brian E. Granger -
Show More
@@ -79,6 +79,11 b' class MultiKernelManager(LoggingConfigurable):'
79
79
80 The caller can pick a kernel_id by passing one in as a keyword arg,
80 The caller can pick a kernel_id by passing one in as a keyword arg,
81 otherwise one will be picked using a uuid.
81 otherwise one will be picked using a uuid.
82
83 To silence the kernel's stdout/stderr, call this using::
84
85 km.start_kernel(stdout=PIPE, stderr=PIPE)
86
82 """
87 """
83 kernel_id = kwargs.pop('kernel_id', unicode(uuid.uuid4()))
88 kernel_id = kwargs.pop('kernel_id', unicode(uuid.uuid4()))
84 if kernel_id in self:
89 if kernel_id in self:
@@ -1,5 +1,6 b''
1 """Tests for the notebook kernel and session manager."""
1 """Tests for the notebook kernel and session manager."""
2
2
3 from subprocess import PIPE
3 import time
4 import time
4 from unittest import TestCase
5 from unittest import TestCase
5
6
@@ -20,7 +21,7 b' class TestKernelManager(TestCase):'
20 return km
21 return km
21
22
22 def _run_lifecycle(self, km):
23 def _run_lifecycle(self, km):
23 kid = km.start_kernel()
24 kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
24 self.assertTrue(kid in km)
25 self.assertTrue(kid in km)
25 self.assertTrue(kid in km.list_kernel_ids())
26 self.assertTrue(kid in km.list_kernel_ids())
26 self.assertEqual(len(km),1)
27 self.assertEqual(len(km),1)
@@ -40,7 +41,7 b' class TestKernelManager(TestCase):'
40 self.assertTrue(not kid in km)
41 self.assertTrue(not kid in km)
41
42
42 def _run_cinfo(self, km, transport, ip):
43 def _run_cinfo(self, km, transport, ip):
43 kid = km.start_kernel()
44 kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
44 k = km.get_kernel(kid)
45 k = km.get_kernel(kid)
45 cinfo = km.get_connection_info(kid)
46 cinfo = km.get_connection_info(kid)
46 self.assertEqual(transport, cinfo['transport'])
47 self.assertEqual(transport, cinfo['transport'])
@@ -1,5 +1,6 b''
1 """Tests for the notebook kernel and session manager."""
1 """Tests for the notebook kernel and session manager."""
2
2
3 from subprocess import PIPE
3 import time
4 import time
4 from unittest import TestCase
5 from unittest import TestCase
5
6
@@ -19,7 +20,7 b' class TestKernelManager(TestCase):'
19 return km
20 return km
20
21
21 def _run_lifecycle(self, km):
22 def _run_lifecycle(self, km):
22 km.start_kernel()
23 km.start_kernel(stdout=PIPE, stderr=PIPE)
23 km.start_channels(shell=True, iopub=False, stdin=False, hb=False)
24 km.start_channels(shell=True, iopub=False, stdin=False, hb=False)
24 km.restart_kernel()
25 km.restart_kernel()
25 # We need a delay here to give the restarting kernel a chance to
26 # We need a delay here to give the restarting kernel a chance to
@@ -28,7 +29,7 b' class TestKernelManager(TestCase):'
28 # message for the restart sometimes hasn't been sent to the kernel.
29 # message for the restart sometimes hasn't been sent to the kernel.
29 # Because linger is oo on the shell channel, the context can't
30 # Because linger is oo on the shell channel, the context can't
30 # close until the message is sent to the kernel, which is not dead.
31 # close until the message is sent to the kernel, which is not dead.
31 time.sleep()
32 time.sleep(1.0)
32 km.interrupt_kernel()
33 km.interrupt_kernel()
33 self.assertTrue(isinstance(km, KernelManager))
34 self.assertTrue(isinstance(km, KernelManager))
34 km.shutdown_kernel()
35 km.shutdown_kernel()
General Comments 0
You need to be logged in to leave comments. Login now