Antoine Pitrou | 5e38aae | 2010-09-07 16:30:09 +0000 | [diff] [blame] | 1 | import array |
Michael W. Hudson | f008998 | 2003-03-03 12:29:42 +0000 | [diff] [blame] | 2 | import unittest |
R. David Murray | a21e4ca | 2009-03-31 23:16:50 +0000 | [diff] [blame] | 3 | from test.support import run_unittest, import_module, get_attribute |
Michael W. Hudson | f008998 | 2003-03-03 12:29:42 +0000 | [diff] [blame] | 4 | import os, struct |
R. David Murray | a21e4ca | 2009-03-31 23:16:50 +0000 | [diff] [blame] | 5 | fcntl = import_module('fcntl') |
| 6 | termios = import_module('termios') |
| 7 | get_attribute(termios, 'TIOCGPGRP') #Can't run tests without this feature |
Michael W. Hudson | f008998 | 2003-03-03 12:29:42 +0000 | [diff] [blame] | 8 | |
Neal Norwitz | 26f42f6 | 2003-03-20 04:33:16 +0000 | [diff] [blame] | 9 | try: |
| 10 | tty = open("/dev/tty", "r") |
Neal Norwitz | 26f42f6 | 2003-03-20 04:33:16 +0000 | [diff] [blame] | 11 | except IOError: |
Benjamin Peterson | e549ead | 2009-03-28 21:42:05 +0000 | [diff] [blame] | 12 | raise unittest.SkipTest("Unable to open /dev/tty") |
Florent Xicluna | 39d795d | 2010-08-08 18:06:13 +0000 | [diff] [blame] | 13 | else: |
| 14 | # Skip if another process is in foreground |
| 15 | r = fcntl.ioctl(tty, termios.TIOCGPGRP, " ") |
| 16 | tty.close() |
| 17 | rpgrp = struct.unpack("i", r)[0] |
| 18 | if rpgrp not in (os.getpgrp(), os.getsid(0)): |
| 19 | raise unittest.SkipTest("Neither the process group nor the session " |
| 20 | "are attached to /dev/tty") |
| 21 | del tty, r, rpgrp |
Neal Norwitz | 26f42f6 | 2003-03-20 04:33:16 +0000 | [diff] [blame] | 22 | |
Christian Heimes | e25f35e | 2008-03-20 10:49:03 +0000 | [diff] [blame] | 23 | try: |
| 24 | import pty |
| 25 | except ImportError: |
| 26 | pty = None |
| 27 | |
Michael W. Hudson | f008998 | 2003-03-03 12:29:42 +0000 | [diff] [blame] | 28 | class IoctlTests(unittest.TestCase): |
| 29 | def test_ioctl(self): |
Guido van Rossum | 8ee3e5a | 2005-09-14 18:09:42 +0000 | [diff] [blame] | 30 | # If this process has been put into the background, TIOCGPGRP returns |
| 31 | # the session ID instead of the process group id. |
| 32 | ids = (os.getpgrp(), os.getsid(0)) |
Brett Cannon | dff6985 | 2010-10-29 23:54:28 +0000 | [diff] [blame] | 33 | with open("/dev/tty", "r") as tty: |
| 34 | r = fcntl.ioctl(tty, termios.TIOCGPGRP, " ") |
| 35 | rpgrp = struct.unpack("i", r)[0] |
| 36 | self.assertIn(rpgrp, ids) |
Michael W. Hudson | f008998 | 2003-03-03 12:29:42 +0000 | [diff] [blame] | 37 | |
Antoine Pitrou | 5e38aae | 2010-09-07 16:30:09 +0000 | [diff] [blame] | 38 | def _check_ioctl_mutate_len(self, nbytes=None): |
| 39 | buf = array.array('i') |
| 40 | intsize = buf.itemsize |
Guido van Rossum | 8ee3e5a | 2005-09-14 18:09:42 +0000 | [diff] [blame] | 41 | ids = (os.getpgrp(), os.getsid(0)) |
Antoine Pitrou | 5e38aae | 2010-09-07 16:30:09 +0000 | [diff] [blame] | 42 | # A fill value unlikely to be in `ids` |
| 43 | fill = -12345 |
| 44 | if nbytes is not None: |
| 45 | # Extend the buffer so that it is exactly `nbytes` bytes long |
| 46 | buf.extend([fill] * (nbytes // intsize)) |
| 47 | self.assertEqual(len(buf) * intsize, nbytes) # sanity check |
| 48 | else: |
| 49 | buf.append(fill) |
| 50 | with open("/dev/tty", "r") as tty: |
| 51 | r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf, 1) |
Guido van Rossum | 8ee3e5a | 2005-09-14 18:09:42 +0000 | [diff] [blame] | 52 | rpgrp = buf[0] |
Ezio Melotti | b3aedd4 | 2010-11-20 19:04:17 +0000 | [diff] [blame] | 53 | self.assertEqual(r, 0) |
Benjamin Peterson | 577473f | 2010-01-19 00:09:57 +0000 | [diff] [blame] | 54 | self.assertIn(rpgrp, ids) |
Michael W. Hudson | f008998 | 2003-03-03 12:29:42 +0000 | [diff] [blame] | 55 | |
Antoine Pitrou | 5e38aae | 2010-09-07 16:30:09 +0000 | [diff] [blame] | 56 | def test_ioctl_mutate(self): |
| 57 | self._check_ioctl_mutate_len() |
| 58 | |
| 59 | def test_ioctl_mutate_1024(self): |
| 60 | # Issue #9758: a mutable buffer of exactly 1024 bytes wouldn't be |
| 61 | # copied back after the system call. |
| 62 | self._check_ioctl_mutate_len(1024) |
| 63 | |
| 64 | def test_ioctl_mutate_2048(self): |
| 65 | # Test with a larger buffer, just for the record. |
| 66 | self._check_ioctl_mutate_len(2048) |
| 67 | |
Christian Heimes | e25f35e | 2008-03-20 10:49:03 +0000 | [diff] [blame] | 68 | def test_ioctl_signed_unsigned_code_param(self): |
| 69 | if not pty: |
Benjamin Peterson | e549ead | 2009-03-28 21:42:05 +0000 | [diff] [blame] | 70 | raise unittest.SkipTest('pty module required') |
Christian Heimes | e25f35e | 2008-03-20 10:49:03 +0000 | [diff] [blame] | 71 | mfd, sfd = pty.openpty() |
| 72 | try: |
| 73 | if termios.TIOCSWINSZ < 0: |
| 74 | set_winsz_opcode_maybe_neg = termios.TIOCSWINSZ |
| 75 | set_winsz_opcode_pos = termios.TIOCSWINSZ & 0xffffffff |
| 76 | else: |
| 77 | set_winsz_opcode_pos = termios.TIOCSWINSZ |
| 78 | set_winsz_opcode_maybe_neg, = struct.unpack("i", |
| 79 | struct.pack("I", termios.TIOCSWINSZ)) |
| 80 | |
Christian Heimes | e25f35e | 2008-03-20 10:49:03 +0000 | [diff] [blame] | 81 | our_winsz = struct.pack("HHHH",80,25,0,0) |
| 82 | # test both with a positive and potentially negative ioctl code |
| 83 | new_winsz = fcntl.ioctl(mfd, set_winsz_opcode_pos, our_winsz) |
| 84 | new_winsz = fcntl.ioctl(mfd, set_winsz_opcode_maybe_neg, our_winsz) |
Christian Heimes | e25f35e | 2008-03-20 10:49:03 +0000 | [diff] [blame] | 85 | finally: |
| 86 | os.close(mfd) |
| 87 | os.close(sfd) |
| 88 | |
Michael W. Hudson | f008998 | 2003-03-03 12:29:42 +0000 | [diff] [blame] | 89 | def test_main(): |
| 90 | run_unittest(IoctlTests) |
| 91 | |
| 92 | if __name__ == "__main__": |
| 93 | test_main() |