Skip to content

Commit 429284b

Browse files
committed
test: replace polling loops with sleep to avoid partial branch coverage
Use await anyio.sleep() instead of while loops to wait for monitor cancellation. The while loop's False-condition branch was never taken because the scope always exits via cancellation, not loop termination.
1 parent c912fd6 commit 429284b

File tree

1 file changed

+19
-17
lines changed

1 file changed

+19
-17
lines changed

tests/server/test_stdio.py

Lines changed: 19 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -88,25 +88,27 @@ async def test_stdin_eof_monitor_detects_hangup():
8888
"""The EOF monitor cancels the task group when stdin pipe closes."""
8989
read_fd, write_fd = os.pipe()
9090
try:
91-
# Patch sys.stdin.buffer.fileno to return our read end
9291
mock_buffer = MagicMock()
9392
mock_buffer.fileno.return_value = read_fd
9493

94+
cancelled = False
9595
with patch.object(sys, "platform", "linux"), patch.object(sys, "stdin", MagicMock(buffer=mock_buffer)):
96-
async with anyio.create_task_group() as tg:
97-
monitor = _create_stdin_eof_monitor(tg)
98-
assert monitor is not None
99-
tg.start_soon(monitor)
100-
101-
# Close the write end to trigger POLLHUP on read end
102-
os.close(write_fd)
103-
write_fd = -1
104-
105-
# Monitor should detect POLLHUP and cancel the scope
106-
# Wait for the cancellation (with timeout to avoid hanging)
107-
with anyio.fail_after(5):
108-
while not tg.cancel_scope.cancel_called:
109-
await anyio.sleep(0.05)
96+
with anyio.CancelScope() as scope:
97+
async with anyio.create_task_group() as tg:
98+
monitor = _create_stdin_eof_monitor(tg)
99+
assert monitor is not None
100+
tg.start_soon(monitor)
101+
102+
# Close the write end to trigger POLLHUP on read end
103+
os.close(write_fd)
104+
write_fd = -1
105+
106+
# The monitor will cancel the task group scope when it
107+
# detects POLLHUP. Wait with a timeout to avoid hanging.
108+
with anyio.fail_after(5):
109+
await anyio.sleep(10) # will be cancelled by monitor
110+
cancelled = scope.cancel_called or tg.cancel_scope.cancel_called
111+
assert cancelled
110112
finally:
111113
os.close(read_fd)
112114
if write_fd != -1: # pragma: no cover
@@ -141,9 +143,9 @@ async def test_stdin_eof_monitor_ignores_pollin_events():
141143
os.close(write_fd)
142144
write_fd = -1
143145

146+
# Wait for the monitor to detect POLLHUP and cancel
144147
with anyio.fail_after(5):
145-
while not tg.cancel_scope.cancel_called:
146-
await anyio.sleep(0.05)
148+
await anyio.sleep(10) # will be cancelled by monitor
147149
finally:
148150
os.close(read_fd)
149151
if write_fd != -1: # pragma: no cover

0 commit comments

Comments
 (0)