##// END OF EJS Templates
Backport PR #6556: Use some more informative asserts in inprocess kernel tests...
Thomas Kluyver -
Show More
@@ -1,95 +1,95 b''
1 #-------------------------------------------------------------------------------
1 #-------------------------------------------------------------------------------
2 # Copyright (C) 2012 The IPython Development Team
2 # Copyright (C) 2012 The IPython Development Team
3 #
3 #
4 # Distributed under the terms of the BSD License. The full license is in
4 # Distributed under the terms of the BSD License. The full license is in
5 # the file COPYING, distributed as part of this software.
5 # the file COPYING, distributed as part of this software.
6 #-------------------------------------------------------------------------------
6 #-------------------------------------------------------------------------------
7
7
8 #-----------------------------------------------------------------------------
8 #-----------------------------------------------------------------------------
9 # Imports
9 # Imports
10 #-----------------------------------------------------------------------------
10 #-----------------------------------------------------------------------------
11 from __future__ import print_function
11 from __future__ import print_function
12
12
13 # Standard library imports
13 # Standard library imports
14 import sys
14 import sys
15 import unittest
15 import unittest
16
16
17 # Local imports
17 # Local imports
18 from IPython.kernel.inprocess.blocking import BlockingInProcessKernelClient
18 from IPython.kernel.inprocess.blocking import BlockingInProcessKernelClient
19 from IPython.kernel.inprocess.manager import InProcessKernelManager
19 from IPython.kernel.inprocess.manager import InProcessKernelManager
20 from IPython.kernel.inprocess.ipkernel import InProcessKernel
20 from IPython.kernel.inprocess.ipkernel import InProcessKernel
21 from IPython.testing.decorators import skipif_not_matplotlib
21 from IPython.testing.decorators import skipif_not_matplotlib
22 from IPython.utils.io import capture_output
22 from IPython.utils.io import capture_output
23 from IPython.utils import py3compat
23 from IPython.utils import py3compat
24
24
25 if py3compat.PY3:
25 if py3compat.PY3:
26 from io import StringIO
26 from io import StringIO
27 else:
27 else:
28 from StringIO import StringIO
28 from StringIO import StringIO
29
29
30 #-----------------------------------------------------------------------------
30 #-----------------------------------------------------------------------------
31 # Test case
31 # Test case
32 #-----------------------------------------------------------------------------
32 #-----------------------------------------------------------------------------
33
33
34 class InProcessKernelTestCase(unittest.TestCase):
34 class InProcessKernelTestCase(unittest.TestCase):
35
35
36 def setUp(self):
36 def setUp(self):
37 self.km = InProcessKernelManager()
37 self.km = InProcessKernelManager()
38 self.km.start_kernel()
38 self.km.start_kernel()
39 self.kc = BlockingInProcessKernelClient(kernel=self.km.kernel)
39 self.kc = BlockingInProcessKernelClient(kernel=self.km.kernel)
40 self.kc.start_channels()
40 self.kc.start_channels()
41
41
42 @skipif_not_matplotlib
42 @skipif_not_matplotlib
43 def test_pylab(self):
43 def test_pylab(self):
44 """ Does pylab work in the in-process kernel?
44 """ Does pylab work in the in-process kernel?
45 """
45 """
46 kc = self.kc
46 kc = self.kc
47 kc.execute('%pylab')
47 kc.execute('%pylab')
48 msg = get_stream_message(kc)
48 msg = get_stream_message(kc)
49 self.assert_('matplotlib' in msg['content']['data'])
49 self.assertIn('matplotlib', msg['content']['data'])
50
50
51 def test_raw_input(self):
51 def test_raw_input(self):
52 """ Does the in-process kernel handle raw_input correctly?
52 """ Does the in-process kernel handle raw_input correctly?
53 """
53 """
54 io = StringIO('foobar\n')
54 io = StringIO('foobar\n')
55 sys_stdin = sys.stdin
55 sys_stdin = sys.stdin
56 sys.stdin = io
56 sys.stdin = io
57 try:
57 try:
58 if py3compat.PY3:
58 if py3compat.PY3:
59 self.kc.execute('x = input()')
59 self.kc.execute('x = input()')
60 else:
60 else:
61 self.kc.execute('x = raw_input()')
61 self.kc.execute('x = raw_input()')
62 finally:
62 finally:
63 sys.stdin = sys_stdin
63 sys.stdin = sys_stdin
64 self.assertEqual(self.km.kernel.shell.user_ns.get('x'), 'foobar')
64 self.assertEqual(self.km.kernel.shell.user_ns.get('x'), 'foobar')
65
65
66 def test_stdout(self):
66 def test_stdout(self):
67 """ Does the in-process kernel correctly capture IO?
67 """ Does the in-process kernel correctly capture IO?
68 """
68 """
69 kernel = InProcessKernel()
69 kernel = InProcessKernel()
70
70
71 with capture_output() as io:
71 with capture_output() as io:
72 kernel.shell.run_cell('print("foo")')
72 kernel.shell.run_cell('print("foo")')
73 self.assertEqual(io.stdout, 'foo\n')
73 self.assertEqual(io.stdout, 'foo\n')
74
74
75 kc = BlockingInProcessKernelClient(kernel=kernel)
75 kc = BlockingInProcessKernelClient(kernel=kernel)
76 kernel.frontends.append(kc)
76 kernel.frontends.append(kc)
77 kc.shell_channel.execute('print("bar")')
77 kc.shell_channel.execute('print("bar")')
78 msg = get_stream_message(kc)
78 msg = get_stream_message(kc)
79 self.assertEqual(msg['content']['data'], 'bar\n')
79 self.assertEqual(msg['content']['data'], 'bar\n')
80
80
81 #-----------------------------------------------------------------------------
81 #-----------------------------------------------------------------------------
82 # Utility functions
82 # Utility functions
83 #-----------------------------------------------------------------------------
83 #-----------------------------------------------------------------------------
84
84
85 def get_stream_message(kernel_client, timeout=5):
85 def get_stream_message(kernel_client, timeout=5):
86 """ Gets a single stream message synchronously from the sub channel.
86 """ Gets a single stream message synchronously from the sub channel.
87 """
87 """
88 while True:
88 while True:
89 msg = kernel_client.get_iopub_msg(timeout=timeout)
89 msg = kernel_client.get_iopub_msg(timeout=timeout)
90 if msg['header']['msg_type'] == 'stream':
90 if msg['header']['msg_type'] == 'stream':
91 return msg
91 return msg
92
92
93
93
94 if __name__ == '__main__':
94 if __name__ == '__main__':
95 unittest.main()
95 unittest.main()
@@ -1,111 +1,111 b''
1 #-------------------------------------------------------------------------------
1 #-------------------------------------------------------------------------------
2 # Copyright (C) 2012 The IPython Development Team
2 # Copyright (C) 2012 The IPython Development Team
3 #
3 #
4 # Distributed under the terms of the BSD License. The full license is in
4 # Distributed under the terms of the BSD License. The full license is in
5 # the file COPYING, distributed as part of this software.
5 # the file COPYING, distributed as part of this software.
6 #-------------------------------------------------------------------------------
6 #-------------------------------------------------------------------------------
7
7
8 #-----------------------------------------------------------------------------
8 #-----------------------------------------------------------------------------
9 # Imports
9 # Imports
10 #-----------------------------------------------------------------------------
10 #-----------------------------------------------------------------------------
11 from __future__ import print_function
11 from __future__ import print_function
12
12
13 # Standard library imports
13 # Standard library imports
14 import unittest
14 import unittest
15
15
16 # Local imports
16 # Local imports
17 from IPython.kernel.inprocess.blocking import BlockingInProcessKernelClient
17 from IPython.kernel.inprocess.blocking import BlockingInProcessKernelClient
18 from IPython.kernel.inprocess.manager import InProcessKernelManager
18 from IPython.kernel.inprocess.manager import InProcessKernelManager
19
19
20 #-----------------------------------------------------------------------------
20 #-----------------------------------------------------------------------------
21 # Test case
21 # Test case
22 #-----------------------------------------------------------------------------
22 #-----------------------------------------------------------------------------
23
23
24 class InProcessKernelManagerTestCase(unittest.TestCase):
24 class InProcessKernelManagerTestCase(unittest.TestCase):
25
25
26 def test_interface(self):
26 def test_interface(self):
27 """ Does the in-process kernel manager implement the basic KM interface?
27 """ Does the in-process kernel manager implement the basic KM interface?
28 """
28 """
29 km = InProcessKernelManager()
29 km = InProcessKernelManager()
30 self.assert_(not km.has_kernel)
30 self.assert_(not km.has_kernel)
31
31
32 km.start_kernel()
32 km.start_kernel()
33 self.assert_(km.has_kernel)
33 self.assert_(km.has_kernel)
34 self.assert_(km.kernel is not None)
34 self.assert_(km.kernel is not None)
35
35
36 kc = BlockingInProcessKernelClient(kernel=km.kernel)
36 kc = BlockingInProcessKernelClient(kernel=km.kernel)
37 self.assert_(not kc.channels_running)
37 self.assert_(not kc.channels_running)
38
38
39 kc.start_channels()
39 kc.start_channels()
40 self.assert_(kc.channels_running)
40 self.assert_(kc.channels_running)
41
41
42 old_kernel = km.kernel
42 old_kernel = km.kernel
43 km.restart_kernel()
43 km.restart_kernel()
44 self.assert_(km.kernel is not None)
44 self.assertIsNotNone(km.kernel)
45 self.assertNotEquals(km.kernel, old_kernel)
45 self.assertNotEquals(km.kernel, old_kernel)
46
46
47 km.shutdown_kernel()
47 km.shutdown_kernel()
48 self.assert_(not km.has_kernel)
48 self.assert_(not km.has_kernel)
49
49
50 self.assertRaises(NotImplementedError, km.interrupt_kernel)
50 self.assertRaises(NotImplementedError, km.interrupt_kernel)
51 self.assertRaises(NotImplementedError, km.signal_kernel, 9)
51 self.assertRaises(NotImplementedError, km.signal_kernel, 9)
52
52
53 kc.stop_channels()
53 kc.stop_channels()
54 self.assert_(not kc.channels_running)
54 self.assert_(not kc.channels_running)
55
55
56 def test_execute(self):
56 def test_execute(self):
57 """ Does executing code in an in-process kernel work?
57 """ Does executing code in an in-process kernel work?
58 """
58 """
59 km = InProcessKernelManager()
59 km = InProcessKernelManager()
60 km.start_kernel()
60 km.start_kernel()
61 kc = BlockingInProcessKernelClient(kernel=km.kernel)
61 kc = BlockingInProcessKernelClient(kernel=km.kernel)
62 kc.start_channels()
62 kc.start_channels()
63 kc.execute('foo = 1')
63 kc.execute('foo = 1')
64 self.assertEquals(km.kernel.shell.user_ns['foo'], 1)
64 self.assertEquals(km.kernel.shell.user_ns['foo'], 1)
65
65
66 def test_complete(self):
66 def test_complete(self):
67 """ Does requesting completion from an in-process kernel work?
67 """ Does requesting completion from an in-process kernel work?
68 """
68 """
69 km = InProcessKernelManager()
69 km = InProcessKernelManager()
70 km.start_kernel()
70 km.start_kernel()
71 kc = BlockingInProcessKernelClient(kernel=km.kernel)
71 kc = BlockingInProcessKernelClient(kernel=km.kernel)
72 kc.start_channels()
72 kc.start_channels()
73 km.kernel.shell.push({'my_bar': 0, 'my_baz': 1})
73 km.kernel.shell.push({'my_bar': 0, 'my_baz': 1})
74 kc.complete('my_ba', 'my_ba', 5)
74 kc.complete('my_ba', 'my_ba', 5)
75 msg = kc.get_shell_msg()
75 msg = kc.get_shell_msg()
76 self.assertEqual(msg['header']['msg_type'], 'complete_reply')
76 self.assertEqual(msg['header']['msg_type'], 'complete_reply')
77 self.assertEqual(sorted(msg['content']['matches']),
77 self.assertEqual(sorted(msg['content']['matches']),
78 ['my_bar', 'my_baz'])
78 ['my_bar', 'my_baz'])
79
79
80 def test_object_info(self):
80 def test_object_info(self):
81 """ Does requesting object information from an in-process kernel work?
81 """ Does requesting object information from an in-process kernel work?
82 """
82 """
83 km = InProcessKernelManager()
83 km = InProcessKernelManager()
84 km.start_kernel()
84 km.start_kernel()
85 kc = BlockingInProcessKernelClient(kernel=km.kernel)
85 kc = BlockingInProcessKernelClient(kernel=km.kernel)
86 kc.start_channels()
86 kc.start_channels()
87 km.kernel.shell.user_ns['foo'] = 1
87 km.kernel.shell.user_ns['foo'] = 1
88 kc.object_info('foo')
88 kc.object_info('foo')
89 msg = kc.get_shell_msg()
89 msg = kc.get_shell_msg()
90 self.assertEquals(msg['header']['msg_type'], 'object_info_reply')
90 self.assertEquals(msg['header']['msg_type'], 'object_info_reply')
91 self.assertEquals(msg['content']['name'], 'foo')
91 self.assertEquals(msg['content']['name'], 'foo')
92 self.assertEquals(msg['content']['type_name'], 'int')
92 self.assertEquals(msg['content']['type_name'], 'int')
93
93
94 def test_history(self):
94 def test_history(self):
95 """ Does requesting history from an in-process kernel work?
95 """ Does requesting history from an in-process kernel work?
96 """
96 """
97 km = InProcessKernelManager()
97 km = InProcessKernelManager()
98 km.start_kernel()
98 km.start_kernel()
99 kc = BlockingInProcessKernelClient(kernel=km.kernel)
99 kc = BlockingInProcessKernelClient(kernel=km.kernel)
100 kc.start_channels()
100 kc.start_channels()
101 kc.execute('%who')
101 kc.execute('%who')
102 kc.history(hist_access_type='tail', n=1)
102 kc.history(hist_access_type='tail', n=1)
103 msg = kc.shell_channel.get_msgs()[-1]
103 msg = kc.shell_channel.get_msgs()[-1]
104 self.assertEquals(msg['header']['msg_type'], 'history_reply')
104 self.assertEquals(msg['header']['msg_type'], 'history_reply')
105 history = msg['content']['history']
105 history = msg['content']['history']
106 self.assertEquals(len(history), 1)
106 self.assertEquals(len(history), 1)
107 self.assertEquals(history[0][2], '%who')
107 self.assertEquals(history[0][2], '%who')
108
108
109
109
110 if __name__ == '__main__':
110 if __name__ == '__main__':
111 unittest.main()
111 unittest.main()
General Comments 0
You need to be logged in to leave comments. Login now