Merge pull request RustPython#4140 from tgsong827/fcntl.ioctl

update ioctl first arg

Author: youknowone

Lib/test/test_ioctl.py

@@ -0,0 +1,94 @@
+import array
+import unittest
+from test.support import get_attribute
+from test.support.import_helper import import_module
+import os, struct
+fcntl = import_module('fcntl')
+termios = import_module('termios')
+get_attribute(termios, 'TIOCGPGRP') #Can't run tests without this feature
+
+try:
+    tty = open("/dev/tty", "rb")
+except OSError:
+    raise unittest.SkipTest("Unable to open /dev/tty")
+else:
+    with tty:
+        # Skip if another process is in foreground
+        r = fcntl.ioctl(tty, termios.TIOCGPGRP, "    ")
+    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")
+    del tty, r, rpgrp
+
+try:
+    import pty
+except ImportError:
+    pty = None
+
+class IoctlTests(unittest.TestCase):
+    def test_ioctl(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:
+            r = fcntl.ioctl(tty, termios.TIOCGPGRP, "    ")
+            rpgrp = struct.unpack("i", r)[0]
+            self.assertIn(rpgrp, ids)
+
+    def _check_ioctl_mutate_len(self, nbytes=None):
+        buf = array.array('i')
+        intsize = buf.itemsize
+        ids = (os.getpgrp(), os.getsid(0))
+        # 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)
+        with open("/dev/tty", "rb") as tty:
+            r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf, True)
+        rpgrp = buf[0]
+        self.assertEqual(r, 0)
+        self.assertIn(rpgrp, ids)
+
+    def test_ioctl_mutate(self):
+        self._check_ioctl_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)
+
+    def test_ioctl_mutate_2048(self):
+        # Test with a larger buffer, just for the record.
+        self._check_ioctl_mutate_len(2048)
+
+    # TODO: RUSTPYTHON
+    @unittest.expectedFailure
+    def test_ioctl_signed_unsigned_code_param(self):
+        if not pty:
+            raise unittest.SkipTest('pty module required')
+        mfd, sfd = pty.openpty()
+        try:
+            if termios.TIOCSWINSZ < 0:
+                set_winsz_opcode_maybe_neg = termios.TIOCSWINSZ
+                set_winsz_opcode_pos = termios.TIOCSWINSZ & 0xffffffff
+            else:
+                set_winsz_opcode_pos = termios.TIOCSWINSZ
+                set_winsz_opcode_maybe_neg, = struct.unpack("i",
+                        struct.pack("I", termios.TIOCSWINSZ))
+
+            our_winsz = struct.pack("HHHH",80,25,0,0)
+            # test both with a positive and potentially negative ioctl code
+            new_winsz = fcntl.ioctl(mfd, set_winsz_opcode_pos, our_winsz)
+            new_winsz = fcntl.ioctl(mfd, set_winsz_opcode_maybe_neg, our_winsz)
+        finally:
+            os.close(mfd)
+            os.close(sfd)
+
+
+if __name__ == "__main__":
+    unittest.main()

stdlib/src/fcntl.rs

@@ -52,6 +52,7 @@ mod fcntl {
     #[cfg(any(target_os = "dragonfly", target_os = "netbsd", target_vendor = "apple"))]
     #[pyattr]
     use libc::F_GETPATH;
+    use rustpython_vm::PyObjectRef;
 
     #[pyfunction]
     fn fcntl(
@@ -89,12 +90,21 @@ mod fcntl {
 
     #[pyfunction]
     fn ioctl(
-        fd: i32,
+        obj: PyObjectRef,
         request: u32,
         arg: OptionalArg<Either<Either<ArgMemoryBuffer, ArgStrOrBytesLike>, i32>>,
         mutate_flag: OptionalArg<bool>,
         vm: &VirtualMachine,
     ) -> PyResult {
+        let fd = obj.try_to_value(vm).or_else(|_| {
+            let meth = vm.get_method_or_type_error(
+                obj.clone(),
+                vm.ctx.interned_str("fileno").unwrap(),
+                || "ioctl first arg must be an int or object with a fileno() method".to_owned(),
+            )?;
+            vm.invoke(&meth, ())?.try_into_value(vm)
+        })?;
+
         let arg = arg.unwrap_or_else(|| Either::B(0));
         match arg {
             Either::A(buf_kind) => {