windows encodings

RustPython · youknowone · Feb 4, 2026 · Feb 3, 2026 · Feb 3, 2026 · Feb 3, 2026
commit 6c28d43ca4c5794ae06adfc8a75ceaf73d16b0e1
diff --git a/Lib/test/test_cmd_line_script.py b/Lib/test/test_cmd_line_script.py
@@ -620,7 +620,6 @@ def test_syntaxerror_unindented_caret_position(self):
             # Confirm that the caret is located under the '=' sign
             self.assertIn("\n    ^^^^^\n", text)
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_syntaxerror_indented_caret_position(self):
         script = textwrap.dedent("""\
             if True:

diff --git a/Lib/test/test_codecs.py b/Lib/test/test_codecs.py
@@ -3451,7 +3451,6 @@ def decode_to_bytes(*args, **kwds):
 class CodePageTest(unittest.TestCase):
     CP_UTF8 = 65001
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     def test_invalid_code_page(self):
         self.assertRaises(ValueError, codecs.code_page_encode, -1, 'a')
         self.assertRaises(ValueError, codecs.code_page_decode, -1, b'a')
@@ -3670,7 +3669,6 @@ def test_multibyte_encoding(self):
             ('[\U0010ffff\uDC80]', 'replace', b'[\xf4\x8f\xbf\xbf?]'),
         ))
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     def test_code_page_decode_flags(self):
         # Issue #36312: For some code pages (e.g. UTF-7) flags for
         # MultiByteToWideChar() must be set to 0.

diff --git a/Lib/test/test_importlib/resources/test_files.py b/Lib/test/test_importlib/resources/test_files.py
@@ -52,7 +52,6 @@ class OpenDiskTests(FilesTests, unittest.TestCase):
     def setUp(self):
         self.data = data01
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON, line ending issue")
     def test_read_bytes(self):
         super().test_read_bytes()
 
@@ -67,7 +66,6 @@ def setUp(self):
 
         self.data = namespacedata01
 
-    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON, line ending issue")
     def test_read_bytes(self):
         super().test_read_bytes()
 

diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py
@@ -3630,7 +3630,6 @@ def test_spawnl(self):
         exitcode = os.spawnl(os.P_WAIT, program, *args)
         self.assertEqual(exitcode, self.exitcode)
 
-    @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; fix spawnve on Windows")
     @requires_os_func('spawnle')
     def test_spawnle(self):
         program, args = self.create_args(with_env=True)
@@ -3659,7 +3658,6 @@ def test_spawnv(self):
         exitcode = os.spawnv(os.P_WAIT, FakePath(program), args)
         self.assertEqual(exitcode, self.exitcode)
 
-    @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; fix spawnve on Windows")
     @requires_os_func('spawnve')
     def test_spawnve(self):
         program, args = self.create_args(with_env=True)
@@ -3767,7 +3765,6 @@ def _test_invalid_env(self, spawn):
         exitcode = spawn(os.P_WAIT, program, args, newenv)
         self.assertEqual(exitcode, 0)
 
-    @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; fix spawnve on Windows")
     @requires_os_func('spawnve')
     def test_spawnve_invalid_env(self):
         self._test_invalid_env(os.spawnve)

diff --git a/crates/vm/src/exceptions.rs b/crates/vm/src/exceptions.rs
@@ -210,8 +210,8 @@ impl VirtualMachine {
         }
 
         if let Some(text) = maybe_text {
-            // if text ends with \n, remove it
-            let r_text = text.as_str().trim_end_matches('\n');
+            // if text ends with \n or \r\n, remove it
+            let r_text = text.as_str().trim_end_matches(['\n', '\r']);
             let l_text = r_text.trim_start_matches([' ', '\n', '\x0c']); // \x0c is \f
             let spaces = (r_text.len() - l_text.len()) as isize;
 

diff --git a/crates/vm/src/stdlib/codecs.rs b/crates/vm/src/stdlib/codecs.rs
@@ -629,6 +629,247 @@ mod _codecs {
         delegate_pycodecs!(oem_decode, args, vm)
     }
 
+    #[cfg(windows)]
+    #[derive(FromArgs)]
+    struct CodePageEncodeArgs {
+        #[pyarg(positional)]
+        code_page: i32,
+        #[pyarg(positional)]
+        s: PyStrRef,
+        #[pyarg(positional, optional)]
+        errors: Option<PyStrRef>,
+    }
+
+    #[cfg(windows)]
+    #[pyfunction]
+    fn code_page_encode(
+        args: CodePageEncodeArgs,
+        vm: &VirtualMachine,
+    ) -> PyResult<(Vec<u8>, usize)> {
+        use crate::common::windows::ToWideString;
+        use windows_sys::Win32::Globalization::{WC_NO_BEST_FIT_CHARS, WideCharToMultiByte};
+
+        if args.code_page < 0 {
+            return Err(vm.new_value_error("invalid code page number".to_owned()));
+        }
+        let errors = args.errors.as_ref().map(|s| s.as_str()).unwrap_or("strict");
+        let code_page = args.code_page as u32;
+        let s = match args.s.to_str() {
+            Some(s) => s,
+            None => {
+                return Err(vm.new_unicode_encode_error(format!(
+                    "'cp{code_page}' codec can't encode character: surrogates not allowed"
+                )));
+            }
+        };
+        let char_len = args.s.char_len();
+
+        if s.is_empty() {
+            return Ok((Vec::new(), char_len));
+        }
+
+        let wide: Vec<u16> = std::ffi::OsStr::new(s).to_wide();
+
+        // Some code pages (like UTF-7/8, 50220-50222, etc.) don't support WC_NO_BEST_FIT_CHARS
+        let flags = if code_page == 65000
+            || code_page == 65001
+            || code_page == 42
+            || (50220..=50222).contains(&code_page)
+            || code_page == 50225
+            || code_page == 50227
+            || code_page == 50229
+            || (57002..=57011).contains(&code_page)
+            || code_page == 54936
+        {
+            0
+        } else {
+            WC_NO_BEST_FIT_CHARS
+        };
+
+        let size = unsafe {
+            WideCharToMultiByte(
+                code_page,
+                flags,
+                wide.as_ptr(),
+                wide.len() as i32,
+                std::ptr::null_mut(),
+                0,
+                core::ptr::null(),
+                std::ptr::null_mut(),
+            )
+        };
+
+        if size == 0 {
+            let err = std::io::Error::last_os_error();
+            return Err(vm.new_os_error(format!("code_page_encode failed: {err}")));
+        }
+
+        let mut buffer = vec![0u8; size as usize];
+        let mut used_default_char: i32 = 0;
+
+        let result = unsafe {
+            WideCharToMultiByte(
+                code_page,
+                flags,
+                wide.as_ptr(),
+                wide.len() as i32,
+                buffer.as_mut_ptr().cast(),
+                size,
+                core::ptr::null(),
+                if errors == "strict" && flags != 0 {
+                    &mut used_default_char
+                } else {
+                    std::ptr::null_mut()
+                },
+            )
+        };
+
+        if result == 0 {
+            let err = std::io::Error::last_os_error();
+            return Err(vm.new_os_error(format!("code_page_encode failed: {err}")));
+        }
+
+        if errors == "strict" && used_default_char != 0 {
+            return Err(vm.new_unicode_encode_error(format!(
+                "'cp{code_page}' codec can't encode characters: invalid character"
+            )));
+        }
+
+        buffer.truncate(result as usize);
+        Ok((buffer, char_len))
+    }
+
+    #[cfg(not(windows))]
+    #[pyfunction]
+    fn code_page_encode(args: FuncArgs, vm: &VirtualMachine) -> PyResult {
+        delegate_pycodecs!(code_page_encode, args, vm)
+    }
+
+    #[cfg(windows)]
+    #[derive(FromArgs)]
+    struct CodePageDecodeArgs {
+        #[pyarg(positional)]
+        code_page: i32,
+        #[pyarg(positional)]
+        data: ArgBytesLike,
+        #[pyarg(positional, optional)]
+        errors: Option<PyStrRef>,
+        #[pyarg(positional, default = false)]
+        #[allow(dead_code)]
+        r#final: bool,
+    }
+
+    #[cfg(windows)]
+    #[pyfunction]
+    fn code_page_decode(
+        args: CodePageDecodeArgs,
+        vm: &VirtualMachine,
+    ) -> PyResult<(String, usize)> {
+        use windows_sys::Win32::Globalization::{MB_ERR_INVALID_CHARS, MultiByteToWideChar};
+
+        if args.code_page < 0 {
+            return Err(vm.new_value_error("invalid code page number".to_owned()));
+        }
+        let _errors = args.errors.as_ref().map(|s| s.as_str()).unwrap_or("strict");
+        let code_page = args.code_page as u32;
+        let data = args.data.borrow_buf();
+        let len = data.len();
+
+        if data.is_empty() {
+            return Ok((String::new(), 0));
+        }
+
+        // Some code pages don't support MB_ERR_INVALID_CHARS
+        let strict_flags = if code_page == 65000
+            || code_page == 42
+            || (50220..=50222).contains(&code_page)
+            || code_page == 50225
+            || code_page == 50227
+            || code_page == 50229
+            || (57002..=57011).contains(&code_page)
+        {
+            0
+        } else {
+            MB_ERR_INVALID_CHARS
+        };
+
+        let size = unsafe {
+            MultiByteToWideChar(
+                code_page,
+                strict_flags,
+                data.as_ptr().cast(),
+                len as i32,
+                std::ptr::null_mut(),
+                0,
+            )
+        };
+
+        if size == 0 {
+            let size = unsafe {
+                MultiByteToWideChar(
+                    code_page,
+                    0,
+                    data.as_ptr().cast(),
+                    len as i32,
+                    std::ptr::null_mut(),
+                    0,
+                )
+            };
+            if size == 0 {
+                let err = std::io::Error::last_os_error();
+                return Err(vm.new_os_error(format!("code_page_decode failed: {err}")));
+            }
+
+            let mut buffer = vec![0u16; size as usize];
+            let result = unsafe {
+                MultiByteToWideChar(
+                    code_page,
+                    0,
+                    data.as_ptr().cast(),
+                    len as i32,
+                    buffer.as_mut_ptr(),
+                    size,
+                )
+            };
+            if result == 0 {
+                let err = std::io::Error::last_os_error();
+                return Err(vm.new_os_error(format!("code_page_decode failed: {err}")));
+            }
+            buffer.truncate(result as usize);
+            let s = String::from_utf16(&buffer).map_err(|e| {
+                vm.new_unicode_decode_error(format!("code_page_decode failed: {e}"))
+            })?;
+            return Ok((s, len));
+        }
+
+        let mut buffer = vec![0u16; size as usize];
+        let result = unsafe {
+            MultiByteToWideChar(
+                code_page,
+                strict_flags,
+                data.as_ptr().cast(),
+                len as i32,
+                buffer.as_mut_ptr(),
+                size,
+            )
+        };
+        if result == 0 {
+            let err = std::io::Error::last_os_error();
+            return Err(vm.new_os_error(format!("code_page_decode failed: {err}")));
+        }
+        buffer.truncate(result as usize);
+        let s = String::from_utf16(&buffer)
+            .map_err(|e| vm.new_unicode_decode_error(format!("code_page_decode failed: {e}")))?;
+
+        Ok((s, len))
+    }
+
+    #[cfg(not(windows))]
+    #[pyfunction]
+    fn code_page_decode(args: FuncArgs, vm: &VirtualMachine) -> PyResult {
+        delegate_pycodecs!(code_page_decode, args, vm)
+    }
+
     #[pyfunction]
     fn readbuffer_encode(args: FuncArgs, vm: &VirtualMachine) -> PyResult {
         delegate_pycodecs!(readbuffer_encode, args, vm)