test_message_spec.py
514 lines
| 13.9 KiB
| text/x-python
|
PythonLexer
|
r2926 | """Test suite for our zeromq-based messaging specification. | ||
""" | ||||
#----------------------------------------------------------------------------- | ||||
|
r5390 | # Copyright (C) 2010-2011 The IPython Development Team | ||
|
r2926 | # | ||
# Distributed under the terms of the BSD License. The full license is in | ||||
# the file COPYING.txt, distributed as part of this software. | ||||
#----------------------------------------------------------------------------- | ||||
|
r6554 | import re | ||
from subprocess import PIPE | ||||
from Queue import Empty | ||||
|
r2926 | |||
import nose.tools as nt | ||||
|
r10385 | from IPython.kernel import KernelManager | ||
|
r2926 | |||
|
r6558 | from IPython.testing import decorators as dec | ||
|
r6554 | from IPython.utils.traitlets import ( | ||
|
r8865 | HasTraits, TraitError, Bool, Unicode, Dict, Integer, List, Enum, Any, | ||
|
r6554 | ) | ||
#----------------------------------------------------------------------------- | ||||
# Global setup and utilities | ||||
#----------------------------------------------------------------------------- | ||||
|
r2926 | |||
|
r11802 | STARTUP_TIMEOUT = 60 | ||
TIMEOUT = 15 | ||||
|
r2926 | def setup(): | ||
|
r10290 | global KM, KC | ||
KM = KernelManager() | ||||
|
r6554 | KM.start_kernel(stdout=PIPE, stderr=PIPE) | ||
|
r10300 | KC = KM.client() | ||
|
r10290 | KC.start_channels() | ||
|
r6664 | |||
# wait for kernel to be ready | ||||
|
r11347 | try: | ||
|
r11802 | msg = KC.iopub_channel.get_msg(block=True, timeout=STARTUP_TIMEOUT) | ||
|
r11347 | except Empty: | ||
pass | ||||
|
r11802 | msg_id = KC.kernel_info() | ||
KC.get_shell_msg(block=True, timeout=STARTUP_TIMEOUT) | ||||
|
r6664 | flush_channels() | ||
|
r2926 | |||
|
r6558 | |||
|
r2926 | def teardown(): | ||
|
r10290 | KC.stop_channels() | ||
|
r6554 | KM.shutdown_kernel() | ||
|
r6558 | |||
|
r10290 | def flush_channels(kc=None): | ||
|
r6554 | """flush any messages waiting on the queue""" | ||
|
r10290 | if kc is None: | ||
kc = KC | ||||
for channel in (kc.shell_channel, kc.iopub_channel): | ||||
|
r6558 | while True: | ||
try: | ||||
msg = channel.get_msg(block=True, timeout=0.1) | ||||
except Empty: | ||||
break | ||||
else: | ||||
|
r6573 | list(validate_message(msg)) | ||
|
r6554 | |||
|
r10290 | def execute(code='', kc=None, **kwargs): | ||
|
r6554 | """wrapper for doing common steps for validating an execution request""" | ||
|
r10294 | msg_id = KC.execute(code=code, **kwargs) | ||
|
r11802 | reply = KC.get_shell_msg(timeout=TIMEOUT) | ||
|
r6573 | list(validate_message(reply, 'execute_reply', msg_id)) | ||
|
r11802 | busy = KC.get_iopub_msg(timeout=TIMEOUT) | ||
|
r6573 | list(validate_message(busy, 'status', msg_id)) | ||
|
r7875 | nt.assert_equal(busy['content']['execution_state'], 'busy') | ||
|
r6554 | |||
if not kwargs.get('silent'): | ||||
|
r11802 | pyin = KC.get_iopub_msg(timeout=TIMEOUT) | ||
|
r6573 | list(validate_message(pyin, 'pyin', msg_id)) | ||
|
r7875 | nt.assert_equal(pyin['content']['code'], code) | ||
|
r6554 | |||
return msg_id, reply['content'] | ||||
#----------------------------------------------------------------------------- | ||||
# MSG Spec References | ||||
#----------------------------------------------------------------------------- | ||||
class Reference(HasTraits): | ||||
|
r8829 | |||
""" | ||||
Base class for message spec specification testing. | ||||
This class is the core of the message specification test. The | ||||
idea is that child classes implement trait attributes for each | ||||
message keys, so that message keys can be tested against these | ||||
traits using :meth:`check` method. | ||||
""" | ||||
|
r6554 | def check(self, d): | ||
"""validate a dict against our traits""" | ||||
for key in self.trait_names(): | ||||
|
r6558 | yield nt.assert_true(key in d, "Missing key: %r, should be found in %s" % (key, d)) | ||
|
r6554 | # FIXME: always allow None, probably not a good idea | ||
if d[key] is None: | ||||
continue | ||||
try: | ||||
setattr(self, key, d[key]) | ||||
except TraitError as e: | ||||
|
r6558 | yield nt.assert_true(False, str(e)) | ||
|
r6554 | |||
class RMessage(Reference): | ||||
msg_id = Unicode() | ||||
msg_type = Unicode() | ||||
header = Dict() | ||||
parent_header = Dict() | ||||
content = Dict() | ||||
class RHeader(Reference): | ||||
msg_id = Unicode() | ||||
msg_type = Unicode() | ||||
session = Unicode() | ||||
username = Unicode() | ||||
class RContent(Reference): | ||||
status = Enum((u'ok', u'error')) | ||||
class ExecuteReply(Reference): | ||||
execution_count = Integer() | ||||
status = Enum((u'ok', u'error')) | ||||
def check(self, d): | ||||
|
r6558 | for tst in Reference.check(self, d): | ||
yield tst | ||||
|
r6554 | if d['status'] == 'ok': | ||
|
r6558 | for tst in ExecuteReplyOkay().check(d): | ||
yield tst | ||||
|
r6554 | elif d['status'] == 'error': | ||
|
r6558 | for tst in ExecuteReplyError().check(d): | ||
yield tst | ||||
|
r6554 | |||
|
r2926 | |||
|
r6554 | class ExecuteReplyOkay(Reference): | ||
payload = List(Dict) | ||||
user_variables = Dict() | ||||
user_expressions = Dict() | ||||
|
r2926 | |||
|
r6554 | |||
class ExecuteReplyError(Reference): | ||||
ename = Unicode() | ||||
evalue = Unicode() | ||||
traceback = List(Unicode) | ||||
class OInfoReply(Reference): | ||||
name = Unicode() | ||||
found = Bool() | ||||
ismagic = Bool() | ||||
isalias = Bool() | ||||
namespace = Enum((u'builtin', u'magics', u'alias', u'Interactive')) | ||||
type_name = Unicode() | ||||
string_form = Unicode() | ||||
base_class = Unicode() | ||||
length = Integer() | ||||
file = Unicode() | ||||
definition = Unicode() | ||||
argspec = Dict() | ||||
init_definition = Unicode() | ||||
docstring = Unicode() | ||||
init_docstring = Unicode() | ||||
class_docstring = Unicode() | ||||
call_def = Unicode() | ||||
call_docstring = Unicode() | ||||
source = Unicode() | ||||
def check(self, d): | ||||
|
r6558 | for tst in Reference.check(self, d): | ||
yield tst | ||||
|
r6554 | if d['argspec'] is not None: | ||
|
r6558 | for tst in ArgSpec().check(d['argspec']): | ||
yield tst | ||||
|
r6554 | |||
class ArgSpec(Reference): | ||||
args = List(Unicode) | ||||
varargs = Unicode() | ||||
varkw = Unicode() | ||||
|
r6558 | defaults = List() | ||
|
r6554 | |||
class Status(Reference): | ||||
|
r10341 | execution_state = Enum((u'busy', u'idle', u'starting')) | ||
|
r6554 | |||
class CompleteReply(Reference): | ||||
matches = List(Unicode) | ||||
|
r8865 | def Version(num, trait=Integer): | ||
return List(trait, default_value=[0] * num, minlen=num, maxlen=num) | ||||
|
r8879 | class KernelInfoReply(Reference): | ||
|
r8865 | |||
protocol_version = Version(2) | ||||
ipython_version = Version(4, Any) | ||||
language_version = Version(3) | ||||
language = Unicode() | ||||
def _ipython_version_changed(self, name, old, new): | ||||
for v in new: | ||||
nt.assert_true( | ||||
isinstance(v, int) or isinstance(v, basestring), | ||||
|
r8875 | 'expected int or string as version component, got {0!r}' | ||
|
r8865 | .format(v)) | ||
|
r8831 | |||
|
r6554 | # IOPub messages | ||
class PyIn(Reference): | ||||
code = Unicode() | ||||
|
r6559 | execution_count = Integer() | ||
|
r6554 | |||
PyErr = ExecuteReplyError | ||||
class Stream(Reference): | ||||
name = Enum((u'stdout', u'stderr')) | ||||
data = Unicode() | ||||
mime_pat = re.compile(r'\w+/\w+') | ||||
class DisplayData(Reference): | ||||
source = Unicode() | ||||
metadata = Dict() | ||||
data = Dict() | ||||
def _data_changed(self, name, old, new): | ||||
for k,v in new.iteritems(): | ||||
nt.assert_true(mime_pat.match(k)) | ||||
nt.assert_true(isinstance(v, basestring), "expected string data, got %r" % v) | ||||
|
r6664 | class PyOut(Reference): | ||
execution_count = Integer() | ||||
data = Dict() | ||||
def _data_changed(self, name, old, new): | ||||
for k,v in new.iteritems(): | ||||
nt.assert_true(mime_pat.match(k)) | ||||
nt.assert_true(isinstance(v, basestring), "expected string data, got %r" % v) | ||||
|
r6554 | references = { | ||
'execute_reply' : ExecuteReply(), | ||||
'object_info_reply' : OInfoReply(), | ||||
'status' : Status(), | ||||
'complete_reply' : CompleteReply(), | ||||
|
r8879 | 'kernel_info_reply': KernelInfoReply(), | ||
|
r6554 | 'pyin' : PyIn(), | ||
|
r6664 | 'pyout' : PyOut(), | ||
|
r6554 | 'pyerr' : PyErr(), | ||
'stream' : Stream(), | ||||
'display_data' : DisplayData(), | ||||
} | ||||
|
r8829 | """ | ||
Specifications of `content` part of the reply messages. | ||||
""" | ||||
|
r6554 | |||
def validate_message(msg, msg_type=None, parent=None): | ||||
|
r6573 | """validate a message | ||
This is a generator, and must be iterated through to actually | ||||
trigger each test. | ||||
If msg_type and/or parent are given, the msg_type and/or parent msg_id | ||||
are compared with the given values. | ||||
""" | ||||
|
r6554 | RMessage().check(msg) | ||
if msg_type: | ||||
|
r7875 | yield nt.assert_equal(msg['msg_type'], msg_type) | ||
|
r6554 | if parent: | ||
|
r6558 | yield nt.assert_equal(msg['parent_header']['msg_id'], parent) | ||
|
r6554 | content = msg['content'] | ||
ref = references[msg['msg_type']] | ||||
|
r6558 | for tst in ref.check(content): | ||
yield tst | ||||
|
r6554 | |||
#----------------------------------------------------------------------------- | ||||
# Tests | ||||
#----------------------------------------------------------------------------- | ||||
# Shell channel | ||||
|
r2926 | |||
|
r6558 | @dec.parametric | ||
|
r2926 | def test_execute(): | ||
|
r6558 | flush_channels() | ||
|
r10294 | msg_id = KC.execute(code='x=1') | ||
|
r11802 | reply = KC.get_shell_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(reply, 'execute_reply', msg_id): | ||
yield tst | ||||
|
r6554 | |||
|
r6558 | @dec.parametric | ||
|
r6554 | def test_execute_silent(): | ||
|
r6558 | flush_channels() | ||
|
r6554 | msg_id, reply = execute(code='x=1', silent=True) | ||
# flush status=idle | ||||
|
r11802 | status = KC.iopub_channel.get_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(status, 'status', msg_id): | ||
yield tst | ||||
|
r7875 | nt.assert_equal(status['content']['execution_state'], 'idle') | ||
|
r6554 | |||
|
r10290 | yield nt.assert_raises(Empty, KC.iopub_channel.get_msg, timeout=0.1) | ||
|
r6554 | count = reply['execution_count'] | ||
msg_id, reply = execute(code='x=2', silent=True) | ||||
# flush status=idle | ||||
|
r11802 | status = KC.iopub_channel.get_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(status, 'status', msg_id): | ||
yield tst | ||||
|
r7875 | yield nt.assert_equal(status['content']['execution_state'], 'idle') | ||
|
r6554 | |||
|
r10290 | yield nt.assert_raises(Empty, KC.iopub_channel.get_msg, timeout=0.1) | ||
|
r6554 | count_2 = reply['execution_count'] | ||
|
r7875 | yield nt.assert_equal(count_2, count) | ||
|
r6554 | |||
|
r6558 | @dec.parametric | ||
|
r6554 | def test_execute_error(): | ||
|
r6558 | flush_channels() | ||
|
r6554 | |||
msg_id, reply = execute(code='1/0') | ||||
|
r7875 | yield nt.assert_equal(reply['status'], 'error') | ||
yield nt.assert_equal(reply['ename'], 'ZeroDivisionError') | ||||
|
r6554 | |||
|
r11802 | pyerr = KC.iopub_channel.get_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(pyerr, 'pyerr', msg_id): | ||
yield tst | ||||
|
r6554 | |||
def test_execute_inc(): | ||||
"""execute request should increment execution_count""" | ||||
|
r6558 | flush_channels() | ||
|
r6554 | msg_id, reply = execute(code='x=1') | ||
count = reply['execution_count'] | ||||
flush_channels() | ||||
|
r2926 | |||
|
r6554 | msg_id, reply = execute(code='x=2') | ||
count_2 = reply['execution_count'] | ||||
|
r7875 | nt.assert_equal(count_2, count+1) | ||
|
r6554 | |||
def test_user_variables(): | ||||
|
r6558 | flush_channels() | ||
|
r6554 | msg_id, reply = execute(code='x=1', user_variables=['x']) | ||
user_variables = reply['user_variables'] | ||||
|
r10637 | nt.assert_equal(user_variables, {u'x': { | ||
u'status': u'ok', | ||||
u'data': {u'text/plain': u'1'}, | ||||
u'metadata': {}, | ||||
}}) | ||||
def test_user_variables_fail(): | ||||
flush_channels() | ||||
msg_id, reply = execute(code='x=1', user_variables=['nosuchname']) | ||||
user_variables = reply['user_variables'] | ||||
foo = user_variables['nosuchname'] | ||||
nt.assert_equal(foo['status'], 'error') | ||||
nt.assert_equal(foo['ename'], 'KeyError') | ||||
|
r6554 | |||
def test_user_expressions(): | ||||
|
r6558 | flush_channels() | ||
|
r6554 | msg_id, reply = execute(code='x=1', user_expressions=dict(foo='x+1')) | ||
user_expressions = reply['user_expressions'] | ||||
|
r10637 | nt.assert_equal(user_expressions, {u'foo': { | ||
u'status': u'ok', | ||||
u'data': {u'text/plain': u'2'}, | ||||
u'metadata': {}, | ||||
}}) | ||||
def test_user_expressions_fail(): | ||||
flush_channels() | ||||
msg_id, reply = execute(code='x=0', user_expressions=dict(foo='nosuchname')) | ||||
user_expressions = reply['user_expressions'] | ||||
foo = user_expressions['foo'] | ||||
nt.assert_equal(foo['status'], 'error') | ||||
nt.assert_equal(foo['ename'], 'NameError') | ||||
|
r6554 | |||
|
r6558 | @dec.parametric | ||
|
r6554 | def test_oinfo(): | ||
|
r6558 | flush_channels() | ||
|
r10294 | msg_id = KC.object_info('a') | ||
|
r11802 | reply = KC.get_shell_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(reply, 'object_info_reply', msg_id): | ||
yield tst | ||||
|
r6554 | |||
|
r6558 | @dec.parametric | ||
|
r6554 | def test_oinfo_found(): | ||
|
r6558 | flush_channels() | ||
|
r6554 | |||
msg_id, reply = execute(code='a=5') | ||||
|
r10294 | msg_id = KC.object_info('a') | ||
|
r11802 | reply = KC.get_shell_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(reply, 'object_info_reply', msg_id): | ||
yield tst | ||||
|
r6554 | content = reply['content'] | ||
|
r6558 | yield nt.assert_true(content['found']) | ||
|
r6560 | argspec = content['argspec'] | ||
yield nt.assert_true(argspec is None, "didn't expect argspec dict, got %r" % argspec) | ||||
|
r6554 | |||
|
r6558 | @dec.parametric | ||
|
r6557 | def test_oinfo_detail(): | ||
|
r6558 | flush_channels() | ||
|
r6557 | |||
msg_id, reply = execute(code='ip=get_ipython()') | ||||
|
r6558 | |||
|
r10294 | msg_id = KC.object_info('ip.object_inspect', detail_level=2) | ||
|
r11802 | reply = KC.get_shell_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(reply, 'object_info_reply', msg_id): | ||
yield tst | ||||
|
r6557 | content = reply['content'] | ||
|
r6558 | yield nt.assert_true(content['found']) | ||
|
r6560 | argspec = content['argspec'] | ||
yield nt.assert_true(isinstance(argspec, dict), "expected non-empty argspec dict, got %r" % argspec) | ||||
|
r7875 | yield nt.assert_equal(argspec['defaults'], [0]) | ||
|
r6557 | |||
|
r6558 | @dec.parametric | ||
|
r6557 | def test_oinfo_not_found(): | ||
|
r6558 | flush_channels() | ||
|
r6557 | |||
|
r10294 | msg_id = KC.object_info('dne') | ||
|
r11802 | reply = KC.get_shell_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(reply, 'object_info_reply', msg_id): | ||
yield tst | ||||
|
r6557 | content = reply['content'] | ||
|
r6558 | yield nt.assert_false(content['found']) | ||
|
r6557 | |||
|
r6558 | @dec.parametric | ||
|
r6554 | def test_complete(): | ||
|
r6558 | flush_channels() | ||
|
r6554 | msg_id, reply = execute(code="alpha = albert = 5") | ||
|
r10294 | msg_id = KC.complete('al', 'al', 2) | ||
|
r11802 | reply = KC.get_shell_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(reply, 'complete_reply', msg_id): | ||
yield tst | ||||
|
r6554 | matches = reply['content']['matches'] | ||
for name in ('alpha', 'albert'): | ||||
|
r6558 | yield nt.assert_true(name in matches, "Missing match: %r" % name) | ||
|
r6554 | |||
|
r8831 | @dec.parametric | ||
|
r8879 | def test_kernel_info_request(): | ||
|
r8831 | flush_channels() | ||
|
r10294 | msg_id = KC.kernel_info() | ||
|
r11802 | reply = KC.get_shell_msg(timeout=TIMEOUT) | ||
|
r8879 | for tst in validate_message(reply, 'kernel_info_reply', msg_id): | ||
|
r8831 | yield tst | ||
|
r6561 | # IOPub channel | ||
|
r6558 | @dec.parametric | ||
|
r6554 | def test_stream(): | ||
|
r6558 | flush_channels() | ||
|
r6554 | msg_id, reply = execute("print('hi')") | ||
|
r11802 | stdout = KC.iopub_channel.get_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(stdout, 'stream', msg_id): | ||
yield tst | ||||
|
r6554 | content = stdout['content'] | ||
|
r7875 | yield nt.assert_equal(content['name'], u'stdout') | ||
yield nt.assert_equal(content['data'], u'hi\n') | ||||
|
r6554 | |||
|
r6558 | @dec.parametric | ||
def test_display_data(): | ||||
flush_channels() | ||||
|
r6554 | msg_id, reply = execute("from IPython.core.display import display; display(1)") | ||
|
r11802 | display = KC.iopub_channel.get_msg(timeout=TIMEOUT) | ||
|
r6558 | for tst in validate_message(display, 'display_data', parent=msg_id): | ||
yield tst | ||||
|
r6554 | data = display['content']['data'] | ||
|
r7875 | yield nt.assert_equal(data['text/plain'], u'1') | ||
|
r6554 | |||