1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
"""test the IPython Kernel"""
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
import os
import warnings
import pytest
if os.name == "nt":
pytest.skip("skipping tests on windows", allow_module_level=True)
async def test_direct_kernel_info_request(kernel):
reply = await kernel.test_shell_message("kernel_info_request", {})
assert reply["header"]["msg_type"] == "kernel_info_reply"
async def test_direct_execute_request(kernel):
reply = await kernel.test_shell_message("execute_request", dict(code="hello", silent=False))
assert reply["header"]["msg_type"] == "execute_reply"
async def test_direct_execute_request_aborting(kernel):
kernel._aborting = True
reply = await kernel.test_shell_message("execute_request", dict(code="hello", silent=False))
assert reply["header"]["msg_type"] == "execute_reply"
assert reply["content"]["status"] == "aborted"
async def test_direct_execute_request_error(kernel):
await kernel.execute_request(None, None, None)
async def test_complete_request(kernel):
reply = await kernel.test_shell_message("complete_request", dict(code="hello", cursor_pos=0))
assert reply["header"]["msg_type"] == "complete_reply"
async def test_inspect_request(kernel):
reply = await kernel.test_shell_message("inspect_request", dict(code="hello", cursor_pos=0))
assert reply["header"]["msg_type"] == "inspect_reply"
async def test_history_request(kernel):
reply = await kernel.test_shell_message(
"history_request", dict(hist_access_type="", output="", raw="")
)
assert reply["header"]["msg_type"] == "history_reply"
reply = await kernel.test_shell_message(
"history_request", dict(hist_access_type="tail", output="", raw="")
)
assert reply["header"]["msg_type"] == "history_reply"
reply = await kernel.test_shell_message(
"history_request", dict(hist_access_type="range", output="", raw="")
)
assert reply["header"]["msg_type"] == "history_reply"
reply = await kernel.test_shell_message(
"history_request", dict(hist_access_type="search", output="", raw="")
)
assert reply["header"]["msg_type"] == "history_reply"
async def test_comm_info_request(kernel):
reply = await kernel.test_shell_message("comm_info_request")
assert reply["header"]["msg_type"] == "comm_info_reply"
async def test_direct_interrupt_request(kernel):
reply = await kernel.test_control_message("interrupt_request", {})
assert reply["header"]["msg_type"] == "interrupt_reply"
assert reply["content"] == {"status": "ok"}
# test failure on interrupt request
def raiseOSError():
msg = "evalue"
raise OSError(msg)
kernel._send_interrupt_children = raiseOSError
reply = await kernel.test_control_message("interrupt_request", {})
assert reply["header"]["msg_type"] == "interrupt_reply"
assert reply["content"]["status"] == "error"
assert reply["content"]["ename"] == "OSError"
assert reply["content"]["evalue"] == "evalue"
assert len(reply["content"]["traceback"]) > 0
async def test_direct_shutdown_request(kernel):
reply = await kernel.test_shell_message("shutdown_request", dict(restart=False))
assert reply["header"]["msg_type"] == "shutdown_reply"
reply = await kernel.test_shell_message("shutdown_request", dict(restart=True))
assert reply["header"]["msg_type"] == "shutdown_reply"
async def test_is_complete_request(kernel):
reply = await kernel.test_shell_message("is_complete_request", dict(code="hello"))
assert reply["header"]["msg_type"] == "is_complete_reply"
async def test_direct_debug_request(kernel):
reply = await kernel.test_control_message("debug_request", {})
assert reply["header"]["msg_type"] == "debug_reply"
async def test_deprecated_features(kernel):
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
header = kernel._parent_header
assert isinstance(header, dict)
shell_streams = kernel.shell_streams
assert len(shell_streams) == 1
assert shell_streams[0] == kernel.shell_stream
warnings.simplefilter("ignore", RuntimeWarning)
kernel.shell_streams = [kernel.shell_stream, kernel.shell_stream]
async def test_process_control(kernel):
from jupyter_client.session import DELIM
class FakeMsg:
def __init__(self, bytes):
self.bytes = bytes
await kernel.process_control([FakeMsg(DELIM), 1])
msg = kernel._prep_msg("does_not_exist")
await kernel.process_control(msg)
def test_should_handle(kernel):
msg = kernel.session.msg("debug_request", {})
assert kernel.should_handle(kernel.control_stream, msg, []) is True
async def test_dispatch_shell(kernel):
from jupyter_client.session import DELIM
class FakeMsg:
def __init__(self, bytes):
self.bytes = bytes
await kernel.dispatch_shell([FakeMsg(DELIM), 1])
msg = kernel._prep_msg("does_not_exist")
await kernel.dispatch_shell(msg)
async def test_publish_debug_event(kernel):
kernel._publish_debug_event({})
async def test_connect_request(kernel):
await kernel.connect_request(kernel.shell_stream, "foo", {})
async def test_send_interrupt_children(kernel):
kernel._send_interrupt_children()
# TODO: this causes deadlock
# async def test_direct_usage_request(kernel):
# reply = await kernel.test_control_message("usage_request", {})
# assert reply['header']['msg_type'] == 'usage_reply'
|
