Expose IPython machinery for testing code completeness

IPython/core/inputsplitter.py

@@ -248,6 +248,22 b' class InputSplitter(object):'
         self.reset()
         return out
 
+    def is_complete(self, source):
+        """Return whether a block of code is ready to execute, or should be continued
+        
+        This is a non-stateful API, and will reset the state of this InputSplitter.
+        """
+        self.reset()
+        try:
+            self.push(source)
+            return not self.push_accepts_more()
+        except SyntaxError:
+            # Transformers in IPythonInputSplitter can raise SyntaxError,
+            # which push() will not catch.
+            return True
+        finally:
+            self.reset()
+
     def push(self, lines):
         """Push one or more lines of input.
 

IPython/core/tests/test_inputsplitter.py

@@ -355,6 +355,13 b' class InputSplitterTestCase(unittest.TestCase):'
         isp.push(r"(1 \ ")
         self.assertFalse(isp.push_accepts_more())
 
+    def test_is_complete(self):
+        isp = self.isp
+        assert isp.is_complete("a = 1")
+        assert not isp.is_complete("for a in range(5):")
+        assert isp.is_complete("raise = 2")  # SyntaxError should mean complete
+        assert not isp.is_complete("a = [1,\n2,")
+
 class InteractiveLoopTestCase(unittest.TestCase):
     """Tests for an interactive loop like a python shell.
     """

IPython/kernel/tests/test_kernel.py

@@ -205,3 +205,19 b' def test_help_output():'
     """ipython kernel --help-all works"""
     tt.help_all_output_test('kernel')
 
+def test_is_complete():
+    with kernel() as kc:
+        # There are more test cases for this in core - here we just check
+        # that the kernel exposes the interface correctly.
+        kc.is_complete('2+2')
+        reply = kc.get_shell_msg(block=True, timeout=TIMEOUT)
+        assert reply['content']['complete']
+
+        # SyntaxError should mean it's complete        
+        kc.is_complete('raise = 2')
+        reply = kc.get_shell_msg(block=True, timeout=TIMEOUT)
+        assert reply['content']['complete']
+        
+        kc.is_complete('a = [1,\n2,')
+        reply = kc.get_shell_msg(block=True, timeout=TIMEOUT)
+        assert not reply['content']['complete']

IPython/kernel/zmq/ipkernel.py

@@ -229,6 +229,10 b' class IPythonKernel(KernelBase):'
         self.shell.exit_now = True
         return dict(status='ok', restart=restart)
 
+    def do_is_complete(self, code):
+        complete = self.shell.input_transformer_manager.is_complete(code)
+        return {'complete': complete}
+
     def do_apply(self, content, bufs, msg_id, reply_metadata):
         shell = self.shell
         try: