IPython/kernel/tests/test_adapter.py

"""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.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)


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'])