_test_multiprocessing.py

#
# Unit tests for the multiprocessing package
#

importunittest
importunittest.mock
importqueueaspyqueue
importtextwrap
importtime
importio
importitertools
importsys
importos
importgc
importimportlib
importerrno
importfunctools
importsignal
importarray
importsocket
importrandom
importlogging
importshutil
importsubprocess
importstruct
importtempfile
importoperator
importpickle
importweakref
importwarnings
importtest.support
importtest.support.script_helper
fromtestimportsupport
fromtest.supportimporthashlib_helper
fromtest.supportimportimport_helper
fromtest.supportimportos_helper
fromtest.supportimportscript_helper
fromtest.supportimportsocket_helper
fromtest.supportimportthreading_helper
fromtest.supportimportwarnings_helper


# Skip tests if _multiprocessing wasn't built.
_multiprocessing=import_helper.import_module('_multiprocessing')
# Skip tests if sem_open implementation is broken.
support.skip_if_broken_multiprocessing_synchronize()
importthreading

importmultiprocessing.connection
importmultiprocessing.dummy
importmultiprocessing.heap
importmultiprocessing.managers
importmultiprocessing.pool
importmultiprocessing.queues
frommultiprocessing.connectionimportwait

frommultiprocessingimportutil

try:
frommultiprocessingimportreduction
HAS_REDUCTION=reduction.HAVE_SEND_HANDLE
exceptImportError:
HAS_REDUCTION=False

try:
frommultiprocessing.sharedctypesimportValue, copy
HAS_SHAREDCTYPES=True
exceptImportError:
HAS_SHAREDCTYPES=False

try:
frommultiprocessingimportshared_memory
HAS_SHMEM=True
exceptImportError:
HAS_SHMEM=False

try:
importmsvcrt
exceptImportError:
msvcrt=None


ifsupport.HAVE_ASAN_FORK_BUG:
# gh-89363: Skip multiprocessing tests if Python is built with ASAN to
# work around a libasan race condition: dead lock in pthread_create().
raiseunittest.SkipTest("libasan has a pthread_create() dead lock related to thread+fork")


# gh-110666: Tolerate a difference of 100 ms when comparing timings
# (clock resolution)
CLOCK_RES=0.100


deflatin(s):
returns.encode('latin')


defclose_queue(queue):
ifisinstance(queue, multiprocessing.queues.Queue):
queue.close()
queue.join_thread()


defjoin_process(process):
# Since multiprocessing.Process has the same API than threading.Thread
# (join() and is_alive(), the support function can be reused
threading_helper.join_thread(process)


ifos.name=="posix":
frommultiprocessingimportresource_tracker

def_resource_unlink(name, rtype):
resource_tracker._CLEANUP_FUNCS[rtype](name)


#
# Constants
#

LOG_LEVEL=util.SUBWARNING
#LOG_LEVEL = logging.DEBUG

DELTA=0.1
CHECK_TIMINGS=False# making true makes tests take a lot longer
# and can sometimes cause some non-serious
# failures because some calls block a bit
# longer than expected
ifCHECK_TIMINGS:
TIMEOUT1, TIMEOUT2, TIMEOUT3=0.82, 0.35, 1.4
else:
TIMEOUT1, TIMEOUT2, TIMEOUT3=0.1, 0.1, 0.1

# BaseManager.shutdown_timeout
SHUTDOWN_TIMEOUT=support.SHORT_TIMEOUT

WAIT_ACTIVE_CHILDREN_TIMEOUT=5.0

HAVE_GETVALUE=notgetattr(_multiprocessing,
'HAVE_BROKEN_SEM_GETVALUE', False)

WIN32= (sys.platform=="win32")

defwait_for_handle(handle, timeout):
iftimeoutisnotNoneandtimeout<0.0:
timeout=None
returnwait([handle], timeout)

try:
MAXFD=os.sysconf("SC_OPEN_MAX")
except:
MAXFD=256

# To speed up tests when using the forkserver, we can preload these:
PRELOAD= ['__main__', 'test.test_multiprocessing_forkserver']

#
# Some tests require ctypes
#

try:
fromctypesimportStructure, c_int, c_double, c_longlong
exceptImportError:
Structure=object
c_int=c_double=c_longlong=None


defcheck_enough_semaphores():
"""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
ifnsems==-1ornsems>=nsems_min:
return
raiseunittest.SkipTest("The OS doesn't support enough semaphores "
"to run the test (required: %d)."%nsems_min)


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

 This can be useful to save overall Python test suite execution time.
 "spawn" is the universal mode available on all platforms so this limits the
 decorated test to only execute within test_multiprocessing_spawn.

 This would not be necessary if we refactored our test suite to split things
 into other test files when they are not start method specific to be rerun
 under all start methods.
 """

defdecorator(test_item):

@functools.wraps(test_item)
defspawn_check_wrapper(*args, **kwargs):
if (start_method:=multiprocessing.get_start_method()) !="spawn":
raiseunittest.SkipTest(f"{start_method=}, not 'spawn'; {reason}")
returntest_item(*args, **kwargs)

returnspawn_check_wrapper

returndecorator


classTestInternalDecorators(unittest.TestCase):
"""Logic within a test suite that could errantly skip tests? Test it!"""

@unittest.skipIf(sys.platform=="win32", "test requires that fork exists.")
deftest_only_run_in_spawn_testsuite(self):
ifmultiprocessing.get_start_method() !="spawn":
raiseunittest.SkipTest("only run in test_multiprocessing_spawn.")

try:
@only_run_in_spawn_testsuite("testing this decorator")
defreturn_four_if_spawn():
return4
exceptExceptionaserr:
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)
withself.assertRaises(unittest.SkipTest) asctx:
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)


#
# Creates a wrapper for a function which records the time it takes to finish
#

classTimingWrapper(object):

def__init__(self, func):
self.func=func
self.elapsed=None

def__call__(self, *args, **kwds):
t=time.monotonic()
try:
returnself.func(*args, **kwds)
finally:
self.elapsed=time.monotonic() -t

#
# Base class for test cases
#

classBaseTestCase(object):

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()

defassertTimingAlmostEqual(self, a, b):
ifCHECK_TIMINGS:
self.assertAlmostEqual(a, b, 1)

defassertReturnsIfImplemented(self, value, func, *args):
try:
res=func(*args)
exceptNotImplementedError:
pass
else:
returnself.assertEqual(value, res)

# For the sanity of Windows users, rather than crashing or freezing in
# multiple ways.
def__reduce__(self, *args):
raiseNotImplementedError("shouldn't try to pickle a test case")

__reduce_ex__=__reduce__

#
# Return the value of a semaphore
#

defget_value(self):
try:
returnself.get_value()
exceptAttributeError:
try:
returnself._Semaphore__value
exceptAttributeError:
try:
returnself._value
exceptAttributeError:
raiseNotImplementedError

#
# Testcases
#

classDummyCallable:
def__call__(self, q, c):
assertisinstance(c, DummyCallable)
q.put(5)


class_TestProcess(BaseTestCase):

ALLOWED_TYPES= ('processes', 'threads')

deftest_current(self):
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))

current=self.current_process()
authkey=current.authkey

self.assertTrue(current.is_alive())
self.assertTrue(notcurrent.daemon)
self.assertIsInstance(authkey, bytes)
self.assertTrue(len(authkey) >0)
self.assertEqual(current.ident, os.getpid())
self.assertEqual(current.exitcode, None)

deftest_set_executable(self):
ifself.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
 ]
forpathinpaths:
self.set_executable(path)
p=self.Process()
p.start()
p.join()
self.assertEqual(p.exitcode, 0)

@support.requires_resource('cpu')
deftest_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)

forargs, args_typeintest_cases:
withself.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)

deftest_daemon_argument(self):
ifself.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)
ifcls.TYPE!='threads':
q.put(bytes(current.authkey))
q.put(current.pid)

deftest_parent_process_attributes(self):
ifself.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):
frommultiprocessing.processimportparent_process
wconn.send([parent_process().pid, parent_process().name])

deftest_parent_process(self):
ifself.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()

ifnotrconn.poll(timeout=support.LONG_TIMEOUT):
raiseAssertionError("Could not communicate with child process")
parent_process_status=rconn.recv()
self.assertEqual(parent_process_status, "alive")

p.terminate()
p.join()

ifnotrconn.poll(timeout=support.LONG_TIMEOUT):
raiseAssertionError("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):
frommultiprocessing.processimportparent_process
wconn.send("alive"ifparent_process().is_alive() else"not alive")
parent_process().join(timeout=support.SHORT_TIMEOUT)
wconn.send("alive"ifparent_process().is_alive() else"not alive")

deftest_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()

ifself.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.assertTrue(type(self.active_children()) islist)
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)
ifself.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")
deftest_process_mainthread_native_id(self):
ifself.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_test_sleep(cls, delay):
time.sleep(delay)

def_kill_process(self, meth):
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))

p=self.Process(target=self._sleep_some)
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)

# XXX maybe terminating too soon causes the problems on Gentoo...
time.sleep(1)

meth(p)

ifhasattr(signal, 'alarm'):
# On the Gentoo buildbot waitpid() often seems to block forever.
# We use alarm() to interrupt it if it blocks for too long.
defhandler(*args):
raiseRuntimeError('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()

returnp.exitcode

deftest_terminate(self):
exitcode=self._kill_process(multiprocessing.Process.terminate)
self.assertEqual(exitcode, -signal.SIGTERM)

deftest_kill(self):
exitcode=self._kill_process(multiprocessing.Process.kill)
ifos.name!='nt':
self.assertEqual(exitcode, -signal.SIGKILL)
else:
self.assertEqual(exitcode, -signal.SIGTERM)

deftest_cpu_count(self):
try:
cpus=multiprocessing.cpu_count()
exceptNotImplementedError:
cpus=1
self.assertTrue(type(cpus) isint)
self.assertTrue(cpus>=1)

deftest_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)
iflen(id) <2:
foriinrange(2):
p=cls.Process(
target=cls._test_recursion, args=(wconn, id+[i])
 )
p.start()
p.join()

deftest_recursion(self):
rconn, wconn=self.Pipe(duplex=False)
self._test_recursion(wconn, [])

time.sleep(DELTA)
result= []
whilerconn.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)

deftest_sentinel(self):
ifself.TYPE=="threads":
self.skipTest('test not appropriate for {}'.format(self.TYPE))
event=self.Event()
p=self.Process(target=self._test_sentinel, args=(event,))
withself.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):
ifqisnotNone:
q.get()
sys.exit(rc)

deftest_close(self):
ifself.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
withself.assertRaises(ValueError):
p.close()

q.put(None)
p.join()
self.assertEqual(p.is_alive(), False)
self.assertEqual(p.exitcode, 0)
p.close()
withself.assertRaises(ValueError):
p.is_alive()
withself.assertRaises(ValueError):
p.join()
withself.assertRaises(ValueError):
p.terminate()
p.close()

wr=weakref.ref(p)
delp
gc.collect()
self.assertIs(wr(), None)

close_queue(q)

@support.requires_resource('walltime')
deftest_many_processes(self):
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm=multiprocessing.get_start_method()
N=5ifsm=='spawn'else100

# Try to overwhelm the forkserver loop with events
procs= [self.Process(target=self._test_sleep, args=(0.01,))
foriinrange(N)]
forpinprocs:
p.start()
forpinprocs:
join_process(p)
forpinprocs:
self.assertEqual(p.exitcode, 0)

procs= [self.Process(target=self._sleep_some)
foriinrange(N)]
forpinprocs:
p.start()
time.sleep(0.001) # let the children start...
forpinprocs:
p.terminate()
forpinprocs:
join_process(p)
ifos.name!='nt':
exitcodes= [-signal.SIGTERM]
ifsys.platform=='darwin':
# bpo-31510: On macOS, killing a freshly started process with
# SIGTERM sometimes kills the process with SIGKILL.
exitcodes.append(-signal.SIGKILL)
forpinprocs:
self.assertIn(p.exitcode, exitcodes)

deftest_lose_target_ref(self):
c=DummyCallable()
wr=weakref.ref(c)
q=self.Queue()
p=self.Process(target=c, args=(q, c))
delc
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()

deftest_child_fd_inflation(self):
# Number of fds in child processes should not grow with the
# number of running children.
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm=multiprocessing.get_start_method()
ifsm=='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))
foriinrange(N)]
forpinprocs:
p.start()

try:
fd_counts= [q.get() foriinrange(N)]
self.assertEqual(len(set(fd_counts)), 1, fd_counts)

finally:
evt.set()
forpinprocs:
p.join()
close_queue(q)

@classmethod
def_test_wait_for_threads(self, evt):
deffunc1():
time.sleep(0.5)
evt.set()

deffunc2():
time.sleep(20)
evt.clear()

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

deftest_wait_for_threads(self):
# A child process should wait for non-daemonic threads to end
# before exiting
ifself.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={}):
forstream_name, actioninbreak_std_streams.items():
ifaction=='close':
stream=io.StringIO()
stream.close()
else:
assertaction=='remove'
stream=None
setattr(sys, stream_name, None)
evt.set()

deftest_error_on_stdio_flush_1(self):
# Check that Process works with broken standard streams
streams= [io.StringIO(), None]
streams[0].close()
forstream_namein ('stdout', 'stderr'):
forstreaminstreams:
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)

deftest_error_on_stdio_flush_2(self):
# Same as test_error_on_stdio_flush_1(), but standard streams are
# broken by the child process
forstream_namein ('stdout', 'stderr'):
foractionin ('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)

@classmethod
def_sleep_and_set_event(self, evt, delay=0.0):
time.sleep(delay)
evt.set()

defcheck_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).
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))
sm=multiprocessing.get_start_method()
ifsm!='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))

frommultiprocessing.forkserverimport_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))

deftest_forkserver_sigint(self):
# Catchable signal
self.check_forkserver_death(signal.SIGINT)

deftest_forkserver_sigkill(self):
# Uncatchable signal
ifos.name!='nt':
self.check_forkserver_death(signal.SIGKILL)


#
#
#

class_UpperCaser(multiprocessing.Process):

def__init__(self):
multiprocessing.Process.__init__(self)
self.child_conn, self.parent_conn=multiprocessing.Pipe()

defrun(self):
self.parent_conn.close()
forsiniter(self.child_conn.recv, None):
self.child_conn.send(s.upper())
self.child_conn.close()

defsubmit(self, s):
asserttype(s) isstr
self.parent_conn.send(s)
returnself.parent_conn.recv()

defstop(self):
self.parent_conn.send(None)
self.parent_conn.close()
self.child_conn.close()

class_TestSubclassingProcess(BaseTestCase):

ALLOWED_TYPES= ('processes',)

deftest_subclassing(self):
uppercaser=_UpperCaser()
uppercaser.daemon=True
uppercaser.start()
self.assertEqual(uppercaser.submit('hello'), 'HELLO')
self.assertEqual(uppercaser.submit('world'), 'WORLD')
uppercaser.stop()
uppercaser.join()

deftest_stderr_flush(self):
# sys.stderr is flushed at process shutdown (issue #13812)
ifself.TYPE=="threads":
self.skipTest('test not appropriate for {}'.format(self.TYPE))

testfn=os_helper.TESTFN
self.addCleanup(os_helper.unlink, testfn)
proc=self.Process(target=self._test_stderr_flush, args=(testfn,))
proc.start()
proc.join()
withopen(testfn, encoding="utf-8") asf:
err=f.read()
# The whole traceback was printed
self.assertIn("ZeroDivisionError", err)
self.assertIn("test_multiprocessing.py", err)
self.assertIn("1/0 # MARKER", err)

@classmethod
def_test_stderr_flush(cls, testfn):
fd=os.open(testfn, os.O_WRONLY|os.O_CREAT|os.O_EXCL)
sys.stderr=open(fd, 'w', encoding="utf-8", closefd=False)
1/0# MARKER


@classmethod
def_test_sys_exit(cls, reason, testfn):
fd=os.open(testfn, os.O_WRONLY|os.O_CREAT|os.O_EXCL)
sys.stderr=open(fd, 'w', encoding="utf-8", closefd=False)
sys.exit(reason)

deftest_sys_exit(self):
# See Issue 13854
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))

testfn=os_helper.TESTFN
self.addCleanup(os_helper.unlink, testfn)

forreasonin (
 [1, 2, 3],
'ignore this',
 ):
p=self.Process(target=self._test_sys_exit, args=(reason, testfn))
p.daemon=True
p.start()
join_process(p)
self.assertEqual(p.exitcode, 1)

withopen(testfn, encoding="utf-8") asf:
content=f.read()
self.assertEqual(content.rstrip(), str(reason))

os.unlink(testfn)

cases= [
 ((True,), 1),
 ((False,), 0),