You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
===================================FAILURES===================================___________________TestAsyncKernelClient.test_input_request___________________self=<tests.test_client.TestAsyncKernelClientobjectat0x40067310f0>kc=<jupyter_client.asynchronous.client.AsyncKernelClientobjectat0x400677cd00>
@pytest.mark.skipif(
sys.platform!='linux'orplatform.python_implementation().lower() =='pypy',
reason='only works with cpython on ubuntu in ci',
)
asyncdeftest_input_request(self, kc):
withmock.patch('builtins.input', return_value='test\n'):
>reply=awaitkc.execute_interactive("a = input()", timeout=TIMEOUT)
tests/test_client.py:132:
________________________________________self=<jupyter_client.asynchronous.client.AsyncKernelClientobjectat0x400677cd00>code='a = input()', silent=False, store_history=Trueuser_expressions=None, allow_stdin=True, stop_on_error=Truetimeout=29.978911502999836output_hook=<boundmethodKernelClient._output_hook_defaultof<jupyter_client.asynchronous.client.AsyncKernelClientobjectat0x400677cd00>>stdin_hook=<boundmethodKernelClient._stdin_hook_defaultof<jupyter_client.asynchronous.client.AsyncKernelClientobjectat0x400677cd00>>asyncdef_async_execute_interactive(
self,
code: str,
silent: bool=False,
store_history: bool=True,
user_expressions: t.Optional[t.Dict[str, t.Any]] =None,
allow_stdin: t.Optional[bool] =None,
stop_on_error: bool=True,
timeout: t.Optional[float] =None,
output_hook: t.Optional[t.Callable] =None,
stdin_hook: t.Optional[t.Callable] =None,
) ->t.Dict[str, t.Any]:
"""Execute code in the kernel interactively Output will be redisplayed, and stdin prompts will be relayed as well. If an IPython kernel is detected, rich output will be displayed. You can pass a custom output_hook callable that will be called with every IOPub message that is produced instead of the default redisplay. .. versionadded:: 5.0 Parameters ---------- code : str A string of code in the kernel's language. silent : bool, optional (default False) If set, the kernel will execute the code as quietly possible, and will force store_history to be False. store_history : bool, optional (default True) If set, the kernel will store command history. This is forced to be False if silent is True. user_expressions : dict, optional A dict mapping names to expressions to be evaluated in the user's dict. The expression values are returned as strings formatted using :func:`repr`. allow_stdin : bool, optional (default self.allow_stdin) Flag for whether the kernel can send stdin requests to frontends. Some frontends (e.g. the Notebook) do not support stdin requests. If raw_input is called from code executed from such a frontend, a StdinNotImplementedError will be raised. stop_on_error: bool, optional (default True) Flag whether to abort the execution queue, if an exception is encountered. timeout: float or None (default: None) Timeout to use when waiting for a reply output_hook: callable(msg) Function to be called with output messages. If not specified, output will be redisplayed. stdin_hook: callable(msg) Function or awaitable to be called with stdin_request messages. If not specified, input/getpass will be called. Returns ------- reply: dict The reply message for this request """ifnotself.iopub_channel.is_alive():
emsg="IOPub channel must be running to receive output"raiseRuntimeError(emsg)
ifallow_stdinisNone:
allow_stdin=self.allow_stdinifallow_stdinandnotself.stdin_channel.is_alive():
emsg="stdin channel must be running to allow input"raiseRuntimeError(emsg)
msg_id=awaitensure_async(
self.execute(
code,
silent=silent,
store_history=store_history,
user_expressions=user_expressions,
allow_stdin=allow_stdin,
stop_on_error=stop_on_error,
)
)
ifstdin_hookisNone:
stdin_hook=self._stdin_hook_default# detect IPython kernelifoutput_hookisNoneand"IPython"insys.modules:
fromIPythonimportget_ipythonip=get_ipython()
in_kernel=getattr(ip, "kernel", False)
ifin_kernel:
output_hook=partial(
self._output_hook_kernel,
ip.display_pub.session,
ip.display_pub.pub_socket,
ip.display_pub.parent_header,
)
ifoutput_hookisNone:
# default: redisplay plain-text outputsoutput_hook=self._output_hook_default# set deadline based on timeoutiftimeoutisnotNone:
deadline=time.monotonic() timeoutelse:
timeout_ms=Nonepoller=zmq.Poller()
iopub_socket=self.iopub_channel.socketpoller.register(iopub_socket, zmq.POLLIN)
ifallow_stdin:
stdin_socket=self.stdin_channel.socketpoller.register(stdin_socket, zmq.POLLIN)
else:
stdin_socket=None# wait for output and redisplay itwhileTrue:
iftimeoutisnotNone:
timeout=max(0, deadline-time.monotonic())
timeout_ms=int(1000*timeout)
events=dict(poller.poll(timeout_ms))
ifnotevents:
emsg="Timeout waiting for output">raiseTimeoutError(emsg)
ETimeoutError: Timeoutwaitingforoutputjupyter_client/client.py:536: TimeoutError-----------------------------Capturedstdoutcall-----------------------------a=input()
It still times out if I modify the TIMEOUT variable in tests/test_client.py.
It seems it is deadlocked. Do you have any idea why that test is failing?
This test is marked as only works with cpython on ubuntu in ci. Is there any reason for that? Maybe the test is failing for the same reason here on linux riscv64.
Any help is greatly appreciated.
The text was updated successfully, but these errors were encountered:
Hi,
I am trying to build jupyter_client on riscv64 linux. I have noticed that there is a failing test.
Failing test: TestAsyncKernelClient.test_input_request
It still times out if I modify the
TIMEOUT
variable intests/test_client.py
.It seems it is deadlocked. Do you have any idea why that test is failing?
This test is marked as
only works with cpython on ubuntu in ci
. Is there any reason for that? Maybe the test is failing for the same reason here on linux riscv64.Any help is greatly appreciated.
The text was updated successfully, but these errors were encountered: