Fix ctypes from_buffer writes and unskip passing MP tests (#8744)

* Fix ctypes from_buffer writes and unskip passing MP tests

Write char-array .value/.raw through borrowed buffers instead of
Cow::to_mut(), which copied shared memory into a private buffer.

Unskip connection pickling, sharedctypes, Condition notify, and
concurrent.futures interpreter-shutdown tests. Cover forkserver
Condition notify and shared ctypes char arrays in extra_tests.

Assisted-by: Grok:4.6

* Unskip linux RLock repr; keep fork pickling hang skip

test_repr_rlock passed on Linux spawn/fork/forkserver. Connection
pickling hangs under the fork start method, so skip it only there.

Build the apple-container image from rust:bookworm; rust:bullseye
apt indexes 404.

Assisted-by: Grok:4.6

.devcontainer/Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:bullseye
+FROM rust:bookworm
 
 # Install clang
 RUN apt-get update \

Lib/test/_test_multiprocessing.py

@@ -4084,7 +4084,6 @@ class _TestPicklingConnections(BaseTestCase):
 
         conn.close()
 
-    @unittest.skip("TODO: RUSTPYTHON; hangs")
     def test_pickling(self):
         families = self.connection.families
 
@@ -4306,7 +4305,6 @@ class _TestSharedCTypes(BaseTestCase):
         for i in range(len(arr)):
             arr[i] *= 2
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; - ctypes Structure shared memory not working
     def test_sharedctypes(self, lock=False):
         x = Value('i', 7, lock=lock)
         y = Value(c_double, 1.0/3.0, lock=lock)
@@ -4330,7 +4328,6 @@ class _TestSharedCTypes(BaseTestCase):
             self.assertAlmostEqual(arr[i], i*2)
         self.assertEqual(string.value, latin('hellohello'))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; - calls test_sharedctypes which fails
     def test_synchronize(self):
         self.test_sharedctypes(lock=True)
 

Lib/test/test_concurrent_futures/test_shutdown.py

@@ -27,7 +27,6 @@ class ExecutorShutdownTest:
                           self.executor.submit,
                           pow, 2, 5)
 
-    @unittest.skip('TODO: RUSTPYTHON; hangs')
     def test_interpreter_shutdown(self):
         # Test the atexit hook for shutdown of worker threads and processes
         rc, out, err = assert_python_ok('-c', """if 1:
@@ -50,7 +49,6 @@ class ExecutorShutdownTest:
         self.assertFalse(err)
         self.assertEqual(out.strip(), b"apple")
 
-    @unittest.skip('TODO: RUSTPYTHON; Hangs')
     def test_submit_after_interpreter_shutdown(self):
         # Test the atexit hook for shutdown of worker threads and processes
         rc, out, err = assert_python_ok('-c', """if 1:

Lib/test/test_multiprocessing_fork/test_processes.py

@@ -3,5 +3,9 @@ from test._test_multiprocessing import install_tests_in_module_dict
 
 install_tests_in_module_dict(globals(), 'fork', only_type="processes")
 
+class WithProcessesTestPicklingConnections(WithProcessesTestPicklingConnections):  # TODO: RUSTPYTHON
+    @unittest.skip('TODO: RUSTPYTHON; hangs')
+    def test_pickling(self): super().test_pickling()
+
 if __name__ == '__main__':
     unittest.main()

Lib/test/test_multiprocessing_forkserver/test_processes.py

@@ -3,19 +3,5 @@ from test._test_multiprocessing import install_tests_in_module_dict
 
 install_tests_in_module_dict(globals(), 'forkserver', only_type="processes")
 
-import os, sys  # TODO: RUSTPYTHON
-class WithProcessesTestCondition(WithProcessesTestCondition):  # TODO: RUSTPYTHON
-    @unittest.skip('TODO: RUSTPYTHON flaky timeout')
-    def test_notify(self): super().test_notify()
-    @unittest.skip('TODO: RUSTPYTHON flaky timeout')
-    def test_notify_n(self): super().test_notify_n()
-
-class WithProcessesTestLock(WithProcessesTestLock):  # TODO: RUSTPYTHON
-    @unittest.skipIf(  # TODO: RUSTPYTHON
-        sys.platform == 'linux',  # TODO: RUSTPYTHON
-        'TODO: RUSTPYTHON flaky BrokenPipeError, flaky ConnectionRefusedError, flaky ConnectionResetError, flaky EOFError'
-    )  # TODO: RUSTPYTHON
-    def test_repr_rlock(self): super().test_repr_rlock()  # TODO: RUSTPYTHON
-
 if __name__ == '__main__':
     unittest.main()

Lib/test/test_multiprocessing_spawn/test_processes.py

@@ -3,17 +3,5 @@ from test._test_multiprocessing import install_tests_in_module_dict
 
 install_tests_in_module_dict(globals(), 'spawn', only_type="processes")
 
-import os, sys  # TODO: RUSTPYTHON
-class WithProcessesTestCondition(WithProcessesTestCondition):  # TODO: RUSTPYTHON
-    @unittest.skipIf(sys.platform == 'darwin', 'TODO: RUSTPYTHON flaky timeout')
-    def test_notify(self): super().test_notify()
-
-class WithProcessesTestLock(WithProcessesTestLock):  # TODO: RUSTPYTHON
-    @unittest.skipIf(  # TODO: RUSTPYTHON
-        sys.platform == 'linux',  # TODO: RUSTPYTHON
-        'TODO: RUSTPYTHON flaky BrokenPipeError, flaky ConnectionRefusedError, flaky ConnectionResetError, flaky EOFError'
-    )  # TODO: RUSTPYTHON
-    def test_repr_rlock(self): super().test_repr_rlock()  # TODO: RUSTPYTHON
-
 if __name__ == '__main__':
     unittest.main()

crates/vm/src/stdlib/_ctypes/array.rs

@@ -23,6 +23,15 @@ use rustpython_host_env::ctypes::{
     write_wchar_array_value, wstring_from_bytes, zeroed_bytes,
 };
 
+fn cow_bytes_as_mut<'buf>(buffer: &'buf mut Cow<'_, [u8]>) -> &'buf mut [u8] {
+    match buffer {
+        Cow::Borrowed(slice) => unsafe {
+            rustpython_host_env::ctypes::borrowed_slice_as_mut(slice)
+        },
+        Cow::Owned(vec) => vec.as_mut_slice(),
+    }
+}
+
 /// Get itemsize from a PEP 3118 format string
 /// Extracts the type code (last char after endianness prefix) and returns its size
 fn get_size_from_format(fmt: &str) -> usize {
@@ -1094,7 +1103,7 @@ fn char_array_set_value(obj: PyObjectRef, value: PyObjectRef, vm: &VirtualMachin
         return Err(vm.new_value_error("byte string too long"));
     }
 
-    write_char_array_value(buffer.to_mut(), src);
+    write_char_array_value(cow_bytes_as_mut(&mut buffer), src);
     Ok(())
 }
 
@@ -1119,7 +1128,7 @@ fn char_array_set_raw(
     if src.len() > buffer.len() {
         return Err(vm.new_value_error("byte string too long"));
     }
-    write_char_array_raw(buffer.to_mut(), &src);
+    write_char_array_raw(cow_bytes_as_mut(&mut buffer), &src);
     Ok(())
 }
 
@@ -1141,9 +1150,11 @@ fn wchar_array_set_value(
         .downcast_ref::<PyStr>()
         .ok_or_else(|| vm.new_type_error("unicode string expected"))?;
     let mut buffer = zelf.0.buffer.write();
-    write_wchar_array_value(buffer.to_mut(), s.as_wtf8()).map_err(|err| match err {
-        WCharArrayWriteError::TooLong => vm.new_value_error("string too long"),
-    })?;
+    write_wchar_array_value(cow_bytes_as_mut(&mut buffer), s.as_wtf8()).map_err(
+        |err| match err {
+            WCharArrayWriteError::TooLong => vm.new_value_error("string too long"),
+        },
+    )?;
     Ok(())
 }
 

extra_tests/snippets/stdlib_ctypes_from_buffer.py

@@ -0,0 +1,11 @@
+"""from_buffer char-array .value writes must mutate the exporter."""
+
+import ctypes
+
+buf = (ctypes.c_char * 20)()
+view = (ctypes.c_char * 20).from_buffer(buf)
+view.value = b"hello"
+assert buf.value == b"hello"
+view.value *= 2
+assert buf.value == b"hellohello"
+assert view.value == b"hellohello"

extra_tests/snippets/stdlib_multiprocessing_condition.py

@@ -80,8 +80,12 @@ def notify_one_then_the_other(ctx):
 
 if __name__ == "__main__":
     multiprocessing.freeze_support()
-    ctx = multiprocessing.get_context("spawn")
+    methods = ["spawn"]
+    if "forkserver" in multiprocessing.get_all_start_methods():
+        methods.append("forkserver")
     # Enough repeats to surface a lost-wakeup race in CI snippets.
-    for _ in range(20):
-        notify_all_six(ctx)
-        notify_one_then_the_other(ctx)
+    for method in methods:
+        ctx = multiprocessing.get_context(method)
+        for _ in range(20):
+            notify_all_six(ctx)
+            notify_one_then_the_other(ctx)

extra_tests/snippets/stdlib_multiprocessing_sharedctypes.py

@@ -0,0 +1,35 @@
+"""Shared ctypes char array .value writes must be visible across processes."""
+
+import multiprocessing
+from ctypes import Structure, c_double, c_int, c_longlong
+
+
+class Foo(Structure):
+    _fields_ = [("x", c_int), ("y", c_double), ("z", c_longlong)]
+
+
+def double(foo, string):
+    foo.x *= 2
+    foo.y *= 2
+    string.value *= 2
+
+
+if __name__ == "__main__":
+    multiprocessing.freeze_support()
+    methods = ["spawn"]
+    if "forkserver" in multiprocessing.get_all_start_methods():
+        methods.append("forkserver")
+    if "fork" in multiprocessing.get_all_start_methods():
+        methods.append("fork")
+    for method in methods:
+        ctx = multiprocessing.get_context(method)
+        foo = ctx.Value(Foo, 3, 2.0)
+        string = ctx.Array("c", 20)
+        string.value = b"hello"
+        proc = ctx.Process(target=double, args=(foo, string))
+        proc.start()
+        proc.join(timeout=10)
+        assert not proc.is_alive(), method
+        assert foo.x == 6, (method, foo.x)
+        assert abs(foo.y - 4.0) < 1e-9, (method, foo.y)
+        assert string.value == b"hellohello", (method, string.value)