|
1 | 1 | """ |
2 | 2 | Very minimal unittests for parts of the readline module. |
3 | 3 | """ |
4 | | -from contextlib import ExitStack |
5 | | -from errno import EIO |
6 | 4 | import locale |
7 | 5 | import os |
8 | | -import selectors |
9 | | -import subprocess |
10 | 6 | import sys |
11 | 7 | import tempfile |
12 | 8 | import unittest |
13 | 9 | from test.support import verbose |
14 | 10 | from test.support.import_helper import import_module |
15 | 11 | from test.support.os_helper import unlink, temp_dir, TESTFN |
| 12 | +from test.support.pty_helper import run_pty |
16 | 13 | from test.support.script_helper import assert_python_ok |
17 | 14 |
|
18 | 15 | # Skip tests if there is no readline module |
@@ -304,55 +301,5 @@ def test_history_size(self): |
304 | 301 | self.assertEqual(lines[-1].strip(), b"last input") |
305 | 302 |
|
306 | 303 |
|
307 | | -def run_pty(script, input=b"dummy input\r", env=None): |
308 | | - pty = import_module('pty') |
309 | | - output = bytearray() |
310 | | - [master, slave] = pty.openpty() |
311 | | - args = (sys.executable, '-c', script) |
312 | | - proc = subprocess.Popen(args, stdin=slave, stdout=slave, stderr=slave, env=env) |
313 | | - os.close(slave) |
314 | | - with ExitStack() as cleanup: |
315 | | - cleanup.enter_context(proc) |
316 | | - def terminate(proc): |
317 | | - try: |
318 | | - proc.terminate() |
319 | | - except ProcessLookupError: |
320 | | - # Workaround for Open/Net BSD bug (Issue 16762) |
321 | | - pass |
322 | | - cleanup.callback(terminate, proc) |
323 | | - cleanup.callback(os.close, master) |
324 | | - # Avoid using DefaultSelector and PollSelector. Kqueue() does not |
325 | | - # work with pseudo-terminals on OS X < 10.9 (Issue 20365) and Open |
326 | | - # BSD (Issue 20667). Poll() does not work with OS X 10.6 or 10.4 |
327 | | - # either (Issue 20472). Hopefully the file descriptor is low enough |
328 | | - # to use with select(). |
329 | | - sel = cleanup.enter_context(selectors.SelectSelector()) |
330 | | - sel.register(master, selectors.EVENT_READ | selectors.EVENT_WRITE) |
331 | | - os.set_blocking(master, False) |
332 | | - while True: |
333 | | - for [_, events] in sel.select(): |
334 | | - if events & selectors.EVENT_READ: |
335 | | - try: |
336 | | - chunk = os.read(master, 0x10000) |
337 | | - except OSError as err: |
338 | | - # Linux raises EIO when slave is closed (Issue 5380) |
339 | | - if err.errno != EIO: |
340 | | - raise |
341 | | - chunk = b"" |
342 | | - if not chunk: |
343 | | - return output |
344 | | - output.extend(chunk) |
345 | | - if events & selectors.EVENT_WRITE: |
346 | | - try: |
347 | | - input = input[os.write(master, input):] |
348 | | - except OSError as err: |
349 | | - # Apparently EIO means the slave was closed |
350 | | - if err.errno != EIO: |
351 | | - raise |
352 | | - input = b"" # Stop writing |
353 | | - if not input: |
354 | | - sel.modify(master, selectors.EVENT_READ) |
355 | | - |
356 | | - |
357 | 304 | if __name__ == "__main__": |
358 | 305 | unittest.main() |
0 commit comments