##// END OF EJS Templates
use assemble_output in in-process kernel test...
Min RK -
Show More
@@ -9,6 +9,7 b' import unittest'
9 from IPython.kernel.inprocess.blocking import BlockingInProcessKernelClient
9 from IPython.kernel.inprocess.blocking import BlockingInProcessKernelClient
10 from IPython.kernel.inprocess.manager import InProcessKernelManager
10 from IPython.kernel.inprocess.manager import InProcessKernelManager
11 from IPython.kernel.inprocess.ipkernel import InProcessKernel
11 from IPython.kernel.inprocess.ipkernel import InProcessKernel
12 from IPython.kernel.tests.utils import assemble_output
12 from IPython.testing.decorators import skipif_not_matplotlib
13 from IPython.testing.decorators import skipif_not_matplotlib
13 from IPython.utils.io import capture_output
14 from IPython.utils.io import capture_output
14 from IPython.utils import py3compat
15 from IPython.utils import py3compat
@@ -33,8 +34,8 b' class InProcessKernelTestCase(unittest.TestCase):'
33 """Does %pylab work in the in-process kernel?"""
34 """Does %pylab work in the in-process kernel?"""
34 kc = self.kc
35 kc = self.kc
35 kc.execute('%pylab')
36 kc.execute('%pylab')
36 msg = get_stream_message(kc)
37 out, err = assemble_output(kc.iopub_channel)
37 self.assertIn('matplotlib', msg['content']['text'])
38 self.assertIn('matplotlib', out)
38
39
39 def test_raw_input(self):
40 def test_raw_input(self):
40 """ Does the in-process kernel handle raw_input correctly?
41 """ Does the in-process kernel handle raw_input correctly?
@@ -63,21 +64,6 b' class InProcessKernelTestCase(unittest.TestCase):'
63 kc = BlockingInProcessKernelClient(kernel=kernel)
64 kc = BlockingInProcessKernelClient(kernel=kernel)
64 kernel.frontends.append(kc)
65 kernel.frontends.append(kc)
65 kc.execute('print("bar")')
66 kc.execute('print("bar")')
66 msg = get_stream_message(kc)
67 out, err = assemble_output(kc.iopub_channel)
67 self.assertEqual(msg['content']['text'], 'bar\n')
68 self.assertEqual(out, 'bar\n')
68
69
69 #-----------------------------------------------------------------------------
70 # Utility functions
71 #-----------------------------------------------------------------------------
72
73 def get_stream_message(kernel_client, timeout=5):
74 """ Gets a single stream message synchronously from the sub channel.
75 """
76 while True:
77 msg = kernel_client.get_iopub_msg(timeout=timeout)
78 if msg['header']['msg_type'] == 'stream':
79 return msg
80
81
82 if __name__ == '__main__':
83 unittest.main()
General Comments 0
You need to be logged in to leave comments. Login now