"""Check that the system supports enough semaphores to run the test."""

# minimum number of semaphores available according to POSIX

nsems_min = 256

try:

nsems = os.sysconf("SC_SEM_NSEMS_MAX")

except (AttributeError, ValueError):

# sysconf not available or setting not available

return

if nsems == -1 or nsems >= nsems_min:

return

raise unittest.SkipTest("The OS doesn't support enough semaphores "

"""Returns a decorator: raises SkipTest when SM != spawn at test time.

def decorator(test_item):

@functools.wraps(test_item)

def spawn_check_wrapper(*args, **kwargs):

if (start_method := multiprocessing.get_start_method()) != "spawn":

raise unittest.SkipTest(f"{start_method=}, not 'spawn'; {reason}")

return test_item(*args, **kwargs)

return spawn_check_wrapper

"""Returns a decorator: raises SkipTest unless fork is supported

def decorator(test_item):

@functools.wraps(test_item)

def forkserver_check_wrapper(*args, **kwargs):

if not support.has_fork_support:

raise unittest.SkipTest("requires working os.fork()")

if (start_method := multiprocessing.get_start_method()) != "forkserver":

raise unittest.SkipTest(

f"{start_method=}, not 'forkserver'; {reason}")

return test_item(*args, **kwargs)

return forkserver_check_wrapper

"""Logic within a test suite that could errantly skip tests? Test it!"""

@support.requires_fork()

def test_only_run_in_spawn_testsuite(self):

if multiprocessing.get_start_method() != "spawn":

raise unittest.SkipTest("only run in test_multiprocessing_spawn.")

try:

@only_run_in_spawn_testsuite("testing this decorator")

def return_four_if_spawn():

return 4

except Exception as err:

self.fail(f"expected decorated `def` not to raise; caught {err}")

orig_start_method = multiprocessing.get_start_method(allow_none=True)

try:

multiprocessing.set_start_method("spawn", force=True)

self.assertEqual(return_four_if_spawn(), 4)

multiprocessing.set_start_method("fork", force=True)

with self.assertRaises(unittest.SkipTest) as ctx:

return_four_if_spawn()

self.assertIn("testing this decorator", str(ctx.exception))

self.assertIn("start_method=", str(ctx.exception))

finally:

multiprocessing.set_start_method(orig_start_method, force=True)

@support.requires_fork()

def test_only_run_in_forkserver_testsuite(self):

if multiprocessing.get_start_method() != "forkserver":

raise unittest.SkipTest("only run in test_multiprocessing_forkserver.")

try:

@only_run_in_forkserver_testsuite("testing this decorator")

def return_four_if_forkserver():

return 4

except Exception as err:

self.fail(f"expected decorated `def` not to raise; caught {err}")

orig_start_method = multiprocessing.get_start_method(allow_none=True)

try:

multiprocessing.set_start_method("forkserver", force=True)

self.assertEqual(return_four_if_forkserver(), 4)

multiprocessing.set_start_method("spawn", force=True)

with self.assertRaises(unittest.SkipTest) as ctx:

return_four_if_forkserver()

self.assertIn("testing this decorator", str(ctx.exception))

self.assertIn("start_method=", str(ctx.exception))

finally:

def __init__(self, func):

self.func = func

self.elapsed = None

def __call__(self, *args, **kwds):

t = time.monotonic()

try:

return self.func(*args, **kwds)

finally:

ALLOWED_TYPES = ('processes', 'manager', 'threads')

# If not empty, limit which start method suites run this class.

START_METHODS: set[str] = set()

start_method = None # set by install_tests_in_module_dict()

def assertTimingAlmostEqual(self, a, b):

if CHECK_TIMINGS:

self.assertAlmostEqual(a, b, 1)

def assertReturnsIfImplemented(self, value, func, *args):

try:

res = func(*args)

except NotImplementedError:

pass

else:

return self.assertEqual(value, res)

# For the sanity of Windows users, rather than crashing or freezing in

# multiple ways.

def __reduce__(self, *args):

raise NotImplementedError("shouldn't try to pickle a test case")

try:

return self.get_value()

except AttributeError:

try:

return self._Semaphore__value

except AttributeError:

try:

return self._value

except AttributeError:

ALLOWED_TYPES = ('processes', 'threads')

def test_current(self):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

current = self.current_process()

authkey = current.authkey

self.assertTrue(current.is_alive())

self.assertFalse(current.daemon)

self.assertIsInstance(authkey, bytes)

self.assertTrue(len(authkey) > 0)

self.assertEqual(current.ident, os.getpid())

self.assertEqual(current.exitcode, None)

def test_set_executable(self):

if self.TYPE == 'threads':

self.skipTest(f'test not appropriate for {self.TYPE}')

paths = [

sys.executable, # str

os.fsencode(sys.executable), # bytes

os_helper.FakePath(sys.executable), # os.PathLike

os_helper.FakePath(os.fsencode(sys.executable)), # os.PathLike bytes

]

for path in paths:

self.set_executable(path)

p = self.Process()

p.start()

p.join()

self.assertEqual(p.exitcode, 0)

@support.requires_resource('cpu')

def test_args_argument(self):

# bpo-45735: Using list or tuple as *args* in constructor could

# achieve the same effect.

args_cases = (1, "str", [1], (1,))

args_types = (list, tuple)

test_cases = itertools.product(args_cases, args_types)

for args, args_type in test_cases:

with self.subTest(args=args, args_type=args_type):

q = self.Queue(1)

# pass a tuple or list as args

p = self.Process(target=self._test_args, args=args_type((q, args)))

p.daemon = True

p.start()

child_args = q.get()

self.assertEqual(child_args, args)

p.join()

close_queue(q)

@classmethod

def _test_args(cls, q, arg):

q.put(arg)

def test_daemon_argument(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

# By default uses the current process's daemon flag.

proc0 = self.Process(target=self._test)

self.assertEqual(proc0.daemon, self.current_process().daemon)

proc1 = self.Process(target=self._test, daemon=True)

self.assertTrue(proc1.daemon)

proc2 = self.Process(target=self._test, daemon=False)

self.assertFalse(proc2.daemon)

@classmethod

def _test(cls, q, *args, **kwds):

current = cls.current_process()

q.put(args)

q.put(kwds)

q.put(current.name)

if cls.TYPE != 'threads':

q.put(bytes(current.authkey))

q.put(current.pid)

def test_parent_process_attributes(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

self.assertIsNone(self.parent_process())

rconn, wconn = self.Pipe(duplex=False)

p = self.Process(target=self._test_send_parent_process, args=(wconn,))

p.start()

p.join()

parent_pid, parent_name = rconn.recv()

self.assertEqual(parent_pid, self.current_process().pid)

self.assertEqual(parent_pid, os.getpid())

self.assertEqual(parent_name, self.current_process().name)

@classmethod

def _test_send_parent_process(cls, wconn):

from multiprocessing.process import parent_process

wconn.send([parent_process().pid, parent_process().name])

def test_parent_process(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

# Launch a child process. Make it launch a grandchild process. Kill the

# child process and make sure that the grandchild notices the death of

# its parent (a.k.a the child process).

rconn, wconn = self.Pipe(duplex=False)

p = self.Process(

target=self._test_create_grandchild_process, args=(wconn, ))

p.start()

if not rconn.poll(timeout=support.LONG_TIMEOUT):

raise AssertionError("Could not communicate with child process")

parent_process_status = rconn.recv()

self.assertEqual(parent_process_status, "alive")

p.terminate()

p.join()

if not rconn.poll(timeout=support.LONG_TIMEOUT):

raise AssertionError("Could not communicate with child process")

parent_process_status = rconn.recv()

self.assertEqual(parent_process_status, "not alive")

@classmethod

def _test_create_grandchild_process(cls, wconn):

p = cls.Process(target=cls._test_report_parent_status, args=(wconn, ))

p.start()

time.sleep(300)

@classmethod

def _test_report_parent_status(cls, wconn):

from multiprocessing.process import parent_process

wconn.send("alive" if parent_process().is_alive() else "not alive")

parent_process().join(timeout=support.SHORT_TIMEOUT)

wconn.send("alive" if parent_process().is_alive() else "not alive")

def test_process(self):

q = self.Queue(1)

e = self.Event()

args = (q, 1, 2)

kwargs = {'hello':23, 'bye':2.54}

name = 'SomeProcess'

p = self.Process(

target=self._test, args=args, kwargs=kwargs, name=name

)

p.daemon = True

current = self.current_process()

if self.TYPE != 'threads':

self.assertEqual(p.authkey, current.authkey)

self.assertEqual(p.is_alive(), False)

self.assertEqual(p.daemon, True)

self.assertNotIn(p, self.active_children())

self.assertIs(type(self.active_children()), list)

self.assertEqual(p.exitcode, None)

p.start()

self.assertEqual(p.exitcode, None)

self.assertEqual(p.is_alive(), True)

self.assertIn(p, self.active_children())

self.assertEqual(q.get(), args[1:])

self.assertEqual(q.get(), kwargs)

self.assertEqual(q.get(), p.name)

if self.TYPE != 'threads':

self.assertEqual(q.get(), current.authkey)

self.assertEqual(q.get(), p.pid)

p.join()

self.assertEqual(p.exitcode, 0)

self.assertEqual(p.is_alive(), False)

self.assertNotIn(p, self.active_children())

close_queue(q)

@unittest.skipUnless(threading._HAVE_THREAD_NATIVE_ID, "needs native_id")

def test_process_mainthread_native_id(self):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

current_mainthread_native_id = threading.main_thread().native_id

q = self.Queue(1)

p = self.Process(target=self._test_process_mainthread_native_id, args=(q,))

p.start()

child_mainthread_native_id = q.get()

p.join()

close_queue(q)

self.assertNotEqual(current_mainthread_native_id, child_mainthread_native_id)

@classmethod

def _test_process_mainthread_native_id(cls, q):

mainthread_native_id = threading.main_thread().native_id

q.put(mainthread_native_id)

@classmethod

def _sleep_some(cls):

time.sleep(100)

@classmethod

def _sleep_some_event(cls, event):

event.set()

time.sleep(100)

@classmethod

def _sleep_no_int_handler(cls, event):

signal.signal(signal.SIGINT, signal.SIG_DFL)

cls._sleep_some_event(event)

@classmethod

def _test_sleep(cls, delay):

time.sleep(delay)

def _kill_process(self, meth, target=None):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

event = self.Event()

if not target:

target = self._sleep_some_event

p = self.Process(target=target, args=(event,))

p.daemon = True

p.start()

self.assertEqual(p.is_alive(), True)

self.assertIn(p, self.active_children())

self.assertEqual(p.exitcode, None)

join = TimingWrapper(p.join)

self.assertEqual(join(0), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), True)

self.assertEqual(join(-1), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), True)

timeout = support.SHORT_TIMEOUT

if not event.wait(timeout):

p.terminate()

p.join()

self.fail(f"event not signaled in {timeout} seconds")

meth(p)

if hasattr(signal, 'alarm'):

# On the Gentoo buildbot waitpid() often seems to block forever.

# We use alarm() to interrupt it if it blocks for too long.

def handler(*args):

raise RuntimeError('join took too long: %s' % p)

old_handler = signal.signal(signal.SIGALRM, handler)

try:

signal.alarm(10)

self.assertEqual(join(), None)

finally:

signal.alarm(0)

signal.signal(signal.SIGALRM, old_handler)

else:

self.assertEqual(join(), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), False)

self.assertNotIn(p, self.active_children())

p.join()

return p.exitcode

@unittest.skipIf(os.name == 'nt', "POSIX only")

def test_interrupt(self):

exitcode = self._kill_process(multiprocessing.Process.interrupt)

self.assertEqual(exitcode, 1)

# exit code 1 is hard-coded for uncaught exceptions

# (KeyboardInterrupt in this case)

# in multiprocessing.BaseProcess._bootstrap

@unittest.skipIf(os.name == 'nt', "POSIX only")

def test_interrupt_no_handler(self):

exitcode = self._kill_process(multiprocessing.Process.interrupt, target=self._sleep_no_int_handler)

self.assertEqual(exitcode, -signal.SIGINT)

def test_terminate(self):

exitcode = self._kill_process(multiprocessing.Process.terminate)

self.assertEqual(exitcode, -signal.SIGTERM)

def test_kill(self):

exitcode = self._kill_process(multiprocessing.Process.kill)

if os.name != 'nt':

self.assertEqual(exitcode, -signal.SIGKILL)

else:

self.assertEqual(exitcode, -signal.SIGTERM)

def test_cpu_count(self):

try:

cpus = multiprocessing.cpu_count()

except NotImplementedError:

cpus = 1

self.assertIsInstance(cpus, int)

self.assertGreaterEqual(cpus, 1)

def test_active_children(self):

self.assertEqual(type(self.active_children()), list)

event = self.Event()

p = self.Process(target=event.wait, args=())

self.assertNotIn(p, self.active_children())

try:

p.daemon = True

p.start()

self.assertIn(p, self.active_children())

finally:

event.set()

p.join()

self.assertNotIn(p, self.active_children())

@classmethod

def _test_recursion(cls, wconn, id):

wconn.send(id)

if len(id) < 2:

for i in range(2):

p = cls.Process(

target=cls._test_recursion, args=(wconn, id+[i])

)

p.start()

p.join()

def test_recursion(self):

rconn, wconn = self.Pipe(duplex=False)

self._test_recursion(wconn, [])

time.sleep(DELTA)

result = []

while rconn.poll():

result.append(rconn.recv())

expected = [

[],

[0],

[0, 0],

[0, 1],

[1],

[1, 0],

[1, 1]

]

self.assertEqual(result, expected)

@classmethod

def _test_sentinel(cls, event):

event.wait(10.0)

def test_sentinel(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

event = self.Event()

p = self.Process(target=self._test_sentinel, args=(event,))

with self.assertRaises(ValueError):

p.sentinel

p.start()

self.addCleanup(p.join)

sentinel = p.sentinel

self.assertIsInstance(sentinel, int)

self.assertFalse(wait_for_handle(sentinel, timeout=0.0))

event.set()

p.join()

self.assertTrue(wait_for_handle(sentinel, timeout=1))

@classmethod

def _test_close(cls, rc=0, q=None):

if q is not None:

q.get()

sys.exit(rc)

def test_close(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

q = self.Queue()

p = self.Process(target=self._test_close, kwargs={'q': q})

p.daemon = True

p.start()

self.assertEqual(p.is_alive(), True)

# Child is still alive, cannot close

with self.assertRaises(ValueError):

p.close()

q.put(None)

p.join()

self.assertEqual(p.is_alive(), False)

self.assertEqual(p.exitcode, 0)

p.close()

with self.assertRaises(ValueError):

p.is_alive()

with self.assertRaises(ValueError):

p.join()

with self.assertRaises(ValueError):

p.terminate()

p.close()

wr = weakref.ref(p)

del p

gc.collect()

self.assertIs(wr(), None)

close_queue(q)

@support.requires_resource('walltime')

def test_many_processes(self):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

N = 5 if sm == 'spawn' else 100

# Try to overwhelm the forkserver loop with events

procs = [self.Process(target=self._test_sleep, args=(0.01,))

for i in range(N)]

for p in procs:

p.start()

for p in procs:

join_process(p)

for p in procs:

self.assertEqual(p.exitcode, 0)

procs = [self.Process(target=self._sleep_some)

for i in range(N)]

for p in procs:

p.start()

time.sleep(0.001) # let the children start...

for p in procs:

p.terminate()

for p in procs:

join_process(p)

if os.name != 'nt':

exitcodes = [-signal.SIGTERM]

if sys.platform == 'darwin':

# bpo-31510: On macOS, killing a freshly started process with

# SIGTERM sometimes kills the process with SIGKILL.

exitcodes.append(-signal.SIGKILL)

for p in procs:

self.assertIn(p.exitcode, exitcodes)

def test_lose_target_ref(self):

c = DummyCallable()

wr = weakref.ref(c)

q = self.Queue()

p = self.Process(target=c, args=(q, c))

del c

p.start()

p.join()

gc.collect() # For PyPy or other GCs.

self.assertIs(wr(), None)

self.assertEqual(q.get(), 5)

close_queue(q)

@classmethod

def _test_child_fd_inflation(self, evt, q):

q.put(os_helper.fd_count())

evt.wait()

def test_child_fd_inflation(self):

# Number of fds in child processes should not grow with the

# number of running children.

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

if sm == 'fork':

# The fork method by design inherits all fds from the parent,

# trying to go against it is a lost battle

self.skipTest('test not appropriate for {}'.format(sm))

N = 5

evt = self.Event()

q = self.Queue()

procs = [self.Process(target=self._test_child_fd_inflation, args=(evt, q))

for i in range(N)]

for p in procs:

p.start()

try:

fd_counts = [q.get() for i in range(N)]

self.assertEqual(len(set(fd_counts)), 1, fd_counts)

finally:

evt.set()

for p in procs:

p.join()

close_queue(q)

@classmethod

def _test_wait_for_threads(self, evt):

def func1():

time.sleep(0.5)

evt.set()

def func2():

time.sleep(20)

evt.clear()

threading.Thread(target=func1).start()

threading.Thread(target=func2, daemon=True).start()

def test_wait_for_threads(self):

# A child process should wait for non-daemonic threads to end

# before exiting

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

evt = self.Event()

proc = self.Process(target=self._test_wait_for_threads, args=(evt,))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

@classmethod

def _test_error_on_stdio_flush(self, evt, break_std_streams={}):

for stream_name, action in break_std_streams.items():

if action == 'close':

stream = io.StringIO()

stream.close()

else:

assert action == 'remove'

stream = None

setattr(sys, stream_name, None)

evt.set()

def test_error_on_stdio_flush_1(self):

# Check that Process works with broken standard streams

streams = [io.StringIO(), None]

streams[0].close()

for stream_name in ('stdout', 'stderr'):

for stream in streams:

old_stream = getattr(sys, stream_name)

setattr(sys, stream_name, stream)

try:

evt = self.Event()

proc = self.Process(target=self._test_error_on_stdio_flush,

args=(evt,))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

self.assertEqual(proc.exitcode, 0)

finally:

setattr(sys, stream_name, old_stream)

def test_error_on_stdio_flush_2(self):

# Same as test_error_on_stdio_flush_1(), but standard streams are

# broken by the child process

for stream_name in ('stdout', 'stderr'):

for action in ('close', 'remove'):

old_stream = getattr(sys, stream_name)

try:

evt = self.Event()

proc = self.Process(target=self._test_error_on_stdio_flush,

args=(evt, {stream_name: action}))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

self.assertEqual(proc.exitcode, 0)

finally:

setattr(sys, stream_name, old_stream)

@staticmethod

def _sleep_and_set_event(evt, delay=0.0):

time.sleep(delay)

evt.set()

def check_forkserver_death(self, signum):

# bpo-31308: if the forkserver process has died, we should still

# be able to create and run new Process instances (the forkserver

# is implicitly restarted).

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

if sm != 'forkserver':

# The fork method by design inherits all fds from the parent,

# trying to go against it is a lost battle

self.skipTest('test not appropriate for {}'.format(sm))

from multiprocessing.forkserver import _forkserver

_forkserver.ensure_running()

# First process sleeps 500 ms

delay = 0.5

evt = self.Event()

proc = self.Process(target=self._sleep_and_set_event, args=(evt, delay))

proc.start()

pid = _forkserver._forkserver_pid

os.kill(pid, signum)

# give time to the fork server to die and time to proc to complete

time.sleep(delay * 2.0)

evt2 = self.Event()

proc2 = self.Process(target=self._sleep_and_set_event, args=(evt2,))

proc2.start()

proc2.join()

self.assertTrue(evt2.is_set())

self.assertEqual(proc2.exitcode, 0)

proc.join()

self.assertTrue(evt.is_set())

self.assertIn(proc.exitcode, (0, 255))

def test_forkserver_sigint(self):

# Catchable signal

self.check_forkserver_death(signal.SIGINT)

def test_forkserver_sigkill(self):

# Uncatchable signal

if os.name != 'nt':

self.check_forkserver_death(signal.SIGKILL)

def test_forkserver_auth_is_enabled(self):

if self.TYPE == "threads":

self.skipTest(f"test not appropriate for {self.TYPE}")

if multiprocessing.get_start_method() != "forkserver":

self.skipTest("forkserver start method specific")

forkserver = multiprocessing.forkserver._forkserver

forkserver.ensure_running()

self.assertTrue(forkserver._forkserver_pid)

authkey = forkserver._forkserver_authkey

self.assertTrue(authkey)

self.assertGreater(len(authkey), 15)

addr = forkserver._forkserver_address

self.assertTrue(addr)