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