##// END OF EJS Templates
skip ipc tests on Windows...
MinRK -
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