Add a failing test for child thread

Author: krassowski

tests/test_kernel.py

@@ -90,6 +90,42 @@ def thread_target():
             received += 1
 
 
+def test_print_to_correct_cell_from_child_thread():
+    """should print to the cell that spawned the thread, not a subsequently run cell"""
+    iterations = 5
+    interval = 0.25
+    code = f"""\
+    from threading import Thread
+    from time import sleep
+
+    def child_target():
+        for i in range({iterations}):
+            print(i, end='', flush=True)
+            sleep({interval})
+
+    def parent_target():
+        sleep({interval})
+        Thread(target=child_target).start()
+
+    Thread(target=parent_target).start()
+    """
+    with kernel() as kc:
+        thread_msg_id = kc.execute(code)
+        _ = kc.execute("pass")
+
+        received = 0
+        while received < iterations:
+            msg = kc.get_iopub_msg(timeout=interval * 2)
+            if msg["msg_type"] != "stream":
+                continue
+            content = msg["content"]
+            assert content["name"] == "stdout"
+            assert content["text"] == str(received)
+            # this is crucial as the parent header decides to which cell the output goes
+            assert msg["parent_header"]["msg_id"] == thread_msg_id
+            received += 1
+
+
 def test_print_to_correct_cell_from_asyncio():
     """should print to the cell that scheduled the task, not a subsequently run cell"""
     iterations = 5