##// END OF EJS Templates
Alternative solution: silently apply some CSS instead of a dialog box.
Alternative solution: silently apply some CSS instead of a dialog box.

File last commit:

r6664:a5d15d6a
r7446:c6830009
Show More
test_message_spec.py
452 lines | 11.9 KiB | text/x-python | PythonLexer
/ IPython / zmq / tests / test_message_spec.py
"""Test suite for our zeromq-based messaging specification.
"""
#-----------------------------------------------------------------------------
# Copyright (C) 2010-2011 The IPython Development Team
#
# Distributed under the terms of the BSD License. The full license is in
# the file COPYING.txt, distributed as part of this software.
#-----------------------------------------------------------------------------
import re
import sys
import time
from subprocess import PIPE
from Queue import Empty
import nose.tools as nt
from ..blockingkernelmanager import BlockingKernelManager
from IPython.testing import decorators as dec
from IPython.utils import io
from IPython.utils.traitlets import (
HasTraits, TraitError, Bool, Unicode, Dict, Integer, List, Enum,
)
#-----------------------------------------------------------------------------
# Global setup and utilities
#-----------------------------------------------------------------------------
def setup():
global KM
KM = BlockingKernelManager()
KM.start_kernel(stdout=PIPE, stderr=PIPE)
KM.start_channels()
# wait for kernel to be ready
KM.shell_channel.execute("pass")
KM.shell_channel.get_msg(block=True, timeout=5)
flush_channels()
def teardown():
KM.stop_channels()
KM.shutdown_kernel()
def flush_channels():
"""flush any messages waiting on the queue"""
for channel in (KM.shell_channel, KM.sub_channel):
while True:
try:
msg = channel.get_msg(block=True, timeout=0.1)
except Empty:
break
else:
list(validate_message(msg))
def execute(code='', **kwargs):
"""wrapper for doing common steps for validating an execution request"""
shell = KM.shell_channel
sub = KM.sub_channel
msg_id = shell.execute(code=code, **kwargs)
reply = shell.get_msg(timeout=2)
list(validate_message(reply, 'execute_reply', msg_id))
busy = sub.get_msg(timeout=2)
list(validate_message(busy, 'status', msg_id))
nt.assert_equals(busy['content']['execution_state'], 'busy')
if not kwargs.get('silent'):
pyin = sub.get_msg(timeout=2)
list(validate_message(pyin, 'pyin', msg_id))
nt.assert_equals(pyin['content']['code'], code)
return msg_id, reply['content']
#-----------------------------------------------------------------------------
# MSG Spec References
#-----------------------------------------------------------------------------
class Reference(HasTraits):
def check(self, d):
"""validate a dict against our traits"""
for key in self.trait_names():
yield nt.assert_true(key in d, "Missing key: %r, should be found in %s" % (key, d))
# 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:
yield nt.assert_true(False, str(e))
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):
for tst in Reference.check(self, d):
yield tst
if d['status'] == 'ok':
for tst in ExecuteReplyOkay().check(d):
yield tst
elif d['status'] == 'error':
for tst in ExecuteReplyError().check(d):
yield tst
class ExecuteReplyOkay(Reference):
payload = List(Dict)
user_variables = Dict()
user_expressions = Dict()
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):
for tst in Reference.check(self, d):
yield tst
if d['argspec'] is not None:
for tst in ArgSpec().check(d['argspec']):
yield tst
class ArgSpec(Reference):
args = List(Unicode)
varargs = Unicode()
varkw = Unicode()
defaults = List()
class Status(Reference):
execution_state = Enum((u'busy', u'idle'))
class CompleteReply(Reference):
matches = List(Unicode)
# IOPub messages
class PyIn(Reference):
code = Unicode()
execution_count = Integer()
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)
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)
references = {
'execute_reply' : ExecuteReply(),
'object_info_reply' : OInfoReply(),
'status' : Status(),
'complete_reply' : CompleteReply(),
'pyin' : PyIn(),
'pyout' : PyOut(),
'pyerr' : PyErr(),
'stream' : Stream(),
'display_data' : DisplayData(),
}
def validate_message(msg, msg_type=None, parent=None):
"""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.
"""
RMessage().check(msg)
if msg_type:
yield nt.assert_equals(msg['msg_type'], msg_type)
if parent:
yield nt.assert_equal(msg['parent_header']['msg_id'], parent)
content = msg['content']
ref = references[msg['msg_type']]
for tst in ref.check(content):
yield tst
#-----------------------------------------------------------------------------
# Tests
#-----------------------------------------------------------------------------
# Shell channel
@dec.parametric
def test_execute():
flush_channels()
shell = KM.shell_channel
msg_id = shell.execute(code='x=1')
reply = shell.get_msg(timeout=2)
for tst in validate_message(reply, 'execute_reply', msg_id):
yield tst
@dec.parametric
def test_execute_silent():
flush_channels()
msg_id, reply = execute(code='x=1', silent=True)
# flush status=idle
status = KM.sub_channel.get_msg(timeout=2)
for tst in validate_message(status, 'status', msg_id):
yield tst
nt.assert_equals(status['content']['execution_state'], 'idle')
yield nt.assert_raises(Empty, KM.sub_channel.get_msg, timeout=0.1)
count = reply['execution_count']
msg_id, reply = execute(code='x=2', silent=True)
# flush status=idle
status = KM.sub_channel.get_msg(timeout=2)
for tst in validate_message(status, 'status', msg_id):
yield tst
yield nt.assert_equals(status['content']['execution_state'], 'idle')
yield nt.assert_raises(Empty, KM.sub_channel.get_msg, timeout=0.1)
count_2 = reply['execution_count']
yield nt.assert_equals(count_2, count)
@dec.parametric
def test_execute_error():
flush_channels()
msg_id, reply = execute(code='1/0')
yield nt.assert_equals(reply['status'], 'error')
yield nt.assert_equals(reply['ename'], 'ZeroDivisionError')
pyerr = KM.sub_channel.get_msg(timeout=2)
for tst in validate_message(pyerr, 'pyerr', msg_id):
yield tst
def test_execute_inc():
"""execute request should increment execution_count"""
flush_channels()
msg_id, reply = execute(code='x=1')
count = reply['execution_count']
flush_channels()
msg_id, reply = execute(code='x=2')
count_2 = reply['execution_count']
nt.assert_equals(count_2, count+1)
def test_user_variables():
flush_channels()
msg_id, reply = execute(code='x=1', user_variables=['x'])
user_variables = reply['user_variables']
nt.assert_equals(user_variables, {u'x' : u'1'})
def test_user_expressions():
flush_channels()
msg_id, reply = execute(code='x=1', user_expressions=dict(foo='x+1'))
user_expressions = reply['user_expressions']
nt.assert_equals(user_expressions, {u'foo' : u'2'})
@dec.parametric
def test_oinfo():
flush_channels()
shell = KM.shell_channel
msg_id = shell.object_info('a')
reply = shell.get_msg(timeout=2)
for tst in validate_message(reply, 'object_info_reply', msg_id):
yield tst
@dec.parametric
def test_oinfo_found():
flush_channels()
shell = KM.shell_channel
msg_id, reply = execute(code='a=5')
msg_id = shell.object_info('a')
reply = shell.get_msg(timeout=2)
for tst in validate_message(reply, 'object_info_reply', msg_id):
yield tst
content = reply['content']
yield nt.assert_true(content['found'])
argspec = content['argspec']
yield nt.assert_true(argspec is None, "didn't expect argspec dict, got %r" % argspec)
@dec.parametric
def test_oinfo_detail():
flush_channels()
shell = KM.shell_channel
msg_id, reply = execute(code='ip=get_ipython()')
msg_id = shell.object_info('ip.object_inspect', detail_level=2)
reply = shell.get_msg(timeout=2)
for tst in validate_message(reply, 'object_info_reply', msg_id):
yield tst
content = reply['content']
yield nt.assert_true(content['found'])
argspec = content['argspec']
yield nt.assert_true(isinstance(argspec, dict), "expected non-empty argspec dict, got %r" % argspec)
yield nt.assert_equals(argspec['defaults'], [0])
@dec.parametric
def test_oinfo_not_found():
flush_channels()
shell = KM.shell_channel
msg_id = shell.object_info('dne')
reply = shell.get_msg(timeout=2)
for tst in validate_message(reply, 'object_info_reply', msg_id):
yield tst
content = reply['content']
yield nt.assert_false(content['found'])
@dec.parametric
def test_complete():
flush_channels()
shell = KM.shell_channel
msg_id, reply = execute(code="alpha = albert = 5")
msg_id = shell.complete('al', 'al', 2)
reply = shell.get_msg(timeout=2)
for tst in validate_message(reply, 'complete_reply', msg_id):
yield tst
matches = reply['content']['matches']
for name in ('alpha', 'albert'):
yield nt.assert_true(name in matches, "Missing match: %r" % name)
# IOPub channel
@dec.parametric
def test_stream():
flush_channels()
msg_id, reply = execute("print('hi')")
stdout = KM.sub_channel.get_msg(timeout=2)
for tst in validate_message(stdout, 'stream', msg_id):
yield tst
content = stdout['content']
yield nt.assert_equals(content['name'], u'stdout')
yield nt.assert_equals(content['data'], u'hi\n')
@dec.parametric
def test_display_data():
flush_channels()
msg_id, reply = execute("from IPython.core.display import display; display(1)")
display = KM.sub_channel.get_msg(timeout=2)
for tst in validate_message(display, 'display_data', parent=msg_id):
yield tst
data = display['content']['data']
yield nt.assert_equals(data['text/plain'], u'1')