-

Author: davidbrochart

ipykernel/tests/test_kernel.py

@@ -44,6 +44,7 @@ def _check_status(content):
 def test_simple_print():
     """simple print statement in kernel"""
     with kernel() as kc:
+        iopub = kc.iopub_channel
         msg_id, content = execute(kc=kc, code="print ('hi')")
         stdout, stderr = assemble_output(kc.get_iopub_msg)
         assert stdout == 'hi\n'
@@ -84,6 +85,8 @@ def test_sys_path_profile_dir():
 def test_subprocess_print():
     """printing from forked mp.Process"""
     with new_kernel() as kc:
+        iopub = kc.iopub_channel
+
         _check_master(kc, expected=True)
         flush_channels(kc)
         np = 5
@@ -110,6 +113,8 @@ def test_subprocess_print():
 def test_subprocess_noprint():
     """mp.Process without print doesn't trigger iostream mp_mode"""
     with kernel() as kc:
+        iopub = kc.iopub_channel
+
         np = 5
         code = '\n'.join([
             "import multiprocessing as mp",
@@ -135,6 +140,8 @@ def test_subprocess_noprint():
 def test_subprocess_error():
     """error in mp.Process doesn't crash"""
     with new_kernel() as kc:
+        iopub = kc.iopub_channel
+
         code = '\n'.join([
             "import multiprocessing as mp",
             "p = mp.Process(target=int, args=('hi',))",
@@ -155,6 +162,8 @@ def test_subprocess_error():
 def test_raw_input():
     """test input"""
     with kernel() as kc:
+        iopub = kc.iopub_channel
+
         input_f = "input"
         theprompt = "prompt> "
         code = 'print({input_f}("{theprompt}"))'.format(**locals())
@@ -304,6 +313,8 @@ def test_unc_paths():
         file_path = os.path.splitdrive(os.path.dirname(drive_file_path))[1]
         unc_file_path = os.path.join(unc_root, file_path[1:])
 
+        iopub = kc.iopub_channel
+
         kc.execute("cd {0:s}".format(unc_file_path))
         reply = kc.get_shell_msg(block=True, timeout=TIMEOUT)
         assert reply['content']['status'] == 'ok'