| import array | |
| import os | |
| import struct | |
| import sys | |
| import threading | |
| import unittest | |
| from test import support | |
| from test.support import os_helper, threading_helper | |
| from test.support.import_helper import import_module | |
| fcntl = import_module('fcntl') | |
| termios = import_module('termios') | |
| class IoctlTestsTty(unittest.TestCase): | |
| @classmethod | |
| def setUpClass(cls): | |
| TIOCGPGRP = support.get_attribute(termios, 'TIOCGPGRP') | |
| try: | |
| tty = open("/dev/tty", "rb") | |
| except OSError: | |
| raise unittest.SkipTest("Unable to open /dev/tty") | |
| with tty: | |
| # Skip if another process is in foreground | |
| r = fcntl.ioctl(tty, TIOCGPGRP, struct.pack("i", 0)) | |
| rpgrp = struct.unpack("i", r)[0] | |
| if rpgrp not in (os.getpgrp(), os.getsid(0)): | |
| raise unittest.SkipTest("Neither the process group nor the session " | |
| "are attached to /dev/tty") | |
| def test_ioctl_immutable_buf(self): | |
| # If this process has been put into the background, TIOCGPGRP returns | |
| # the session ID instead of the process group id. | |
| ids = (os.getpgrp(), os.getsid(0)) | |
| with open("/dev/tty", "rb") as tty: | |
| # string | |
| buf = " "*8 | |
| r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf) | |
| self.assertIsInstance(r, bytes) | |
| rpgrp = memoryview(r).cast('i')[0] | |
| self.assertIn(rpgrp, ids) | |
| # bytes | |
| buf = b" "*8 | |
| r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf) | |
| self.assertIsInstance(r, bytes) | |
| rpgrp = memoryview(r).cast('i')[0] | |
| self.assertIn(rpgrp, ids) | |
| # read-only buffer | |
| r = fcntl.ioctl(tty, termios.TIOCGPGRP, memoryview(buf)) | |
| self.assertIsInstance(r, bytes) | |
| rpgrp = memoryview(r).cast('i')[0] | |
| self.assertIn(rpgrp, ids) | |
| def test_ioctl_mutable_buf(self): | |
| ids = (os.getpgrp(), os.getsid(0)) | |
| with open("/dev/tty", "rb") as tty: | |
| buf = bytearray(b" "*8) | |
| r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf) | |
| self.assertEqual(r, 0) | |
| rpgrp = memoryview(buf).cast('i')[0] | |
| self.assertIn(rpgrp, ids) | |
| def test_ioctl_no_mutate_buf(self): | |
| ids = (os.getpgrp(), os.getsid(0)) | |
| with open("/dev/tty", "rb") as tty: | |
| buf = bytearray(b" "*8) | |
| save_buf = bytes(buf) | |
| r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf, False) | |
| self.assertEqual(bytes(buf), save_buf) | |
| self.assertIsInstance(r, bytes) | |
| rpgrp = memoryview(r).cast('i')[0] | |
| self.assertIn(rpgrp, ids) | |
| def _create_int_buf(self, nbytes=None): | |
| buf = array.array('i') | |
| intsize = buf.itemsize | |
| # A fill value unlikely to be in `ids` | |
| fill = -12345 | |
| if nbytes is not None: | |
| # Extend the buffer so that it is exactly `nbytes` bytes long | |
| buf.extend([fill] * (nbytes // intsize)) | |
| self.assertEqual(len(buf) * intsize, nbytes) # sanity check | |
| else: | |
| buf.append(fill) | |
| return buf | |
| def _check_ioctl_mutate_len(self, nbytes=None): | |
| ids = (os.getpgrp(), os.getsid(0)) | |
| buf = self._create_int_buf(nbytes) | |
| with open("/dev/tty", "rb") as tty: | |
| r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf) | |
| rpgrp = buf[0] | |
| self.assertEqual(r, 0) | |
| self.assertIn(rpgrp, ids) | |
| def _check_ioctl_not_mutate_len(self, nbytes=None): | |
| ids = (os.getpgrp(), os.getsid(0)) | |
| buf = self._create_int_buf(nbytes) | |
| save_buf = bytes(buf) | |
| with open("/dev/tty", "rb") as tty: | |
| r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf, False) | |
| self.assertIsInstance(r, bytes) | |
| self.assertEqual(len(r), len(save_buf)) | |
| self.assertEqual(bytes(buf), save_buf) | |
| rpgrp = array.array('i', r)[0] | |
| rpgrp = memoryview(r).cast('i')[0] | |
| self.assertIn(rpgrp, ids) | |
| buf = bytes(buf) | |
| with open("/dev/tty", "rb") as tty: | |
| r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf, True) | |
| self.assertIsInstance(r, bytes) | |
| self.assertEqual(len(r), len(save_buf)) | |
| self.assertEqual(buf, save_buf) | |
| rpgrp = array.array('i', r)[0] | |
| rpgrp = memoryview(r).cast('i')[0] | |
| self.assertIn(rpgrp, ids) | |
| def test_ioctl_mutate(self): | |
| self._check_ioctl_mutate_len() | |
| self._check_ioctl_not_mutate_len() | |
| def test_ioctl_mutate_1024(self): | |
| # Issue #9758: a mutable buffer of exactly 1024 bytes wouldn't be | |
| # copied back after the system call. | |
| self._check_ioctl_mutate_len(1024) | |
| self._check_ioctl_not_mutate_len(1024) | |
| def test_ioctl_mutate_2048(self): | |
| self._check_ioctl_mutate_len(2048) | |
| self._check_ioctl_not_mutate_len(1024) | |
| @unittest.skipUnless(hasattr(os, 'openpty'), "need os.openpty()") | |
| class IoctlTestsPty(unittest.TestCase): | |
| def setUp(self): | |
| self.master_fd, self.slave_fd = os.openpty() | |
| self.addCleanup(os.close, self.slave_fd) | |
| self.addCleanup(os.close, self.master_fd) | |
| @unittest.skipUnless(hasattr(termios, 'TCFLSH'), 'requires termios.TCFLSH') | |
| @unittest.skipIf(sys.platform == 'cygwin', 'test failed on Cygwin') | |
| def test_ioctl_clear_input_or_output(self): | |
| wfd = self.slave_fd | |
| rfd = self.master_fd | |
| # The data is buffered in the input buffer on Linux, and in | |
| # the output buffer on other platforms. | |
| inbuf = sys.platform in ('linux', 'android') | |
| os.write(wfd, b'abcdef') | |
| self.assertEqual(os.read(rfd, 2), b'ab') | |
| if inbuf: | |
| # don't flush input | |
| fcntl.ioctl(rfd, termios.TCFLSH, termios.TCOFLUSH) | |
| else: | |
| # don't flush output | |
| fcntl.ioctl(wfd, termios.TCFLSH, termios.TCIFLUSH) | |
| self.assertEqual(os.read(rfd, 2), b'cd') | |
| if inbuf: | |
| # flush input | |
| fcntl.ioctl(rfd, termios.TCFLSH, termios.TCIFLUSH) | |
| else: | |
| # flush output | |
| fcntl.ioctl(wfd, termios.TCFLSH, termios.TCOFLUSH) | |
| os.write(wfd, b'ABCDEF') | |
| self.assertEqual(os.read(rfd, 1024), b'ABCDEF') | |
| @support.skip_android_selinux('tcflow') | |
| @unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on Linux') | |
| @unittest.skipUnless(hasattr(termios, 'TCXONC'), 'requires termios.TCXONC') | |
| def test_ioctl_suspend_and_resume_output(self): | |
| wfd = self.slave_fd | |
| rfd = self.master_fd | |
| write_suspended = threading.Event() | |
| write_finished = threading.Event() | |
| def writer(): | |
| os.write(wfd, b'abc') | |
| self.assertTrue(write_suspended.wait(support.SHORT_TIMEOUT)) | |
| os.write(wfd, b'def') | |
| write_finished.set() | |
| with threading_helper.start_threads([threading.Thread(target=writer)]): | |
| self.assertEqual(os.read(rfd, 3), b'abc') | |
| try: | |
| try: | |
| fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF) | |
| finally: | |
| write_suspended.set() | |
| self.assertFalse(write_finished.wait(0.5), | |
| 'output was not suspended') | |
| finally: | |
| fcntl.ioctl(wfd, termios.TCXONC, termios.TCOON) | |
| self.assertTrue(write_finished.wait(support.SHORT_TIMEOUT), | |
| 'output was not resumed') | |
| self.assertEqual(os.read(rfd, 1024), b'def') | |
| def test_ioctl_set_window_size(self): | |
| # (rows, columns, xpixel, ypixel) | |
| our_winsz = struct.pack("HHHH", 20, 40, 0, 0) | |
| result = fcntl.ioctl(self.master_fd, termios.TIOCSWINSZ, our_winsz) | |
| new_winsz = struct.unpack("HHHH", result) | |
| self.assertEqual(new_winsz[:2], (20, 40)) | |
| @unittest.skipUnless(hasattr(fcntl, 'FICLONE'), 'need fcntl.FICLONE') | |
| def test_bad_fd(self): | |
| # gh-134744: Test error handling | |
| fd = os_helper.make_bad_fd() | |
| with self.assertRaises(OSError): | |
| fcntl.ioctl(fd, fcntl.FICLONE, fd) | |
| with self.assertRaises(OSError): | |
| fcntl.ioctl(fd, fcntl.FICLONE, b'\0' * 10) | |
| with self.assertRaises(OSError): | |
| fcntl.ioctl(fd, fcntl.FICLONE, b'\0' * 2048) | |
| if __name__ == "__main__": | |
| unittest.main() |