Tests: Add comprehensive tests for silent flag in subshells

Author: pankaj-bind

tests/test_subshells.py

@@ -6,11 +6,12 @@
 
 import platform
 import time
+from queue import Empty
 
 import pytest
 from jupyter_client.blocking.client import BlockingKernelClient
 
-from .utils import TIMEOUT, assemble_output, get_replies, get_reply, new_kernel
+from .utils import TIMEOUT, assemble_output, get_replies, get_reply, new_kernel, flush_channels
 
 # Helpers
 
@@ -39,8 +40,8 @@ def list_subshell_helper(kc: BlockingKernelClient):
     return reply["content"]
 
 
-def execute_request(kc: BlockingKernelClient, code: str, subshell_id: str | None):
-    msg = kc.session.msg("execute_request", {"code": code})
+def execute_request(kc: BlockingKernelClient, code: str, subshell_id: str | None, silent: bool = False):
+    msg = kc.session.msg("execute_request", {"code": code, "silent": silent})
     msg["header"]["subshell_id"] = subshell_id
     kc.shell_channel.send(msg)
     return msg
@@ -51,17 +52,17 @@ def execute_request_subshell_id(
 ):
     msg = execute_request(kc, code, subshell_id)
     msg_id = msg["header"]["msg_id"]
-    stdout, _ = assemble_output(kc.get_iopub_msg, None, msg_id)
+    stdout, _ = assemble_output(kc.get_iopub_msg, parent_msg_id=msg_id)
     return stdout.strip()
 
 
 def execute_thread_count(kc: BlockingKernelClient) -> int:
-    code = "print(threading.active_count())"
+    code = "import threading; print(threading.active_count())"
     return int(execute_request_subshell_id(kc, code, None))
 
 
 def execute_thread_ids(kc: BlockingKernelClient, subshell_id: str | None = None) -> tuple[str, str]:
-    code = "print(threading.get_ident(), threading.main_thread().ident)"
+    code = "import threading; print(threading.get_ident(), threading.main_thread().ident)"
     return execute_request_subshell_id(kc, code, subshell_id).split()
 
 
@@ -223,16 +224,16 @@ def test_execute_stop_on_error(are_subshells):
         msg = execute_request(
             kc, "import asyncio; await asyncio.sleep(1); raise ValueError()", subshell_ids[0]
         )
-        msg_ids.append(msg["msg_id"])
+        msg_ids.append(msg["header"]["msg_id"])
         msg = execute_request(kc, "print('hello')", subshell_ids[0])
-        msg_ids.append(msg["msg_id"])
+        msg_ids.append(msg["header"]["msg_id"])
         msg = execute_request(kc, "print('goodbye')", subshell_ids[0])
-        msg_ids.append(msg["msg_id"])
+        msg_ids.append(msg["header"]["msg_id"])
 
         msg = execute_request(kc, "import time; time.sleep(1.5)", subshell_ids[1])
-        msg_ids.append(msg["msg_id"])
+        msg_ids.append(msg["header"]["msg_id"])
         msg = execute_request(kc, "print('other')", subshell_ids[1])
-        msg_ids.append(msg["msg_id"])
+        msg_ids.append(msg["header"]["msg_id"])
 
         replies = get_replies(kc, msg_ids)
 
@@ -258,3 +259,57 @@ def test_execute_stop_on_error(are_subshells):
         for subshell_id in subshell_ids:
             if subshell_id:
                 delete_subshell_helper(kc, subshell_id)
+
+def test_silent_flag_in_subshells():
+    """Verifies that the 'silent' flag suppresses output in main and subshell contexts."""
+    with new_kernel() as kc:
+        flush_channels(kc)
+        # Test silent execution in main shell
+        msg_main_silent = execute_request(kc, "a=1", None, silent=True)
+        reply_main_silent = get_reply(kc, msg_main_silent["header"]["msg_id"])
+        assert reply_main_silent['content']['status'] == 'ok'
+
+        # Test silent execution in subshell
+        subshell_id = create_subshell_helper(kc)["subshell_id"]
+        msg_sub_silent = execute_request(kc, "b=2", subshell_id, silent=True)
+        reply_sub_silent = get_reply(kc, msg_sub_silent["header"]["msg_id"])
+        assert reply_sub_silent['content']['status'] == 'ok'
+
+        # Check for no output from silent requests. We should only see status messages,
+        # so we expect a timeout here when looking for other messages.
+        while True:
+            try:
+                msg = kc.get_iopub_msg(timeout=0.2)
+                # We should only receive status messages
+                if msg['header']['msg_type'] == 'status':
+                    continue
+                # If we get anything else, it's a failure
+                pytest.fail(f"Silent execution produced an unexpected IOPub message: {msg['header']['msg_type']}")
+            except Empty:
+                # No more messages, which is the expected behavior for silent execution
+                break
+
+        # Test concurrent silent and non-silent execution
+        msg_silent = execute_request(kc, "import time; time.sleep(0.5); c=3", subshell_id, silent=True)
+        msg_noisy = execute_request(kc, "print('noisy')", None, silent=False)
+
+        # Get replies for both messages
+        replies = get_replies(kc, [msg_silent['header']['msg_id'], msg_noisy['header']['msg_id']])
+        assert len(replies) == 2
+
+        # Verify that we only receive stream output from the noisy message
+        stdout, stderr = assemble_output(kc.get_iopub_msg, parent_msg_id=msg_noisy['header']['msg_id'])
+        assert "noisy" in stdout
+        assert not stderr
+
+        # Verify there is no output from the silent message.
+        try:
+            stdout, stderr = assemble_output(kc.get_iopub_msg, parent_msg_id=msg_silent['header']['msg_id'])
+            assert not stdout
+            assert not stderr
+        except Empty:
+            # This is expected since there should be no output
+            pass
+
+        # Cleanup
+        delete_subshell_helper(kc, subshell_id)