@@ -207,3 +207,119 @@ def sigint_handler(signum, frame):
207207 )
208208 else :
209209 raise
210+
211+
212+ @pytest .mark .anyio
213+ async def test_stdio_client_graceful_stdin_exit ():
214+ """
215+ Test that a process exits gracefully when stdin is closed,
216+ without needing SIGTERM or SIGKILL.
217+ """
218+ # Create a Python script that exits when stdin is closed
219+ script_content = textwrap .dedent (
220+ """
221+ import sys
222+
223+ # Read from stdin until it's closed
224+ try:
225+ while True:
226+ line = sys.stdin.readline()
227+ if not line: # EOF/stdin closed
228+ break
229+ except:
230+ pass
231+
232+ # Exit gracefully
233+ sys.exit(0)
234+ """
235+ )
236+
237+ server_params = StdioServerParameters (
238+ command = sys .executable ,
239+ args = ["-c" , script_content ],
240+ )
241+
242+ start_time = time .time ()
243+
244+ # Use anyio timeout to prevent test from hanging forever
245+ with anyio .move_on_after (5.0 ) as cancel_scope :
246+ async with stdio_client (server_params ) as (read_stream , write_stream ):
247+ # Let the process start and begin reading stdin
248+ await anyio .sleep (0.2 )
249+ # Exit context triggers cleanup - process should exit from stdin closure
250+ pass
251+
252+ if cancel_scope .cancelled_caught :
253+ pytest .fail (
254+ "stdio_client cleanup timed out after 5.0 seconds. "
255+ "Process should have exited gracefully when stdin was closed."
256+ )
257+
258+ end_time = time .time ()
259+ elapsed = end_time - start_time
260+
261+ # Should complete quickly with just stdin closure (no signals needed)
262+ assert elapsed < 3.0 , (
263+ f"stdio_client cleanup took { elapsed :.1f} seconds for stdin-aware process. "
264+ f"Expected < 3.0 seconds since process should exit on stdin closure."
265+ )
266+
267+
268+ @pytest .mark .anyio
269+ async def test_stdio_client_stdin_close_ignored ():
270+ """
271+ Test that when a process ignores stdin closure, the shutdown sequence
272+ properly escalates to SIGTERM.
273+ """
274+ # Create a Python script that ignores stdin closure but responds to SIGTERM
275+ script_content = textwrap .dedent (
276+ """
277+ import signal
278+ import sys
279+ import time
280+
281+ # Set up SIGTERM handler to exit cleanly
282+ def sigterm_handler(signum, frame):
283+ sys.exit(0)
284+
285+ signal.signal(signal.SIGTERM, sigterm_handler)
286+
287+ # Close stdin immediately to simulate ignoring it
288+ sys.stdin.close()
289+
290+ # Keep running until SIGTERM
291+ while True:
292+ time.sleep(0.1)
293+ """
294+ )
295+
296+ server_params = StdioServerParameters (
297+ command = sys .executable ,
298+ args = ["-c" , script_content ],
299+ )
300+
301+ start_time = time .time ()
302+
303+ # Use anyio timeout to prevent test from hanging forever
304+ with anyio .move_on_after (7.0 ) as cancel_scope :
305+ async with stdio_client (server_params ) as (read_stream , write_stream ):
306+ # Let the process start
307+ await anyio .sleep (0.2 )
308+ # Exit context triggers cleanup
309+ pass
310+
311+ if cancel_scope .cancelled_caught :
312+ pytest .fail (
313+ "stdio_client cleanup timed out after 7.0 seconds. "
314+ "Process should have been terminated via SIGTERM escalation."
315+ )
316+
317+ end_time = time .time ()
318+ elapsed = end_time - start_time
319+
320+ # Should take ~2 seconds (stdin close timeout) before SIGTERM is sent
321+ # Total time should be between 2-4 seconds
322+ assert 1.5 < elapsed < 4.5 , (
323+ f"stdio_client cleanup took { elapsed :.1f} seconds for stdin-ignoring process. "
324+ f"Expected between 2-4 seconds (2s stdin timeout + termination time)."
325+ )
0 commit comments