Merge pull request #6657 from takluyver/adapt-no-code-to-line

Fix code_to_line when called with empty code

IPython/kernel/adapter.py

@@ -1,354 +1,356 b''
 """Adapters for IPython msg spec versions."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import json
 
 from IPython.core.release import kernel_protocol_version_info
 from IPython.utils.tokenutil import token_at_cursor
 
 
 def code_to_line(code, cursor_pos):
     """Turn a multiline code block and cursor position into a single line
     and new cursor position.
     
     For adapting ``complete_`` and ``object_info_request``.
     """
+    if not code:
+        return "", 0
     for line in code.splitlines(True):
         n = len(line)
         if cursor_pos > n:
             cursor_pos -= n
         else:
             break
     return line, cursor_pos
 
 
 class Adapter(object):
     """Base class for adapting messages
     
     Override message_type(msg) methods to create adapters.
     """
     
     msg_type_map = {}
     
     def update_header(self, msg):
         return msg
     
     def update_metadata(self, msg):
         return msg
     
     def update_msg_type(self, msg):
         header = msg['header']
         msg_type = header['msg_type']
         if msg_type in self.msg_type_map:
             msg['msg_type'] = header['msg_type'] = self.msg_type_map[msg_type]
         return msg
     
     def handle_reply_status_error(self, msg):
         """This will be called *instead of* the regular handler
         
         on any reply with status != ok
         """
         return msg
     
     def __call__(self, msg):
         msg = self.update_header(msg)
         msg = self.update_metadata(msg)
         msg = self.update_msg_type(msg)
         header = msg['header']
         
         handler = getattr(self, header['msg_type'], None)
         if handler is None:
             return msg
         
         # handle status=error replies separately (no change, at present)
         if msg['content'].get('status', None) in {'error', 'aborted'}:
             return self.handle_reply_status_error(msg)
         return handler(msg)
 
 def _version_str_to_list(version):
     """convert a version string to a list of ints
     
     non-int segments are excluded
     """
     v = []
     for part in version.split('.'):
         try:
             v.append(int(part))
         except ValueError:
             pass
     return v
 
 class V5toV4(Adapter):
     """Adapt msg protocol v5 to v4"""
     
     version = '4.1'
     
     msg_type_map = {
         'execute_result' : 'pyout',
         'execute_input' : 'pyin',
         'error' : 'pyerr',
         'inspect_request' : 'object_info_request',
         'inspect_reply' : 'object_info_reply',
     }
     
     def update_header(self, msg):
         msg['header'].pop('version', None)
         return msg
     
     # shell channel
     
     def kernel_info_reply(self, msg):
         content = msg['content']
         content.pop('banner', None)
         for key in ('language_version', 'protocol_version'):
             if key in content:
                 content[key] = _version_str_to_list(content[key])
         if content.pop('implementation', '') == 'ipython' \
             and 'implementation_version' in content:
             content['ipython_version'] = content.pop('implmentation_version')
         content.pop('implementation_version', None)
         content.setdefault("implmentation", content['language'])
         return msg
     
     def execute_request(self, msg):
         content = msg['content']
         content.setdefault('user_variables', [])
         return msg
     
     def execute_reply(self, msg):
         content = msg['content']
         content.setdefault('user_variables', {})
         # TODO: handle payloads
         return msg
     
     def complete_request(self, msg):
         content = msg['content']
         code = content['code']
         cursor_pos = content['cursor_pos']
         line, cursor_pos = code_to_line(code, cursor_pos)
         
         new_content = msg['content'] = {}
         new_content['text'] = ''
         new_content['line'] = line
         new_content['block'] = None
         new_content['cursor_pos'] = cursor_pos
         return msg
     
     def complete_reply(self, msg):
         content = msg['content']
         cursor_start = content.pop('cursor_start')
         cursor_end = content.pop('cursor_end')
         match_len = cursor_end - cursor_start
         content['matched_text'] = content['matches'][0][:match_len]
         content.pop('metadata', None)
         return msg
     
     def object_info_request(self, msg):
         content = msg['content']
         code = content['code']
         cursor_pos = content['cursor_pos']
         line, _ = code_to_line(code, cursor_pos)
         
         new_content = msg['content'] = {}
         new_content['oname'] = token_at_cursor(code, cursor_pos)
         new_content['detail_level'] = content['detail_level']
         return msg
     
     def object_info_reply(self, msg):
         """inspect_reply can't be easily backward compatible"""
         msg['content'] = {'found' : False, 'name' : 'unknown'}
         return msg
     
     # iopub channel
     
     def stream(self, msg):
         content = msg['content']
         content['data'] = content.pop('text')
         return msg
 
     def display_data(self, msg):
         content = msg['content']
         content.setdefault("source", "display")
         data = content['data']
         if 'application/json' in data:
             try:
                 data['application/json'] = json.dumps(data['application/json'])
             except Exception:
                 # warn?
                 pass
         return msg
     
     # stdin channel
     
     def input_request(self, msg):
         msg['content'].pop('password', None)
         return msg
 
 
 class V4toV5(Adapter):
     """Convert msg spec V4 to V5"""
     version = '5.0'
     
     # invert message renames above
     msg_type_map = {v:k for k,v in V5toV4.msg_type_map.items()}
     
     def update_header(self, msg):
         msg['header']['version'] = self.version
         return msg
     
     # shell channel
     
     def kernel_info_reply(self, msg):
         content = msg['content']
         for key in ('language_version', 'protocol_version', 'ipython_version'):
             if key in content:
                 content[key] = ".".join(map(str, content[key]))
         
         if content['language'].startswith('python') and 'ipython_version' in content:
             content['implementation'] = 'ipython'
             content['implementation_version'] = content.pop('ipython_version')
         
         content['banner'] = ''
         return msg
     
     def execute_request(self, msg):
         content = msg['content']
         user_variables = content.pop('user_variables', [])
         user_expressions = content.setdefault('user_expressions', {})
         for v in user_variables:
             user_expressions[v] = v
         return msg
     
     def execute_reply(self, msg):
         content = msg['content']
         user_expressions = content.setdefault('user_expressions', {})
         user_variables = content.pop('user_variables', {})
         if user_variables:
             user_expressions.update(user_variables)
 
         # Pager payloads became a mime bundle
         for payload in content.get('payload', []):
             if payload.get('source', None) == 'page' and ('text' in payload):
                 if 'data' not in payload:
                     payload['data'] = {}
                 payload['data']['text/plain'] = payload.pop('text')
 
         return msg
     
     def complete_request(self, msg):
         old_content = msg['content']
         
         new_content = msg['content'] = {}
         new_content['code'] = old_content['line']
         new_content['cursor_pos'] = old_content['cursor_pos']
         return msg
     
     def complete_reply(self, msg):
         # complete_reply needs more context than we have to get cursor_start and end.
         # use special value of `-1` to indicate to frontend that it should be at
         # the current cursor position.
         content = msg['content']
         new_content = msg['content'] = {'status' : 'ok'}
         new_content['matches'] = content['matches']
         new_content['cursor_start'] = -len(content['matched_text'])
         new_content['cursor_end'] = None
         new_content['metadata'] = {}
         return msg
     
     def inspect_request(self, msg):
         content = msg['content']
         name = content['oname']
         
         new_content = msg['content'] = {}
         new_content['code'] = name
         new_content['cursor_pos'] = len(name)
         new_content['detail_level'] = content['detail_level']
         return msg
     
     def inspect_reply(self, msg):
         """inspect_reply can't be easily backward compatible"""
         content = msg['content']
         new_content = msg['content'] = {'status' : 'ok'}
         found = new_content['found'] = content['found']
         new_content['name'] = content['name']
         new_content['data'] = data = {}
         new_content['metadata'] = {}
         if found:
             lines = []
             for key in ('call_def', 'init_definition', 'definition'):
                 if content.get(key, False):
                     lines.append(content[key])
                     break
             for key in ('call_docstring', 'init_docstring', 'docstring'):
                 if content.get(key, False):
                     lines.append(content[key])
                     break
             if not lines:
                 lines.append("<empty docstring>")
             data['text/plain'] = '\n'.join(lines)
         return msg
     
     # iopub channel
     
     def stream(self, msg):
         content = msg['content']
         content['text'] = content.pop('data')
         return msg
 
     def display_data(self, msg):
         content = msg['content']
         content.pop("source", None)
         data = content['data']
         if 'application/json' in data:
             try:
                 data['application/json'] = json.loads(data['application/json'])
             except Exception:
                 # warn?
                 pass
         return msg
     
     # stdin channel
     
     def input_request(self, msg):
         msg['content'].setdefault('password', False)
         return msg
     
 
 
 def adapt(msg, to_version=kernel_protocol_version_info[0]):
     """Adapt a single message to a target version
     
     Parameters
     ----------
     
     msg : dict
         An IPython message.
     to_version : int, optional
         The target major version.
         If unspecified, adapt to the current version for IPython.
     
     Returns
     -------
     
     msg : dict
         An IPython message appropriate in the new version.
     """
     header = msg['header']
     if 'version' in header:
         from_version = int(header['version'].split('.')[0])
     else:
         # assume last version before adding the key to the header
         from_version = 4
     adapter = adapters.get((from_version, to_version), None)
     if adapter is None:
         return msg
     return adapter(msg)
 
 
 # one adapter per major version from,to
 adapters = {
     (5,4) : V5toV4(),
     (4,5) : V4toV5(),
 }

IPython/kernel/tests/test_adapter.py

@@ -1,330 +1,334 b''
 """Tests for adapting IPython msg spec versions"""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import copy
 import json
 from unittest import TestCase
 import nose.tools as nt
 
-from IPython.kernel.adapter import adapt, V4toV5, V5toV4
+from IPython.kernel.adapter import adapt, V4toV5, V5toV4, code_to_line
 from IPython.kernel.zmq.session import Session
 
 
 def test_default_version():
     s = Session()
     msg = s.msg("msg_type")
     msg['header'].pop('version')
     original = copy.deepcopy(msg)
     adapted = adapt(original)
     nt.assert_equal(adapted['header']['version'], V4toV5.version)
 
+def test_code_to_line_no_code():
+    line, pos = code_to_line("", 0)
+    nt.assert_equal(line, "")
+    nt.assert_equal(pos, 0)
 
 class AdapterTest(TestCase):
     
     def setUp(self):
         self.session = Session()
     
     def adapt(self, msg, version=None):
         original = copy.deepcopy(msg)
         adapted = adapt(msg, version or self.to_version)
         return original, adapted
     
     def check_header(self, msg):
         pass
 
 
 class V4toV5TestCase(AdapterTest):
     from_version = 4
     to_version = 5
     
     def msg(self, msg_type, content):
         """Create a v4 msg (same as v5, minus version header)"""
         msg = self.session.msg(msg_type, content)
         msg['header'].pop('version')
         return msg
     
     def test_same_version(self):
         msg = self.msg("execute_result",
             content={'status' : 'ok'}
         )
         original, adapted = self.adapt(msg, self.from_version)
     
         self.assertEqual(original, adapted)
     
     def test_no_adapt(self):
         msg = self.msg("input_reply", {'value' : 'some text'})
         v4, v5 = self.adapt(msg)
         self.assertEqual(v5['header']['version'], V4toV5.version)
         v5['header'].pop('version')
         self.assertEqual(v4, v5)
     
     def test_rename_type(self):
         for v5_type, v4_type in [
                 ('execute_result', 'pyout'),
                 ('execute_input', 'pyin'),
                 ('error', 'pyerr'),
             ]:
             msg = self.msg(v4_type, {'key' : 'value'})
             v4, v5 = self.adapt(msg)
             self.assertEqual(v5['header']['version'], V4toV5.version)
             self.assertEqual(v5['header']['msg_type'], v5_type)
             self.assertEqual(v4['content'], v5['content'])
     
     def test_execute_request(self):
         msg = self.msg("execute_request", {
             'code' : 'a=5',
             'silent' : False,
             'user_expressions' : {'a' : 'apple'},
             'user_variables' : ['b'],
         })
         v4, v5 = self.adapt(msg)
         self.assertEqual(v4['header']['msg_type'], v5['header']['msg_type'])
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(v5c['user_expressions'], {'a' : 'apple', 'b': 'b'})
         self.assertNotIn('user_variables', v5c)
         self.assertEqual(v5c['code'], v4c['code'])
 
     def test_execute_reply(self):
         msg = self.msg("execute_reply", {
             'status': 'ok',
             'execution_count': 7,
             'user_variables': {'a': 1},
             'user_expressions': {'a+a': 2},
             'payload': [{'source':'page', 'text':'blah'}]
         })
         v4, v5 = self.adapt(msg)
         v5c = v5['content']
         self.assertNotIn('user_variables', v5c)
         self.assertEqual(v5c['user_expressions'], {'a': 1, 'a+a': 2})
         self.assertEqual(v5c['payload'], [{'source': 'page',
                                            'data': {'text/plain': 'blah'}}
                                          ])
     
     def test_complete_request(self):
         msg = self.msg("complete_request", {
             'text' : 'a.is',
             'line' : 'foo = a.is',
             'block' : None,
             'cursor_pos' : 10,
         })
         v4, v5 = self.adapt(msg)
         v4c = v4['content']
         v5c = v5['content']
         for key in ('text', 'line', 'block'):
             self.assertNotIn(key, v5c)
         self.assertEqual(v5c['cursor_pos'], v4c['cursor_pos'])
         self.assertEqual(v5c['code'], v4c['line'])
     
     def test_complete_reply(self):
         msg = self.msg("complete_reply", {
             'matched_text' : 'a.is',
             'matches' : ['a.isalnum',
                         'a.isalpha',
                         'a.isdigit',
                         'a.islower',
                         ],
         })
         v4, v5 = self.adapt(msg)
         v4c = v4['content']
         v5c = v5['content']
         
         self.assertEqual(v5c['matches'], v4c['matches'])
         self.assertEqual(v5c['metadata'], {})
         self.assertEqual(v5c['cursor_start'], -4)
         self.assertEqual(v5c['cursor_end'], None)
     
     def test_object_info_request(self):
         msg = self.msg("object_info_request", {
             'oname' : 'foo',
             'detail_level' : 1,
         })
         v4, v5 = self.adapt(msg)
         self.assertEqual(v5['header']['msg_type'], 'inspect_request')
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(v5c['code'], v4c['oname'])
         self.assertEqual(v5c['cursor_pos'], len(v4c['oname']))
         self.assertEqual(v5c['detail_level'], v4c['detail_level'])
     
     def test_object_info_reply(self):
         msg = self.msg("object_info_reply", {
             'name' : 'foo',
             'found' : True,
             'status' : 'ok',
             'definition' : 'foo(a=5)',
             'docstring' : "the docstring",
         })
         v4, v5 = self.adapt(msg)
         self.assertEqual(v5['header']['msg_type'], 'inspect_reply')
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(sorted(v5c), [ 'data', 'found', 'metadata', 'name', 'status'])
         text = v5c['data']['text/plain']
         self.assertEqual(text, '\n'.join([v4c['definition'], v4c['docstring']]))
     
     # iopub channel
     
     def test_display_data(self):
         jsondata = dict(a=5)
         msg = self.msg("display_data", {
             'data' : {
                 'text/plain' : 'some text',
                 'application/json' : json.dumps(jsondata)
             },
             'metadata' : {'text/plain' : { 'key' : 'value' }},
         })
         v4, v5 = self.adapt(msg)
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(v5c['metadata'], v4c['metadata'])
         self.assertEqual(v5c['data']['text/plain'], v4c['data']['text/plain'])
         self.assertEqual(v5c['data']['application/json'], jsondata)
     
     # stdin channel
     
     def test_input_request(self):
         msg = self.msg('input_request', {'prompt': "$>"})
         v4, v5 = self.adapt(msg)
         self.assertEqual(v5['content']['prompt'], v4['content']['prompt'])
         self.assertFalse(v5['content']['password'])
 
 
 class V5toV4TestCase(AdapterTest):
     from_version = 5
     to_version = 4
     
     def msg(self, msg_type, content):
         return self.session.msg(msg_type, content)
     
     def test_same_version(self):
         msg = self.msg("execute_result",
             content={'status' : 'ok'}
         )
         original, adapted = self.adapt(msg, self.from_version)
     
         self.assertEqual(original, adapted)
     
     def test_no_adapt(self):
         msg = self.msg("input_reply", {'value' : 'some text'})
         v5, v4 = self.adapt(msg)
         self.assertNotIn('version', v4['header'])
         v5['header'].pop('version')
         self.assertEqual(v4, v5)
     
     def test_rename_type(self):
         for v5_type, v4_type in [
                 ('execute_result', 'pyout'),
                 ('execute_input', 'pyin'),
                 ('error', 'pyerr'),
             ]:
             msg = self.msg(v5_type, {'key' : 'value'})
             v5, v4 = self.adapt(msg)
             self.assertEqual(v4['header']['msg_type'], v4_type)
             nt.assert_not_in('version', v4['header'])
             self.assertEqual(v4['content'], v5['content'])
     
     def test_execute_request(self):
         msg = self.msg("execute_request", {
             'code' : 'a=5',
             'silent' : False,
             'user_expressions' : {'a' : 'apple'},
         })
         v5, v4 = self.adapt(msg)
         self.assertEqual(v4['header']['msg_type'], v5['header']['msg_type'])
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(v4c['user_variables'], [])
         self.assertEqual(v5c['code'], v4c['code'])
     
     def test_complete_request(self):
         msg = self.msg("complete_request", {
             'code' : 'def foo():\n'
                      '    a.is\n'
                      'foo()',
             'cursor_pos': 19,
         })
         v5, v4 = self.adapt(msg)
         v4c = v4['content']
         v5c = v5['content']
         self.assertNotIn('code', v4c)
         self.assertEqual(v4c['line'], v5c['code'].splitlines(True)[1])
         self.assertEqual(v4c['cursor_pos'], 8)
         self.assertEqual(v4c['text'], '')
         self.assertEqual(v4c['block'], None)
     
     def test_complete_reply(self):
         msg = self.msg("complete_reply", {
             'cursor_start' : 10,
             'cursor_end' : 14,
             'matches' : ['a.isalnum',
                         'a.isalpha',
                         'a.isdigit',
                         'a.islower',
                         ],
             'metadata' : {},
         })
         v5, v4 = self.adapt(msg)
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(v4c['matched_text'], 'a.is')
         self.assertEqual(v4c['matches'], v5c['matches'])
     
     def test_inspect_request(self):
         msg = self.msg("inspect_request", {
             'code' : 'def foo():\n'
                      '    apple\n'
                      'bar()',
             'cursor_pos': 18,
             'detail_level' : 1,
         })
         v5, v4 = self.adapt(msg)
         self.assertEqual(v4['header']['msg_type'], 'object_info_request')
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(v4c['oname'], 'apple')
         self.assertEqual(v5c['detail_level'], v4c['detail_level'])
     
     def test_inspect_reply(self):
         msg = self.msg("inspect_reply", {
             'name' : 'foo',
             'found' : True,
             'data' : {'text/plain' : 'some text'},
             'metadata' : {},
         })
         v5, v4 = self.adapt(msg)
         self.assertEqual(v4['header']['msg_type'], 'object_info_reply')
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(sorted(v4c), ['found', 'name'])
         self.assertEqual(v4c['found'], False)
     
     # iopub channel
     
     def test_display_data(self):
         jsondata = dict(a=5)
         msg = self.msg("display_data", {
             'data' : {
                 'text/plain' : 'some text',
                 'application/json' : jsondata,
             },
             'metadata' : {'text/plain' : { 'key' : 'value' }},
         })
         v5, v4 = self.adapt(msg)
         v4c = v4['content']
         v5c = v5['content']
         self.assertEqual(v5c['metadata'], v4c['metadata'])
         self.assertEqual(v5c['data']['text/plain'], v4c['data']['text/plain'])
         self.assertEqual(v4c['data']['application/json'], json.dumps(jsondata))
     
     # stdin channel
     
     def test_input_request(self):
         msg = self.msg('input_request', {'prompt': "$>", 'password' : True})
         v5, v4 = self.adapt(msg)
         self.assertEqual(v5['content']['prompt'], v4['content']['prompt'])
         self.assertNotIn('password', v4['content'])