Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

TST: Add some unit tests for in-process kernel and kernel manager.

  • Loading branch information...
commit fc75cebc9db79570fa63b3757e78405d0d3f23b0 1 parent 6aaeb87
@epatters epatters authored
View
0  IPython/inprocess/tests/__init__.py
No changes.
View
66 IPython/inprocess/tests/test_kernel.py
@@ -0,0 +1,66 @@
+#-------------------------------------------------------------------------------
+# Copyright (C) 2012 The IPython Development Team
+#
+# Distributed under the terms of the BSD License. The full license is in
+# the file COPYING, distributed as part of this software.
+#-------------------------------------------------------------------------------
+
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
+from __future__ import print_function
+
+# Standard library imports
+import unittest
+
+# Local imports
+from IPython.inprocess.blockingkernelmanager import \
+ BlockingInProcessKernelManager
+from IPython.inprocess.ipkernel import InProcessKernel
+from IPython.utils.io import capture_output
+
+#-----------------------------------------------------------------------------
+# Test case
+#-----------------------------------------------------------------------------
+
+class InProcessKernelTestCase(unittest.TestCase):
+
+ def test_pylab(self):
+ """ Does pylab work in the in-process kernel?
+ """
+ km = BlockingInProcessKernelManager()
+ km.start_kernel()
+ km.shell_channel.execute('%pylab')
+ msg = get_stream_message(km)
+ self.assert_('Welcome to pylab' in msg['content']['data'])
+
+ def test_stdout(self):
+ """ Does the in-process kernel correctly capture IO?
+ """
+ kernel = InProcessKernel()
+
+ with capture_output() as io:
+ kernel.shell.run_cell('print("foo")')
+ self.assertEqual(io.stdout, 'foo\n')
+
+ km = BlockingInProcessKernelManager(kernel=kernel)
+ kernel.frontends.append(km)
+ km.shell_channel.execute('print("bar")')
+ msg = get_stream_message(km)
+ self.assertEqual(msg['content']['data'], 'bar\n')
+
+#-----------------------------------------------------------------------------
+# Utility functions
+#-----------------------------------------------------------------------------
+
+def get_stream_message(kernel_manager, timeout=5):
+ """ Gets a single stream message synchronously from the sub channel.
+ """
+ while True:
+ msg = kernel_manager.sub_channel.get_msg(timeout=timeout)
+ if msg['header']['msg_type'] == 'stream':
+ return msg
+
+
+if __name__ == '__main__':
+ unittest.main()
View
77 IPython/inprocess/tests/test_kernelmanager.py
@@ -0,0 +1,77 @@
+#-------------------------------------------------------------------------------
+# Copyright (C) 2012 The IPython Development Team
+#
+# Distributed under the terms of the BSD License. The full license is in
+# the file COPYING, distributed as part of this software.
+#-------------------------------------------------------------------------------
+
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
+from __future__ import print_function
+
+# Standard library imports
+import unittest
+
+# Local imports
+from IPython.inprocess.blockingkernelmanager import \
+ BlockingInProcessKernelManager
+from IPython.inprocess.ipkernel import InProcessKernel
+
+#-----------------------------------------------------------------------------
+# Test case
+#-----------------------------------------------------------------------------
+
+class InProcessKernelManagerTestCase(unittest.TestCase):
+
+ def test_inteface(self):
+ """ Does the in-process kernel manager implement the basic KM interface?
+ """
+ km = BlockingInProcessKernelManager()
+ self.assert_(not km.channels_running)
+ self.assert_(not km.has_kernel)
+
+ km.start_channels()
+ self.assert_(km.channels_running)
+
+ km.start_kernel()
+ self.assert_(km.has_kernel)
+ self.assert_(km.kernel is not None)
+
+ old_kernel = km.kernel
+ km.restart_kernel()
+ self.assert_(km.kernel is not None)
+ self.assertNotEquals(km.kernel, old_kernel)
+
+ km.shutdown_kernel()
+ self.assert_(not km.has_kernel)
+
+ self.assertRaises(NotImplementedError, km.interrupt_kernel)
+ self.assertRaises(NotImplementedError, km.signal_kernel, 9)
+
+ km.stop_channels()
+ self.assert_(not km.channels_running)
+
+ def test_execute(self):
+ """ Does executing code in an in-process kernel work?
+ """
+ km = BlockingInProcessKernelManager()
+ km.start_kernel()
+ km.shell_channel.execute('foo = 1')
+ self.assertEquals(km.kernel.shell.user_ns['foo'], 1)
+
+ def test_object_info(self):
+ """ Does requesting object information from an in-process kernel work?
+ """
+ km = BlockingInProcessKernelManager()
+ km.start_kernel()
+ km.kernel.shell.user_ns['foo'] = 1
+ km.shell_channel.object_info('foo')
+ msg = km.shell_channel.get_msg()
+ self.assertEquals(msg['header']['msg_type'], 'object_info_reply')
+ self.assertEquals(msg['content']['name'], 'foo')
+ self.assertEquals(msg['content']['type_name'], 'int')
+
+
+if __name__ == '__main__':
+ unittest.main()
Please sign in to comment.
Something went wrong with that request. Please try again.