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 |
|
|
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 |
|
|
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