winapi._findfirstfile,nt.chmod (#6401)

Lib/test/test_compileall.py

@@ -459,7 +459,6 @@ class CompileallTestsBase:
                 except Exception:
                     pass
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @os_helper.skip_unless_symlink
     def test_ignore_symlink_destination(self):
         # Create folders for allowed files, symlinks and prohibited area
@@ -933,7 +932,6 @@ class CommandLineTestsBase:
                 except Exception:
                     pass
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @os_helper.skip_unless_symlink
     def test_ignore_symlink_destination(self):
         # Create folders for allowed files, symlinks and prohibited area

Lib/test/test_ntpath.py

@@ -650,8 +650,6 @@ class TestNtpath(NtpathTestCase):
         os.symlink(ABSTFN, ntpath.relpath(ABSTFN + "1"))
         self.assertPathEqual(ntpath.realpath(ABSTFN + "1", **kwargs), ABSTFN)
 
-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
     @os_helper.skip_unless_symlink
     @unittest.skipUnless(HAVE_GETFINALPATHNAME, 'need _getfinalpathname')
     def test_realpath_broken_symlinks(self):
@@ -1385,6 +1383,7 @@ class TestNtpath(NtpathTestCase):
             self.assertIsInstance(b_final_path, bytes)
             self.assertGreater(len(b_final_path), 0)
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
     @unittest.skipIf(sys.platform != 'win32', "Can only test junctions with creation on win32.")
     def test_isjunction(self):
         with os_helper.temp_dir() as d:

Lib/test/test_os.py

@@ -3144,7 +3144,6 @@ class Win32JunctionTests(unittest.TestCase):
         self.assertEqual(os.path.normcase("\\\\?\\" + self.junction_target),
                          os.path.normcase(os.readlink(self.junction)))
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON; (AttributeError: module '_winapi' has no attribute 'CreateJunction')")
     def test_unlink_removes_junction(self):
         _winapi.CreateJunction(self.junction_target, self.junction)
         self.assertTrue(os.path.exists(self.junction))

Lib/test/test_posix.py

@@ -1063,7 +1063,6 @@ class PosixTester(unittest.TestCase):
         self.assertEqual(os.stat(target).st_mode, target_mode)
         self.assertEqual(os.lstat(link).st_mode, new_mode)
 
-    @unittest.expectedFailureIfWindows('TODO: RUSTPYTHON')
     @os_helper.skip_unless_symlink
     def test_chmod_file_symlink(self):
         target = os_helper.TESTFN

Lib/test/test_shutil.py

@@ -257,7 +257,6 @@ class TestRmTree(BaseTest, unittest.TestCase):
         self.assertTrue(os.path.exists(dir3))
         self.assertTrue(os.path.exists(file1))
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @unittest.skipUnless(_winapi, 'only relevant on Windows')
     def test_rmtree_fails_on_junctions_onerror(self):
         tmp = self.mkdtemp()
@@ -278,7 +277,6 @@ class TestRmTree(BaseTest, unittest.TestCase):
         self.assertEqual(errors[0][1], link)
         self.assertIsInstance(errors[0][2][1], OSError)
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @unittest.skipUnless(_winapi, 'only relevant on Windows')
     def test_rmtree_fails_on_junctions_onexc(self):
         tmp = self.mkdtemp()
@@ -299,7 +297,6 @@ class TestRmTree(BaseTest, unittest.TestCase):
         self.assertEqual(errors[0][1], link)
         self.assertIsInstance(errors[0][2], OSError)
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @unittest.skipUnless(_winapi, 'only relevant on Windows')
     def test_rmtree_works_on_junctions(self):
         tmp = self.mkdtemp()
@@ -659,7 +656,6 @@ class TestRmTree(BaseTest, unittest.TestCase):
         finally:
             shutil.rmtree(TESTFN, ignore_errors=True)
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @unittest.skipUnless(_winapi, 'only relevant on Windows')
     def test_rmtree_on_junction(self):
         os.mkdir(TESTFN)
@@ -977,7 +973,6 @@ class TestCopyTree(BaseTest, unittest.TestCase):
         shutil.copytree(src_dir, dst_dir, copy_function=_copy)
         self.assertEqual(len(copied), 2)
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @os_helper.skip_unless_symlink
     def test_copytree_dangling_symlinks(self):
         src_dir = self.mkdtemp()
@@ -1051,7 +1046,6 @@ class TestCopy(BaseTest, unittest.TestCase):
 
     ### shutil.copymode
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @os_helper.skip_unless_symlink
     def test_copymode_follow_symlinks(self):
         tmp_dir = self.mkdtemp()
@@ -2594,7 +2588,6 @@ class TestMove(BaseTest, unittest.TestCase):
         self.assertTrue(os.path.islink(final_link))
         self.assertTrue(os.path.samefile(self.src_file, final_link))
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @os_helper.skip_unless_symlink
     @mock_rename
     def test_move_dangling_symlink(self):

crates/common/src/windows.rs

@@ -1,3 +1,4 @@
+use rustpython_wtf8::Wtf8;
 use std::{
     ffi::{OsStr, OsString},
     os::windows::ffi::{OsStrExt, OsStringExt},
@@ -10,6 +11,7 @@ pub trait ToWideString {
     fn to_wide(&self) -> Vec<u16>;
     fn to_wide_with_nul(&self) -> Vec<u16>;
 }
+
 impl<T> ToWideString for T
 where
     T: AsRef<OsStr>,
@@ -22,6 +24,24 @@ where
     }
 }
 
+impl ToWideString for OsStr {
+    fn to_wide(&self) -> Vec<u16> {
+        self.encode_wide().collect()
+    }
+    fn to_wide_with_nul(&self) -> Vec<u16> {
+        self.encode_wide().chain(Some(0)).collect()
+    }
+}
+
+impl ToWideString for Wtf8 {
+    fn to_wide(&self) -> Vec<u16> {
+        self.encode_wide().collect()
+    }
+    fn to_wide_with_nul(&self) -> Vec<u16> {
+        self.encode_wide().chain(Some(0)).collect()
+    }
+}
+
 pub trait FromWideString
 where
     Self: Sized,

crates/vm/src/stdlib/nt.rs

@@ -19,7 +19,7 @@ pub(crate) mod module {
         convert::ToPyException,
         function::{Either, OptionalArg},
         ospath::OsPath,
-        stdlib::os::{_os, DirFd, FollowSymlinks, SupportFunc, TargetIsDirectory},
+        stdlib::os::{_os, DirFd, SupportFunc, TargetIsDirectory},
     };
 
     use libc::intptr_t;
@@ -137,25 +137,104 @@ pub(crate) mod module {
         environ
     }
 
-    #[pyfunction]
-    fn chmod(
+    #[derive(FromArgs)]
+    struct ChmodArgs {
+        #[pyarg(any)]
         path: OsPath,
-        dir_fd: DirFd<'_, 0>,
+        #[pyarg(any)]
         mode: u32,
-        follow_symlinks: FollowSymlinks,
-        vm: &VirtualMachine,
-    ) -> PyResult<()> {
+        #[pyarg(flatten)]
+        dir_fd: DirFd<'static, 0>,
+        #[pyarg(named, name = "follow_symlinks", optional)]
+        follow_symlinks: OptionalArg<bool>,
+    }
+
+    #[pyfunction]
+    fn chmod(args: ChmodArgs, vm: &VirtualMachine) -> PyResult<()> {
+        let ChmodArgs {
+            path,
+            mode,
+            dir_fd,
+            follow_symlinks,
+        } = args;
         const S_IWRITE: u32 = 128;
         let [] = dir_fd.0;
-        let metadata = if follow_symlinks.0 {
-            fs::metadata(&path)
-        } else {
-            fs::symlink_metadata(&path)
+
+        // On Windows, os.chmod behavior differs based on whether follow_symlinks is explicitly provided:
+        // - Not provided (default): use SetFileAttributesW on the path directly (doesn't follow symlinks)
+        // - Explicitly True: resolve symlink first, then apply permissions to target
+        // - Explicitly False: raise NotImplementedError (Windows can't change symlink permissions)
+        let actual_path: std::borrow::Cow<'_, std::path::Path> = match follow_symlinks.into_option()
+        {
+            None => {
+                // Default behavior: don't resolve symlinks, operate on path directly
+                std::borrow::Cow::Borrowed(path.as_ref())
+            }
+            Some(true) => {
+                // Explicitly follow symlinks: resolve the path first
+                match fs::canonicalize(&path) {
+                    Ok(p) => std::borrow::Cow::Owned(p),
+                    Err(_) => std::borrow::Cow::Borrowed(path.as_ref()),
+                }
+            }
+            Some(false) => {
+                // follow_symlinks=False on Windows - not supported for symlinks
+                // Check if path is a symlink
+                if let Ok(meta) = fs::symlink_metadata(&path)
+                    && meta.file_type().is_symlink()
+                {
+                    return Err(vm.new_not_implemented_error(
+                        "chmod: follow_symlinks=False is not supported on Windows for symlinks"
+                            .to_owned(),
+                    ));
+                }
+                std::borrow::Cow::Borrowed(path.as_ref())
+            }
         };
-        let meta = metadata.map_err(|err| err.to_pyexception(vm))?;
+
+        // Use symlink_metadata to avoid following dangling symlinks
+        let meta = fs::symlink_metadata(&actual_path).map_err(|err| err.to_pyexception(vm))?;
         let mut permissions = meta.permissions();
         permissions.set_readonly(mode & S_IWRITE == 0);
-        fs::set_permissions(&path, permissions).map_err(|err| err.to_pyexception(vm))
+        fs::set_permissions(&*actual_path, permissions).map_err(|err| err.to_pyexception(vm))
+    }
+
+    /// Get the real file name (with correct case) without accessing the file.
+    /// Uses FindFirstFileW to get the name as stored on the filesystem.
+    #[pyfunction]
+    fn _findfirstfile(path: OsPath, vm: &VirtualMachine) -> PyResult<PyStrRef> {
+        use crate::common::windows::ToWideString;
+        use std::os::windows::ffi::OsStringExt;
+        use windows_sys::Win32::Storage::FileSystem::{
+            FindClose, FindFirstFileW, WIN32_FIND_DATAW,
+        };
+
+        let wide_path = path.as_ref().to_wide_with_nul();
+        let mut find_data: WIN32_FIND_DATAW = unsafe { std::mem::zeroed() };
+
+        let handle = unsafe { FindFirstFileW(wide_path.as_ptr(), &mut find_data) };
+        if handle == INVALID_HANDLE_VALUE {
+            return Err(vm.new_os_error(format!(
+                "FindFirstFileW failed for path: {}",
+                path.as_ref().display()
+            )));
+        }
+
+        unsafe { FindClose(handle) };
+
+        // Convert the filename from the find data to a Rust string
+        // cFileName is a null-terminated wide string
+        let len = find_data
+            .cFileName
+            .iter()
+            .position(|&c| c == 0)
+            .unwrap_or(find_data.cFileName.len());
+        let filename = std::ffi::OsString::from_wide(&find_data.cFileName[..len]);
+        let filename_str = filename
+            .to_str()
+            .ok_or_else(|| vm.new_unicode_decode_error("filename contains invalid UTF-8"))?;
+
+        Ok(vm.ctx.new_str(filename_str).to_owned())
     }
 
     // cwait is available on MSVC only (according to CPython)

crates/vm/src/stdlib/winapi.rs

@@ -316,7 +316,7 @@ mod _winapi {
 
     #[pyfunction]
     fn NeedCurrentDirectoryForExePath(exe_name: PyStrRef) -> bool {
-        let exe_name = exe_name.as_str().to_wide_with_nul();
+        let exe_name = exe_name.as_wtf8().to_wide_with_nul();
         let return_value = unsafe {
             windows_sys::Win32::System::Environment::NeedCurrentDirectoryForExePathW(
                 exe_name.as_ptr(),
@@ -334,7 +334,7 @@ mod _winapi {
         let src_path = std::path::Path::new(src_path.as_str());
         let dest_path = std::path::Path::new(dest_path.as_str());
 
-        junction::create(dest_path, src_path).map_err(|e| e.to_pyexception(vm))
+        junction::create(src_path, dest_path).map_err(|e| e.to_pyexception(vm))
     }
 
     fn getenvironment(env: ArgMapping, vm: &VirtualMachine) -> PyResult<Vec<u16>> {
@@ -485,9 +485,9 @@ mod _winapi {
     // TODO: ctypes.LibraryLoader.LoadLibrary
     #[allow(dead_code)]
     fn LoadLibrary(path: PyStrRef, vm: &VirtualMachine) -> PyResult<isize> {
-        let path = path.as_str().to_wide_with_nul();
+        let path_wide = path.as_wtf8().to_wide_with_nul();
         let handle =
-            unsafe { windows_sys::Win32::System::LibraryLoader::LoadLibraryW(path.as_ptr()) };
+            unsafe { windows_sys::Win32::System::LibraryLoader::LoadLibraryW(path_wide.as_ptr()) };
         if handle.is_null() {
             return Err(vm.new_runtime_error("LoadLibrary failed"));
         }
@@ -520,7 +520,7 @@ mod _winapi {
         name: PyStrRef,
         vm: &VirtualMachine,
     ) -> PyResult<isize> {
-        let name_wide = name.as_str().to_wide_with_nul();
+        let name_wide = name.as_wtf8().to_wide_with_nul();
         let handle = unsafe {
             windows_sys::Win32::System::Threading::OpenMutexW(
                 desired_access,
@@ -565,13 +565,9 @@ mod _winapi {
             return Err(vm.new_value_error("unsupported flags"));
         }
 
-        // Use encode_wide() which properly handles WTF-8 (including surrogates)
-        let locale_wide: Vec<u16> = locale
-            .as_wtf8()
-            .encode_wide()
-            .chain(std::iter::once(0))
-            .collect();
-        let src_wide: Vec<u16> = src.as_wtf8().encode_wide().collect();
+        // Use ToWideString which properly handles WTF-8 (including surrogates)
+        let locale_wide = locale.as_wtf8().to_wide_with_nul();
+        let src_wide = src.as_wtf8().to_wide();
 
         if src_wide.len() > i32::MAX as usize {
             return Err(vm.new_overflow_error("input string is too long".to_string()));
@@ -648,7 +644,7 @@ mod _winapi {
     fn CreateNamedPipe(args: CreateNamedPipeArgs, vm: &VirtualMachine) -> PyResult<WinHandle> {
         use windows_sys::Win32::System::Pipes::CreateNamedPipeW;
 
-        let name_wide = args.name.as_str().to_wide_with_nul();
+        let name_wide = args.name.as_wtf8().to_wide_with_nul();
 
         let handle = unsafe {
             CreateNamedPipeW(