Show More
@@ -1,70 +1,73 b'' | |||
|
1 | 1 | """Tests for the notebook kernel and session manager.""" |
|
2 | 2 | |
|
3 | 3 | from subprocess import PIPE |
|
4 | 4 | import time |
|
5 | 5 | from unittest import TestCase |
|
6 | 6 | |
|
7 | from IPython.testing import decorators as dec | |
|
8 | ||
|
7 | 9 | from IPython.config.loader import Config |
|
8 | 10 | from IPython.frontend.html.notebook.kernelmanager import MultiKernelManager |
|
9 | 11 | from IPython.zmq.kernelmanager import KernelManager |
|
10 | 12 | |
|
11 | 13 | class TestKernelManager(TestCase): |
|
12 | 14 | |
|
13 | 15 | def _get_tcp_km(self): |
|
14 | 16 | return MultiKernelManager() |
|
15 | 17 | |
|
16 | 18 | def _get_ipc_km(self): |
|
17 | 19 | c = Config() |
|
18 | 20 | c.KernelManager.transport = 'ipc' |
|
19 | 21 | c.KernelManager.ip = 'test' |
|
20 | 22 | km = MultiKernelManager(config=c) |
|
21 | 23 | return km |
|
22 | 24 | |
|
23 | 25 | def _run_lifecycle(self, km): |
|
24 | 26 | kid = km.start_kernel(stdout=PIPE, stderr=PIPE) |
|
25 | 27 | self.assertTrue(kid in km) |
|
26 | 28 | self.assertTrue(kid in km.list_kernel_ids()) |
|
27 | 29 | self.assertEqual(len(km),1) |
|
28 | 30 | km.restart_kernel(kid) |
|
29 | 31 | self.assertTrue(kid in km.list_kernel_ids()) |
|
30 | 32 | # We need a delay here to give the restarting kernel a chance to |
|
31 | 33 | # restart. Otherwise, the interrupt will kill it, causing the test |
|
32 | 34 | # suite to hang. The reason it *hangs* is that the shutdown |
|
33 | 35 | # message for the restart sometimes hasn't been sent to the kernel. |
|
34 | 36 | # Because linger is oo on the shell channel, the context can't |
|
35 | 37 | # close until the message is sent to the kernel, which is not dead. |
|
36 | 38 | time.sleep(1.0) |
|
37 | 39 | km.interrupt_kernel(kid) |
|
38 | 40 | k = km.get_kernel(kid) |
|
39 | 41 | self.assertTrue(isinstance(k, KernelManager)) |
|
40 | 42 | km.shutdown_kernel(kid) |
|
41 | 43 | self.assertTrue(not kid in km) |
|
42 | 44 | |
|
43 | 45 | def _run_cinfo(self, km, transport, ip): |
|
44 | 46 | kid = km.start_kernel(stdout=PIPE, stderr=PIPE) |
|
45 | 47 | k = km.get_kernel(kid) |
|
46 | 48 | cinfo = km.get_connection_info(kid) |
|
47 | 49 | self.assertEqual(transport, cinfo['transport']) |
|
48 | 50 | self.assertEqual(ip, cinfo['ip']) |
|
49 | 51 | self.assertTrue('stdin_port' in cinfo) |
|
50 | 52 | self.assertTrue('iopub_port' in cinfo) |
|
51 | 53 | self.assertTrue('shell_port' in cinfo) |
|
52 | 54 | self.assertTrue('hb_port' in cinfo) |
|
53 | 55 | km.shutdown_kernel(kid) |
|
54 | 56 | |
|
55 | 57 | def test_tcp_lifecycle(self): |
|
56 | 58 | km = self._get_tcp_km() |
|
57 | 59 | self._run_lifecycle(km) |
|
58 | 60 | |
|
61 | @dec.skip_win32 | |
|
59 | 62 | def test_tcp_cinfo(self): |
|
60 | 63 | km = self._get_tcp_km() |
|
61 | 64 | self._run_cinfo(km, 'tcp', '127.0.0.1') |
|
62 | 65 | |
|
63 | 66 | def test_ipc_lifecycle(self): |
|
64 | 67 | km = self._get_ipc_km() |
|
65 | 68 | self._run_lifecycle(km) |
|
66 | 69 | |
|
67 | 70 | def test_ipc_cinfo(self): |
|
68 | 71 | km = self._get_ipc_km() |
|
69 | 72 | self._run_cinfo(km, 'ipc', 'test') |
|
70 | 73 |
@@ -1,45 +1,48 b'' | |||
|
1 | 1 | """Tests for the notebook kernel and session manager""" |
|
2 | 2 | |
|
3 | 3 | from subprocess import PIPE |
|
4 | 4 | import time |
|
5 | 5 | from unittest import TestCase |
|
6 | 6 | |
|
7 | from IPython.testing import decorators as dec | |
|
8 | ||
|
7 | 9 | from IPython.config.loader import Config |
|
8 | 10 | from IPython.zmq.kernelmanager import KernelManager |
|
9 | 11 | |
|
10 | 12 | class TestKernelManager(TestCase): |
|
11 | 13 | |
|
12 | 14 | def _get_tcp_km(self): |
|
13 | 15 | return KernelManager() |
|
14 | 16 | |
|
15 | 17 | def _get_ipc_km(self): |
|
16 | 18 | c = Config() |
|
17 | 19 | c.KernelManager.transport = 'ipc' |
|
18 | 20 | c.KernelManager.ip = 'test' |
|
19 | 21 | km = KernelManager(config=c) |
|
20 | 22 | return km |
|
21 | 23 | |
|
22 | 24 | def _run_lifecycle(self, km): |
|
23 | 25 | km.start_kernel(stdout=PIPE, stderr=PIPE) |
|
24 | 26 | km.start_channels(shell=True, iopub=False, stdin=False, hb=False) |
|
25 | 27 | km.restart_kernel() |
|
26 | 28 | # We need a delay here to give the restarting kernel a chance to |
|
27 | 29 | # restart. Otherwise, the interrupt will kill it, causing the test |
|
28 | 30 | # suite to hang. The reason it *hangs* is that the shutdown |
|
29 | 31 | # message for the restart sometimes hasn't been sent to the kernel. |
|
30 | 32 | # Because linger is oo on the shell channel, the context can't |
|
31 | 33 | # close until the message is sent to the kernel, which is not dead. |
|
32 | 34 | time.sleep(1.0) |
|
33 | 35 | km.interrupt_kernel() |
|
34 | 36 | self.assertTrue(isinstance(km, KernelManager)) |
|
35 | 37 | km.shutdown_kernel() |
|
36 | 38 | km.shell_channel.stop() |
|
37 | 39 | |
|
38 | 40 | def test_tcp_lifecycle(self): |
|
39 | 41 | km = self._get_tcp_km() |
|
40 | 42 | self._run_lifecycle(km) |
|
41 | 43 | |
|
44 | @dec.skip_win32 | |
|
42 | 45 | def testipc_lifecycle(self): |
|
43 | 46 | km = self._get_ipc_km() |
|
44 | 47 | self._run_lifecycle(km) |
|
45 | 48 |
General Comments 0
You need to be logged in to leave comments.
Login now