Register latin-1 codec and fix stdio bootstrap

Lib/test/test_sys.py

@@ -209,7 +209,6 @@ class SysModuleTest(unittest.TestCase):
     def tearDown(self):
         test.support.reap_children()
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; latin-1 codec not registered
     def test_exit(self):
         # call with two arguments
         self.assertRaises(TypeError, sys.exit, 42, 42)
@@ -1057,12 +1056,10 @@ class SysModuleTest(unittest.TestCase):
                          'stdout: surrogateescape\n'
                          'stderr: backslashreplace\n')
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; stderr: backslashreplace
     @support.requires_subprocess()
     def test_c_locale_surrogateescape(self):
         self.check_locale_surrogateescape('C')
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; stderr: backslashreplace
     @support.requires_subprocess()
     def test_posix_locale_surrogateescape(self):
         self.check_locale_surrogateescape('POSIX')

crates/vm/src/stdlib/io.rs

@@ -2607,6 +2607,92 @@ mod _io {
         }
     }
 
+    #[pyclass(module = "_io", name, no_attr)]
+    #[derive(Debug, PyPayload)]
+    struct StatelessIncrementalEncoder {
+        encode: PyObjectRef,
+        errors: Option<PyStrRef>,
+        name: Option<PyStrRef>,
+    }
+
+    #[pyclass]
+    impl StatelessIncrementalEncoder {
+        #[pymethod]
+        fn encode(
+            &self,
+            input: PyObjectRef,
+            _final: OptionalArg<bool>,
+            vm: &VirtualMachine,
+        ) -> PyResult {
+            let mut args: Vec<PyObjectRef> = vec![input];
+            if let Some(errors) = &self.errors {
+                args.push(errors.to_owned().into());
+            }
+            let res = self.encode.call(args, vm)?;
+            let tuple: PyTupleRef = res.try_into_value(vm)?;
+            if tuple.len() != 2 {
+                return Err(vm.new_type_error("encoder must return a tuple (object, integer)"));
+            }
+            Ok(tuple[0].clone())
+        }
+
+        #[pymethod]
+        fn reset(&self) {}
+
+        #[pymethod]
+        fn setstate(&self, _state: PyObjectRef) {}
+
+        #[pymethod]
+        fn getstate(&self, vm: &VirtualMachine) -> PyObjectRef {
+            vm.ctx.new_int(0).into()
+        }
+
+        #[pygetset]
+        fn name(&self) -> Option<PyStrRef> {
+            self.name.clone()
+        }
+    }
+
+    #[pyclass(module = "_io", name, no_attr)]
+    #[derive(Debug, PyPayload)]
+    struct StatelessIncrementalDecoder {
+        decode: PyObjectRef,
+        errors: Option<PyStrRef>,
+    }
+
+    #[pyclass]
+    impl StatelessIncrementalDecoder {
+        #[pymethod]
+        fn decode(
+            &self,
+            input: PyObjectRef,
+            _final: OptionalArg<bool>,
+            vm: &VirtualMachine,
+        ) -> PyResult {
+            let mut args: Vec<PyObjectRef> = vec![input];
+            if let Some(errors) = &self.errors {
+                args.push(errors.to_owned().into());
+            }
+            let res = self.decode.call(args, vm)?;
+            let tuple: PyTupleRef = res.try_into_value(vm)?;
+            if tuple.len() != 2 {
+                return Err(vm.new_type_error("decoder must return a tuple (object, integer)"));
+            }
+            Ok(tuple[0].clone())
+        }
+
+        #[pymethod]
+        fn getstate(&self, vm: &VirtualMachine) -> (PyBytesRef, u64) {
+            (vm.ctx.empty_bytes.to_owned(), 0)
+        }
+
+        #[pymethod]
+        fn setstate(&self, _state: PyTupleRef, _vm: &VirtualMachine) {}
+
+        #[pymethod]
+        fn reset(&self) {}
+    }
+
     #[pyattr]
     #[pyclass(name = "TextIOWrapper", base = _TextIOBase)]
     #[derive(Debug, Default)]
@@ -2830,7 +2916,25 @@ mod _io {
 
             let encoder = if vm.call_method(buffer, "writable", ())?.try_to_bool(vm)? {
                 let incremental_encoder =
-                    codec.get_incremental_encoder(Some(errors.to_owned()), vm)?;
+                    match codec.get_incremental_encoder(Some(errors.to_owned()), vm) {
+                        Ok(encoder) => encoder,
+                        Err(err)
+                            if err.fast_isinstance(vm.ctx.exceptions.type_error)
+                                || err.fast_isinstance(vm.ctx.exceptions.attribute_error) =>
+                        {
+                            let name = vm
+                                .get_attribute_opt(codec.as_tuple().to_owned().into(), "name")?
+                                .and_then(|obj| obj.downcast::<PyStr>().ok());
+                            StatelessIncrementalEncoder {
+                                encode: codec.get_encode_func().to_owned(),
+                                errors: Some(errors.to_owned()),
+                                name,
+                            }
+                            .into_ref(&vm.ctx)
+                            .into()
+                        }
+                        Err(err) => return Err(err),
+                    };
                 let encoding_name = vm.get_attribute_opt(incremental_encoder.clone(), "name")?;
                 let encode_func = encoding_name.and_then(|name| {
                     let name = name.downcast_ref::<PyStr>()?;
@@ -2845,7 +2949,21 @@ mod _io {
             };
 
             let decoder = if vm.call_method(buffer, "readable", ())?.try_to_bool(vm)? {
-                let decoder = codec.get_incremental_decoder(Some(errors.to_owned()), vm)?;
+                let decoder = match codec.get_incremental_decoder(Some(errors.to_owned()), vm) {
+                    Ok(decoder) => decoder,
+                    Err(err)
+                        if err.fast_isinstance(vm.ctx.exceptions.type_error)
+                            || err.fast_isinstance(vm.ctx.exceptions.attribute_error) =>
+                    {
+                        StatelessIncrementalDecoder {
+                            decode: codec.get_decode_func().to_owned(),
+                            errors: Some(errors.to_owned()),
+                        }
+                        .into_ref(&vm.ctx)
+                        .into()
+                    }
+                    Err(err) => return Err(err),
+                };
                 if let Newlines::Universal | Newlines::Passthrough = newline {
                     let args = IncrementalNewlineDecoderArgs {
                         decoder,

crates/vm/src/stdlib/sys.rs

@@ -73,7 +73,7 @@ mod sys {
         RUST_MULTIARCH.replace("-unknown", "")
     }
 
-    #[pyclass(no_attr, name = "_BootstrapStderr", module = "sys")]
+    #[pyclass(no_attr, name = "_BootstrapStderr")]
     #[derive(Debug, PyPayload)]
     pub(super) struct BootstrapStderr;
 
@@ -95,7 +95,7 @@ mod sys {
 
     /// Lightweight stdio wrapper for sandbox mode (no host_env).
     /// Directly uses Rust's std::io for stdin/stdout/stderr without FileIO.
-    #[pyclass(no_attr, name = "_SandboxStdio", module = "sys")]
+    #[pyclass(no_attr, name = "_SandboxStdio")]
     #[derive(Debug, PyPayload)]
     pub struct SandboxStdio {
         pub fd: i32,

crates/vm/src/vm/mod.rs

@@ -331,6 +331,21 @@ impl VirtualMachine {
         self.state
             .codec_registry
             .register_manual("utf8", utf8_codec)?;
+
+        // Register latin-1 / iso8859-1 aliases needed very early for stdio
+        // bootstrap (e.g. PYTHONIOENCODING=latin-1).
+        if cfg!(feature = "freeze-stdlib") {
+            self.import("encodings.latin_1", 0)?;
+            let latin1_module = sys_modules.get_item("encodings.latin_1", self)?;
+            let getregentry = latin1_module.get_attr("getregentry", self)?;
+            let codec_info = getregentry.call((), self)?;
+            let latin1_codec: crate::codecs::PyCodec = codec_info.try_into_value(self)?;
+            for name in ["latin-1", "latin_1", "latin1", "iso8859-1", "iso8859_1"] {
+                self.state
+                    .codec_registry
+                    .register_manual(name, latin1_codec.clone())?;
+            }
+        }
         Ok(())
     }
 
@@ -367,8 +382,8 @@ impl VirtualMachine {
                 let io = import::import_builtin(self, "_io")?;
 
                 // Full stdio: FileIO → BufferedWriter → TextIOWrapper
-                #[cfg(feature = "host_env")]
-                let make_stdio = |name: &str, fd: i32, write: bool| {
+                #[cfg(all(feature = "host_env", feature = "stdio"))]
+                let make_stdio = |name: &str, fd: i32, write: bool| -> PyResult<PyObjectRef> {
                     let buffered_stdio = self.state.config.settings.buffered_stdio;
                     let unbuffered = write && !buffered_stdio;
                     let buf = crate::stdlib::io::open(
@@ -397,12 +412,13 @@ impl VirtualMachine {
                     let errors = if fd == 2 {
                         Some("backslashreplace")
                     } else {
-                        self.state
-                            .config
-                            .settings
-                            .stdio_errors
-                            .as_deref()
-                            .or(Some("surrogateescape"))
+                        self.state.config.settings.stdio_errors.as_deref().or(
+                            if self.state.config.settings.stdio_encoding.is_some() {
+                                Some("strict")
+                            } else {
+                                Some("surrogateescape")
+                            },
+                        )
                     };
 
                     let stdio = self.call_method(