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