##// END OF EJS Templates
skip remaining ipc test on Windows...
MinRK -
Show More
@@ -1,80 +1,81 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 from subprocess import PIPE
4 import time
4 import time
5 from unittest import TestCase
5 from unittest import TestCase
6
6
7 from IPython.testing import decorators as dec
7 from IPython.testing import decorators as dec
8
8
9 from IPython.config.loader import Config
9 from IPython.config.loader import Config
10 from IPython.utils.localinterfaces import LOCALHOST
10 from IPython.utils.localinterfaces import LOCALHOST
11 from IPython.kernel.kernelmanager import KernelManager
11 from IPython.kernel.kernelmanager import KernelManager
12 from IPython.kernel.multikernelmanager import MultiKernelManager
12 from IPython.kernel.multikernelmanager import MultiKernelManager
13
13
14 class TestKernelManager(TestCase):
14 class TestKernelManager(TestCase):
15
15
16 def _get_tcp_km(self):
16 def _get_tcp_km(self):
17 return MultiKernelManager()
17 return MultiKernelManager()
18
18
19 def _get_ipc_km(self):
19 def _get_ipc_km(self):
20 c = Config()
20 c = Config()
21 c.KernelManager.transport = 'ipc'
21 c.KernelManager.transport = 'ipc'
22 c.KernelManager.ip = 'test'
22 c.KernelManager.ip = 'test'
23 km = MultiKernelManager(config=c)
23 km = MultiKernelManager(config=c)
24 return km
24 return km
25
25
26 def _run_lifecycle(self, km):
26 def _run_lifecycle(self, km):
27 kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
27 kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
28 self.assertTrue(kid in km)
28 self.assertTrue(kid in km)
29 self.assertTrue(kid in km.list_kernel_ids())
29 self.assertTrue(kid in km.list_kernel_ids())
30 self.assertEqual(len(km),1)
30 self.assertEqual(len(km),1)
31 km.restart_kernel(kid)
31 km.restart_kernel(kid)
32 self.assertTrue(kid in km.list_kernel_ids())
32 self.assertTrue(kid in km.list_kernel_ids())
33 # We need a delay here to give the restarting kernel a chance to
33 # We need a delay here to give the restarting kernel a chance to
34 # restart. Otherwise, the interrupt will kill it, causing the test
34 # restart. Otherwise, the interrupt will kill it, causing the test
35 # suite to hang. The reason it *hangs* is that the shutdown
35 # suite to hang. The reason it *hangs* is that the shutdown
36 # message for the restart sometimes hasn't been sent to the kernel.
36 # message for the restart sometimes hasn't been sent to the kernel.
37 # Because linger is oo on the shell channel, the context can't
37 # Because linger is oo on the shell channel, the context can't
38 # close until the message is sent to the kernel, which is not dead.
38 # close until the message is sent to the kernel, which is not dead.
39 time.sleep(1.0)
39 time.sleep(1.0)
40 km.interrupt_kernel(kid)
40 km.interrupt_kernel(kid)
41 k = km.get_kernel(kid)
41 k = km.get_kernel(kid)
42 self.assertTrue(isinstance(k, KernelManager))
42 self.assertTrue(isinstance(k, KernelManager))
43 km.shutdown_kernel(kid)
43 km.shutdown_kernel(kid)
44 self.assertTrue(not kid in km)
44 self.assertTrue(not kid in km)
45
45
46 def _run_cinfo(self, km, transport, ip):
46 def _run_cinfo(self, km, transport, ip):
47 kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
47 kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
48 k = km.get_kernel(kid)
48 k = km.get_kernel(kid)
49 cinfo = km.get_connection_info(kid)
49 cinfo = km.get_connection_info(kid)
50 self.assertEqual(transport, cinfo['transport'])
50 self.assertEqual(transport, cinfo['transport'])
51 self.assertEqual(ip, cinfo['ip'])
51 self.assertEqual(ip, cinfo['ip'])
52 self.assertTrue('stdin_port' in cinfo)
52 self.assertTrue('stdin_port' in cinfo)
53 self.assertTrue('iopub_port' in cinfo)
53 self.assertTrue('iopub_port' in cinfo)
54 stream = km.create_iopub_stream(kid)
54 stream = km.create_iopub_stream(kid)
55 stream.close()
55 stream.close()
56 self.assertTrue('shell_port' in cinfo)
56 self.assertTrue('shell_port' in cinfo)
57 stream = km.create_shell_stream(kid)
57 stream = km.create_shell_stream(kid)
58 stream.close()
58 stream.close()
59 self.assertTrue('hb_port' in cinfo)
59 self.assertTrue('hb_port' in cinfo)
60 stream = km.create_hb_stream(kid)
60 stream = km.create_hb_stream(kid)
61 stream.close()
61 stream.close()
62 km.shutdown_kernel(kid)
62 km.shutdown_kernel(kid)
63
63
64 def test_tcp_lifecycle(self):
64 def test_tcp_lifecycle(self):
65 km = self._get_tcp_km()
65 km = self._get_tcp_km()
66 self._run_lifecycle(km)
66 self._run_lifecycle(km)
67
67
68 @dec.skip_win32
69 def test_tcp_cinfo(self):
68 def test_tcp_cinfo(self):
70 km = self._get_tcp_km()
69 km = self._get_tcp_km()
71 self._run_cinfo(km, 'tcp', LOCALHOST)
70 self._run_cinfo(km, 'tcp', LOCALHOST)
72
71
72 @dec.skip_win32
73 def test_ipc_lifecycle(self):
73 def test_ipc_lifecycle(self):
74 km = self._get_ipc_km()
74 km = self._get_ipc_km()
75 self._run_lifecycle(km)
75 self._run_lifecycle(km)
76
76
77 @dec.skip_win32
77 def test_ipc_cinfo(self):
78 def test_ipc_cinfo(self):
78 km = self._get_ipc_km()
79 km = self._get_ipc_km()
79 self._run_cinfo(km, 'ipc', 'test')
80 self._run_cinfo(km, 'ipc', 'test')
80
81
General Comments 0
You need to be logged in to leave comments. Login now