From 8b5a30422268ed58e7a4dcb6ec4e513758db6b2d Mon Sep 17 00:00:00 2001
From: krypticmouse <herumbshandilya123@gmail.com>
Date: Wed, 20 May 2026 03:36:37 +0000
Subject: [PATCH] test(mcp): regression test for
 StdioTransport.send_notification

Follow-up on @Dilligaf371's PR #339 fix. Adds a regression test that
spawns a subprocess which consumes stdin without ever writing to
stdout, then issues `send_notification` from a worker thread. If the
override is missing, the thread blocks forever on `proc.stdout.readline()`
and the 2-second join timeout fires.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 tests/mcp/test_transport.py | 48 +++++++++++++++++++++++++++++++++++++
 1 file changed, 48 insertions(+)

diff --git a/tests/mcp/test_transport.py b/tests/mcp/test_transport.py
index 4cd10e86..17ec4bb5 100644
--- a/tests/mcp/test_transport.py
+++ b/tests/mcp/test_transport.py
@@ -136,6 +136,54 @@ class TestStdioTransport:
         finally:
             transport.close()
 
+    def test_send_notification_does_not_read_stdout(self, tmp_path):
+        """Regression for #339: stdio servers don't reply to notifications.
+
+        The base ``MCPTransport.send_notification`` falls back to ``send()``,
+        which writes the request and then blocks on ``proc.stdout.readline()``.
+        For stdio MCP servers that never reply to a notification, that read
+        hangs forever, breaking the JSON-RPC ``notifications/initialized``
+        handshake. ``StdioTransport.send_notification`` must override that
+        behavior to be write-only.
+
+        This test spawns a subprocess that consumes stdin without ever
+        writing to stdout, then issues ``send_notification`` from a worker
+        thread. If the override is missing, the thread blocks on
+        ``readline()`` and the join timeout fires.
+        """
+        import threading
+
+        script = tmp_path / "silent_consumer.py"
+        script.write_text(
+            textwrap.dedent("""\
+            import sys
+            for _ in sys.stdin:
+                pass
+        """)
+        )
+
+        transport = StdioTransport([sys.executable, str(script)])
+        try:
+            notification = MCPRequest(method="notifications/initialized")
+            result_box: dict = {}
+
+            def call():
+                try:
+                    transport.send_notification(notification)
+                    result_box["ok"] = True
+                except Exception as exc:  # noqa: BLE001
+                    result_box["error"] = exc
+
+            worker = threading.Thread(target=call, daemon=True)
+            worker.start()
+            worker.join(timeout=2.0)
+            assert not worker.is_alive(), (
+                "send_notification blocked on stdout — override missing"
+            )
+            assert result_box.get("ok") is True, result_box
+        finally:
+            transport.close()
+
     def test_close_terminates_process(self, tmp_path):
         script = tmp_path / "sleep_server.py"
         script.write_text(