traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Issue #26870: Avoid using kqueue() with pseudo-terminals 

Also force terminate the child process in case it hangs for any reason.
This commit is contained in:
Martin Panter 2016-05-15 13:21:25 +00:00
parent f47fc5553b
commit 2e1d8683c1
1 changed files with 7 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
Lib/test/test_readline.py
View File

@ -9,7 +9,7 @@ import subprocess
import sys import sys
import tempfile import tempfile
import unittest import unittest
from test.support import import_module, unlink, get_original_stdout from test.support import import_module, unlink
from test.support.script_helper import assert_python_ok from test.support.script_helper import assert_python_ok
# Skip tests if there is no readline module # Skip tests if there is no readline module
@ -126,30 +126,30 @@ def run_pty(script, input=b"dummy input\r"):
os.close(slave) os.close(slave)
with ExitStack() as cleanup: with ExitStack() as cleanup:
cleanup.enter_context(proc) cleanup.enter_context(proc)
cleanup.callback(proc.terminate)
cleanup.callback(os.close, master) cleanup.callback(os.close, master)
sel = cleanup.enter_context(selectors.DefaultSelector()) # Avoid using DefaultSelector, because it may choose a kqueue()
# implementation, which does not work with pseudo-terminals on OS X
# < 10.9 (Issue 20365) and Open BSD (Issue 20667).
sel = getattr(selectors, "PollSelector", selectors.DefaultSelector)()
cleanup.enter_context(sel)
sel.register(master, selectors.EVENT_READ | selectors.EVENT_WRITE) sel.register(master, selectors.EVENT_READ | selectors.EVENT_WRITE)
os.set_blocking(master, False) os.set_blocking(master, False)
while True: while True:
get_original_stdout().write(f"test_readline: select()\n")
for [_, events] in sel.select(): for [_, events] in sel.select():
if events & selectors.EVENT_READ: if events & selectors.EVENT_READ:
try: try:
get_original_stdout().write(f"test_readline: read()\n")
chunk = os.read(master, 0x10000) chunk = os.read(master, 0x10000)
except OSError as err: except OSError as err:
# Linux raises EIO when the slave is closed # Linux raises EIO when the slave is closed
if err.errno != EIO: if err.errno != EIO:
raise raise
chunk = b"" chunk = b""
get_original_stdout().write(f"test_readline: read {chunk!r}\n")
if not chunk: if not chunk:
return output return output
output.extend(chunk) output.extend(chunk)
if events & selectors.EVENT_WRITE: if events & selectors.EVENT_WRITE:
get_original_stdout().write(f"test_readline: write()\n")
input = input[os.write(master, input):] input = input[os.write(master, input):]
get_original_stdout().write(f"test_readline: remaining input = {input!r}\n")
if not input: if not input:
sel.modify(master, selectors.EVENT_READ) sel.modify(master, selectors.EVENT_READ)