Avoid the multiprocessing forkserver method

Author: hroncok

tests/test_multiprocessing.py

@@ -18,6 +18,15 @@ def exit_callback(callback, *args):
 
 
 class TestMultiprocessing:
+
+    # Python 3.14 changed the non-macOS POSIX default to forkserver
+    # but the code in this module does not work with it
+    # See https://github.com/python/cpython/issues/125714
+    if multiprocessing.get_start_method() == 'forkserver':
+        _mp_context = multiprocessing.get_context(method='fork')
+    else:
+        _mp_context = multiprocessing.get_context()
+
     # Test connection sharing between forks.
     # See issue #1085 for details.
 
@@ -41,7 +50,7 @@ def target(conn):
             assert conn.read_response() == b"PONG"
             conn.disconnect()
 
-        proc = multiprocessing.Process(target=target, args=(conn,))
+        proc = self._mp_context.Process(target=target, args=(conn,))
         proc.start()
         proc.join(3)
         assert proc.exitcode == 0
@@ -71,7 +80,7 @@ def target(conn, ev):
                 conn.send_command("ping")
 
         ev = multiprocessing.Event()
-        proc = multiprocessing.Process(target=target, args=(conn, ev))
+        proc = self._mp_context.Process(target=target, args=(conn, ev))
         proc.start()
 
         conn.disconnect()
@@ -105,7 +114,7 @@ def target(pool):
                     assert conn.send_command("ping") is None
                     assert conn.read_response() == b"PONG"
 
-        proc = multiprocessing.Process(target=target, args=(pool,))
+        proc = self._mp_context.Process(target=target, args=(pool,))
         proc.start()
         proc.join(3)
         assert proc.exitcode == 0
@@ -143,7 +152,7 @@ def target(pool, disconnect_event):
 
         ev = multiprocessing.Event()
 
-        proc = multiprocessing.Process(target=target, args=(pool, ev))
+        proc = self._mp_context.Process(target=target, args=(pool, ev))
         proc.start()
 
         pool.disconnect()
@@ -159,7 +168,7 @@ def target(client):
             assert client.ping() is True
             del client
 
-        proc = multiprocessing.Process(target=target, args=(r,))
+        proc = self._mp_context.Process(target=target, args=(r,))
         proc.start()
         proc.join(3)
         assert proc.exitcode == 0