os.waitstatus_to_exitcode for windows (RustPython#6355)

Author: youknowone

Lib/test/test_os.py

@@ -3426,7 +3426,6 @@ def test_waitstatus_to_exitcode(self):
         with self.assertRaises(TypeError):
             os.waitstatus_to_exitcode(0.0)
 
-    @unittest.expectedFailureIfWindows('TODO: RUSTPYTHON; os.spawnv not implemented yet for all platforms')
     @unittest.skipUnless(sys.platform == 'win32', 'win32-specific test')
     def test_waitpid_windows(self):
         # bpo-40138: test os.waitpid() and os.waitstatus_to_exitcode()
@@ -3435,7 +3434,6 @@ def test_waitpid_windows(self):
         code = f'import _winapi; _winapi.ExitProcess({STATUS_CONTROL_C_EXIT})'
         self.check_waitpid(code, exitcode=STATUS_CONTROL_C_EXIT)
 
-    @unittest.expectedFailureIfWindows('TODO: RUSTPYTHON; (OverflowError: Python int too large to convert to Rust i32)')
     @unittest.skipUnless(sys.platform == 'win32', 'win32-specific test')
     def test_waitstatus_to_exitcode_windows(self):
         max_exitcode = 2 ** 32 - 1

crates/vm/src/stdlib/os.rs

@@ -1444,29 +1444,32 @@ pub(super) mod _os {
         Ok((loadavg[0], loadavg[1], loadavg[2]))
     }
 
-    #[cfg(any(unix, windows))]
+    #[cfg(unix)]
     #[pyfunction]
     fn waitstatus_to_exitcode(status: i32, vm: &VirtualMachine) -> PyResult<i32> {
         let status = u32::try_from(status)
             .map_err(|_| vm.new_value_error(format!("invalid WEXITSTATUS: {status}")))?;
 
-        cfg_if::cfg_if! {
-            if #[cfg(not(windows))] {
-                let status = status as libc::c_int;
-                if libc::WIFEXITED(status) {
-                    return Ok(libc::WEXITSTATUS(status));
-                }
-
-                if libc::WIFSIGNALED(status) {
-                    return Ok(-libc::WTERMSIG(status));
-                }
+        let status = status as libc::c_int;
+        if libc::WIFEXITED(status) {
+            return Ok(libc::WEXITSTATUS(status));
+        }
 
-                Err(vm.new_value_error(format!("Invalid wait status: {status}")))
-            } else {
-                i32::try_from(status.rotate_right(8))
-                    .map_err(|_| vm.new_value_error(format!("invalid wait status: {status}")))
-            }
+        if libc::WIFSIGNALED(status) {
+            return Ok(-libc::WTERMSIG(status));
         }
+
+        Err(vm.new_value_error(format!("Invalid wait status: {status}")))
+    }
+
+    #[cfg(windows)]
+    #[pyfunction]
+    fn waitstatus_to_exitcode(status: u64, vm: &VirtualMachine) -> PyResult<u32> {
+        let exitcode = status >> 8;
+        // ExitProcess() accepts an UINT type:
+        // reject exit code which doesn't fit in an UINT
+        u32::try_from(exitcode)
+            .map_err(|_| vm.new_value_error(format!("Invalid exit code: {exitcode}")))
     }
 
     #[pyfunction]