Merge pull request #578 from davidbrochart/fix_inprocess

Rework wait_for_ready logic

ipykernel/inprocess/blocking.py

@@ -76,10 +76,21 @@ class BlockingInProcessKernelClient(InProcessKernelClient):
  def wait_for_ready(self):
  # Wait for kernel info reply on shell channel
  while True:
- msg = self.shell_channel.get_msg(block=True)
- if msg['msg_type'] == 'kernel_info_reply':
- self._handle_kernel_info_reply(msg)
- break
+ self.kernel_info()
+ try:
+ msg = self.shell_channel.get_msg(block=True, timeout=1)
+ except Empty:
+ pass
+ else:
+ if msg['msg_type'] == 'kernel_info_reply':
+ # Checking that IOPub is connected. If it is not connected, start over.
+ try:
+ self.iopub_channel.get_msg(block=True, timeout=0.2)
+ except Empty:
+ pass
+ else:
+ self._handle_kernel_info_reply(msg)
+ break
 
  # Flush IOPub channel
  while True:

ipykernel/tests/test_kernel.py

@@ -21,7 +21,7 @@
 
 from .utils import (
  new_kernel, kernel, TIMEOUT, assemble_output, execute,
- flush_channels, wait_for_idle,
+ flush_channels, wait_for_idle, get_reply,
 )
 
 
@@ -360,9 +360,9 @@ def test_interrupt_during_input():
  msg_id = kc.execute("input()")
  time.sleep(1) # Make sure it's actually waiting for input.
  km.interrupt_kernel()
- # If we failed to interrupt interrupt, this will timeout:
- reply = kc.get_shell_msg(timeout=TIMEOUT)
  from .test_message_spec import validate_message
+ # If we failed to interrupt interrupt, this will timeout:
+ reply = get_reply(kc, msg_id, TIMEOUT)
  validate_message(reply, 'execute_reply', msg_id)
 
 
@@ -385,9 +385,10 @@ def test_interrupt_during_pdb_set_trace():
  msg_id2 = kc.execute("3 + 4")
  time.sleep(1) # Make sure it's actually waiting for input.
  km.interrupt_kernel()
- # If we failed to interrupt interrupt, this will timeout:
  from .test_message_spec import validate_message
- reply = kc.get_shell_msg(timeout=TIMEOUT)
+ # If we failed to interrupt interrupt, this will timeout:
+ reply = get_reply(kc, msg_id, TIMEOUT)
  validate_message(reply, 'execute_reply', msg_id)
- reply = kc.get_shell_msg(timeout=TIMEOUT)
+ # If we failed to interrupt interrupt, this will timeout:
+ reply = get_reply(kc, msg_id2, TIMEOUT)
  validate_message(reply, 'execute_reply', msg_id2)

ipykernel/tests/test_message_spec.py

@@ -15,7 +15,8 @@
  HasTraits, TraitError, Bool, Unicode, Dict, Integer, List, Enum
 )
 
-from .utils import TIMEOUT, start_global_kernel, flush_channels, execute
+from .utils import (TIMEOUT, start_global_kernel, flush_channels, execute,
+ get_reply, )
 
 #-----------------------------------------------------------------------------
 # Globals
@@ -278,7 +279,7 @@ def test_execute():
  flush_channels()
 
  msg_id = KC.execute(code='x=1')
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'execute_reply', msg_id)
 
 
@@ -405,7 +406,7 @@ def test_oinfo():
  flush_channels()
 
  msg_id = KC.inspect('a')
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'inspect_reply', msg_id)
 
 
@@ -415,7 +416,7 @@ def test_oinfo_found():
  msg_id, reply = execute(code='a=5')
 
  msg_id = KC.inspect('a')
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'inspect_reply', msg_id)
  content = reply['content']
  assert content['found']
@@ -430,7 +431,7 @@ def test_oinfo_detail():
  msg_id, reply = execute(code='ip=get_ipython()')
 
  msg_id = KC.inspect('ip.object_inspect', cursor_pos=10, detail_level=1)
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'inspect_reply', msg_id)
  content = reply['content']
  assert content['found']
@@ -443,7 +444,7 @@ def test_oinfo_not_found():
  flush_channels()
 
  msg_id = KC.inspect('dne')
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'inspect_reply', msg_id)
  content = reply['content']
  assert not content['found']
@@ -455,7 +456,7 @@ def test_complete():
  msg_id, reply = execute(code="alpha = albert = 5")
 
  msg_id = KC.complete('al', 2)
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'complete_reply', msg_id)
  matches = reply['content']['matches']
  for name in ('alpha', 'albert'):
@@ -466,7 +467,7 @@ def test_kernel_info_request():
  flush_channels()
 
  msg_id = KC.kernel_info()
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'kernel_info_reply', msg_id)
 
 
@@ -477,7 +478,7 @@ def test_connect_request():
  return msg['header']['msg_id']
 
  msg_id = KC.kernel_info()
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'connect_reply', msg_id)
 
 
@@ -486,7 +487,7 @@ def test_comm_info_request():
  if not hasattr(KC, 'comm_info'):
  raise SkipTest()
  msg_id = KC.comm_info()
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'comm_info_reply', msg_id)
 
 
@@ -512,7 +513,7 @@ def test_is_complete():
  flush_channels()
 
  msg_id = KC.is_complete("a = 1")
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'is_complete_reply', msg_id)
 
 def test_history_range():
@@ -522,7 +523,7 @@ def test_history_range():
  reply_exec = KC.get_shell_msg(timeout=TIMEOUT)
 
  msg_id = KC.history(hist_access_type = 'range', raw = True, output = True, start = 1, stop = 2, session = 0)
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'history_reply', msg_id)
  content = reply['content']
  assert len(content['history']) == 1
@@ -534,7 +535,7 @@ def test_history_tail():
  reply_exec = KC.get_shell_msg(timeout=TIMEOUT)
 
  msg_id = KC.history(hist_access_type = 'tail', raw = True, output = True, n = 1, session = 0)
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'history_reply', msg_id)
  content = reply['content']
  assert len(content['history']) == 1
@@ -546,7 +547,7 @@ def test_history_search():
  reply_exec = KC.get_shell_msg(timeout=TIMEOUT)
 
  msg_id = KC.history(hist_access_type = 'search', raw = True, output = True, n = 1, pattern = '*', session = 0)
- reply = KC.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(KC, msg_id, TIMEOUT)
  validate_message(reply, 'history_reply', msg_id)
  content = reply['content']
  assert len(content['history']) == 1

ipykernel/tests/utils.py

@@ -6,6 +6,7 @@
 import atexit
 import os
 import sys
+from time import time
 
 from contextlib import contextmanager
 from queue import Empty
@@ -52,13 +53,26 @@ def flush_channels(kc=None):
  validate_message(msg)
 
 
+def get_reply(kc, msg_id, timeout):
+ timeout = TIMEOUT
+ t0 = time()
+ while True:
+ reply = kc.get_shell_msg(timeout=timeout)
+ if reply['parent_header']['msg_id'] == msg_id:
+ break
+ t1 = time()
+ timeout -= t1 - t0
+ t0 = t1
+ return reply
+
+
 def execute(code='', kc=None, **kwargs):
  """wrapper for doing common steps for validating an execution request"""
  from .test_message_spec import validate_message
  if kc is None:
  kc = KC
  msg_id = kc.execute(code=code, **kwargs)
- reply = kc.get_shell_msg(timeout=TIMEOUT)
+ reply = get_reply(kc, msg_id, TIMEOUT)
  validate_message(reply, 'execute_reply', msg_id)
  busy = kc.get_iopub_msg(timeout=TIMEOUT)
  validate_message(busy, 'status', msg_id)

setup.py

@@ -97,6 +97,7 @@ def run(self):
  'pytest-cov',
  'flaky',
  'nose', # nose because there are still a few nose.tools imports hanging around
+ 'jedi<=0.17.2'
  ],
  },
  classifiers=[