"""

Tests for the threading module.

"""

import test.support

from test.support import threading_helper

from test.support import verbose, cpython_only, os_helper

from test.support.import_helper import import_module

from test.support.script_helper import assert_python_ok, assert_python_failure

import random

import sys

import _thread

import threading

import time

import unittest

import weakref

import os

import subprocess

import signal

import textwrap

import traceback

from unittest import mock

from test import lock_tests

from test import support

# Between fork() and exec(), only async-safe functions are allowed (issues

# #12316 and #11870), and fork() from a worker thread is known to trigger

# problems with some operating systems (issue #3863): skip problematic tests

# on platforms known to behave badly.

platforms_to_skip = ('netbsd5', 'hp-ux11')

# Is Python built with Py_DEBUG macro defined?

Py_DEBUG = hasattr(sys, 'gettotalrefcount')

def restore_default_excepthook(testcase):

testcase.addCleanup(setattr, threading, 'excepthook', threading.excepthook)

threading.excepthook = threading.__excepthook__

# A trivial mutable counter.

class Counter(object):

def __init__(self):

self.value = 0

def inc(self):

self.value += 1

def dec(self):

self.value -= 1

def get(self):

return self.value

class TestThread(threading.Thread):

def __init__(self, name, testcase, sema, mutex, nrunning):

threading.Thread.__init__(self, name=name)

self.testcase = testcase

self.sema = sema

self.mutex = mutex

self.nrunning = nrunning

def run(self):

delay = random.random() / 10000.0

if verbose:

print('task %s will run for %.1f usec' %

(self.name, delay * 1e6))

with self.sema:

with self.mutex:

self.nrunning.inc()

if verbose:

print(self.nrunning.get(), 'tasks are running')

self.testcase.assertLessEqual(self.nrunning.get(), 3)

time.sleep(delay)

if verbose:

print('task', self.name, 'done')

with self.mutex:

self.nrunning.dec()

self.testcase.assertGreaterEqual(self.nrunning.get(), 0)

if verbose:

print('%s is finished. %d tasks are running' %

(self.name, self.nrunning.get()))

class BaseTestCase(unittest.TestCase):

def setUp(self):

self._threads = threading_helper.threading_setup()

def tearDown(self):

threading_helper.threading_cleanup(*self._threads)

test.support.reap_children()

class ThreadTests(BaseTestCase):

@cpython_only

def test_name(self):

def func(): pass

thread = threading.Thread(name="myname1")

self.assertEqual(thread.name, "myname1")

# Convert int name to str

thread = threading.Thread(name=123)

self.assertEqual(thread.name, "123")

# target name is ignored if name is specified

thread = threading.Thread(target=func, name="myname2")

self.assertEqual(thread.name, "myname2")

with mock.patch.object(threading, '_counter', return_value=2):

thread = threading.Thread(name="")

self.assertEqual(thread.name, "Thread-2")

with mock.patch.object(threading, '_counter', return_value=3):

thread = threading.Thread()

self.assertEqual(thread.name, "Thread-3")

with mock.patch.object(threading, '_counter', return_value=5):

thread = threading.Thread(target=func)

self.assertEqual(thread.name, "Thread-5 (func)")

@cpython_only

def test_disallow_instantiation(self):

# Ensure that the type disallows instantiation (bpo-43916)

lock = threading.Lock()

test.support.check_disallow_instantiation(self, type(lock))

# Create a bunch of threads, let each do some work, wait until all are

# done.

def test_various_ops(self):

# This takes about n/3 seconds to run (about n/3 clumps of tasks,

# times about 1 second per clump).

NUMTASKS = 10

# no more than 3 of the 10 can run at once

sema = threading.BoundedSemaphore(value=3)

mutex = threading.RLock()

numrunning = Counter()

threads = []

for i in range(NUMTASKS):

t = TestThread("<thread %d>"%i, self, sema, mutex, numrunning)

threads.append(t)

self.assertIsNone(t.ident)

self.assertRegex(repr(t), r'^<TestThread\(.*, initial\)>$')

t.start()

if hasattr(threading, 'get_native_id'):

native_ids = set(t.native_id for t in threads) | {threading.get_native_id()}

self.assertNotIn(None, native_ids)

self.assertEqual(len(native_ids), NUMTASKS + 1)

if verbose:

print('waiting for all tasks to complete')

for t in threads:

t.join()

self.assertFalse(t.is_alive())

self.assertNotEqual(t.ident, 0)

self.assertIsNotNone(t.ident)

self.assertRegex(repr(t), r'^<TestThread\(.*, stopped -?\d+\)>$')

if verbose:

print('all tasks done')

self.assertEqual(numrunning.get(), 0)

def test_ident_of_no_threading_threads(self):

# The ident still must work for the main thread and dummy threads.

self.assertIsNotNone(threading.current_thread().ident)

def f():

ident.append(threading.current_thread().ident)

done.set()

done = threading.Event()

ident = []

with threading_helper.wait_threads_exit():

tid = _thread.start_new_thread(f, ())

done.wait()

self.assertEqual(ident[0], tid)

# Kill the "immortal" _DummyThread

del threading._active[ident[0]]

# run with a small(ish) thread stack size (256 KiB)

def test_various_ops_small_stack(self):

if verbose:

print('with 256 KiB thread stack size...')

try:

threading.stack_size(262144)

except _thread.error:

raise unittest.SkipTest(

'platform does not support changing thread stack size')

self.test_various_ops()

threading.stack_size(0)

# run with a large thread stack size (1 MiB)

def test_various_ops_large_stack(self):

if verbose:

print('with 1 MiB thread stack size...')

try:

threading.stack_size(0x100000)

except _thread.error:

raise unittest.SkipTest(

'platform does not support changing thread stack size')

self.test_various_ops()

threading.stack_size(0)

def test_foreign_thread(self):

# Check that a "foreign" thread can use the threading module.

def f(mutex):

# Calling current_thread() forces an entry for the foreign

# thread to get made in the threading._active map.

threading.current_thread()

mutex.release()

mutex = threading.Lock()

mutex.acquire()

with threading_helper.wait_threads_exit():

tid = _thread.start_new_thread(f, (mutex,))

# Wait for the thread to finish.

mutex.acquire()

self.assertIn(tid, threading._active)

self.assertIsInstance(threading._active[tid], threading._DummyThread)

#Issue 29376

self.assertTrue(threading._active[tid].is_alive())

self.assertRegex(repr(threading._active[tid]), '_DummyThread')

del threading._active[tid]

# PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)

# exposed at the Python level. This test relies on ctypes to get at it.

def test_PyThreadState_SetAsyncExc(self):

ctypes = import_module("ctypes")

set_async_exc = ctypes.pythonapi.PyThreadState_SetAsyncExc

set_async_exc.argtypes = (ctypes.c_ulong, ctypes.py_object)

class AsyncExc(Exception):

pass

exception = ctypes.py_object(AsyncExc)

# First check it works when setting the exception from the same thread.

tid = threading.get_ident()

self.assertIsInstance(tid, int)

self.assertGreater(tid, 0)

try:

result = set_async_exc(tid, exception)

# The exception is async, so we might have to keep the VM busy until

# it notices.

while True:

pass

except AsyncExc:

pass

else:

# This code is unreachable but it reflects the intent. If we wanted

# to be smarter the above loop wouldn't be infinite.

self.fail("AsyncExc not raised")

try:

self.assertEqual(result, 1) # one thread state modified

except UnboundLocalError:

# The exception was raised too quickly for us to get the result.

pass

# `worker_started` is set by the thread when it's inside a try/except

# block waiting to catch the asynchronously set AsyncExc exception.

# `worker_saw_exception` is set by the thread upon catching that

# exception.

worker_started = threading.Event()

worker_saw_exception = threading.Event()

class Worker(threading.Thread):

def run(self):

self.id = threading.get_ident()

self.finished = False

try:

while True:

worker_started.set()

time.sleep(0.1)

except AsyncExc:

self.finished = True

worker_saw_exception.set()

t = Worker()

t.daemon = True # so if this fails, we don't hang Python at shutdown

t.start()

if verbose:

print(" started worker thread")

# Try a thread id that doesn't make sense.

if verbose:

print(" trying nonsensical thread id")

result = set_async_exc(-1, exception)

self.assertEqual(result, 0) # no thread states modified

# Now raise an exception in the worker thread.

if verbose:

print(" waiting for worker thread to get started")

ret = worker_started.wait()

self.assertTrue(ret)

if verbose:

print(" verifying worker hasn't exited")

self.assertFalse(t.finished)

if verbose:

print(" attempting to raise asynch exception in worker")

result = set_async_exc(t.id, exception)

self.assertEqual(result, 1) # one thread state modified

if verbose:

print(" waiting for worker to say it caught the exception")

worker_saw_exception.wait(timeout=support.SHORT_TIMEOUT)

self.assertTrue(t.finished)

if verbose:

print(" all OK -- joining worker")

if t.finished:

t.join()

# else the thread is still running, and we have no way to kill it

def test_limbo_cleanup(self):

# Issue 7481: Failure to start thread should cleanup the limbo map.

def fail_new_thread(*args):

raise threading.ThreadError()

_start_new_thread = threading._start_new_thread

threading._start_new_thread = fail_new_thread

try:

t = threading.Thread(target=lambda: None)

self.assertRaises(threading.ThreadError, t.start)

self.assertFalse(

t in threading._limbo,

"Failed to cleanup _limbo map on failure of Thread.start().")

finally:

threading._start_new_thread = _start_new_thread

def test_finalize_running_thread(self):

# Issue 1402: the PyGILState_Ensure / _Release functions may be called

# very late on python exit: on deallocation of a running thread for

# example.

import_module("ctypes")

rc, out, err = assert_python_failure("-c", """if 1:

import ctypes, sys, time, _thread

# This lock is used as a simple event variable.

ready = _thread.allocate_lock()

ready.acquire()

# Module globals are cleared before __del__ is run

# So we save the functions in class dict

class C:

ensure = ctypes.pythonapi.PyGILState_Ensure

release = ctypes.pythonapi.PyGILState_Release

def __del__(self):

state = self.ensure()

self.release(state)

def waitingThread():

x = C()

ready.release()

time.sleep(100)

_thread.start_new_thread(waitingThread, ())

ready.acquire() # Be sure the other thread is waiting.

sys.exit(42)

""")

self.assertEqual(rc, 42)

def test_finalize_with_trace(self):

# Issue1733757

# Avoid a deadlock when sys.settrace steps into threading._shutdown

assert_python_ok("-c", """if 1:

import sys, threading

# A deadlock-killer, to prevent the

# testsuite to hang forever

def killer():

import os, time

time.sleep(2)

print('program blocked; aborting')

os._exit(2)

t = threading.Thread(target=killer)

t.daemon = True

t.start()

# This is the trace function

def func(frame, event, arg):

threading.current_thread()

return func

sys.settrace(func)

""")

def test_join_nondaemon_on_shutdown(self):

# Issue 1722344

# Raising SystemExit skipped threading._shutdown

rc, out, err = assert_python_ok("-c", """if 1:

import threading

from time import sleep

def child():

sleep(1)

# As a non-daemon thread we SHOULD wake up and nothing

# should be torn down yet

print("Woke up, sleep function is:", sleep)

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

raise SystemExit

""")

self.assertEqual(out.strip(),

b"Woke up, sleep function is: <built-in function sleep>")

self.assertEqual(err, b"")

def test_enumerate_after_join(self):

# Try hard to trigger #1703448: a thread is still returned in

# threading.enumerate() after it has been join()ed.

enum = threading.enumerate

old_interval = sys.getswitchinterval()

try:

for i in range(1, 100):

sys.setswitchinterval(i * 0.0002)

t = threading.Thread(target=lambda: None)

t.start()

t.join()

l = enum()

self.assertNotIn(t, l,

"#1703448 triggered after %d trials: %s" % (i, l))

finally:

sys.setswitchinterval(old_interval)

def test_no_refcycle_through_target(self):

class RunSelfFunction(object):

def __init__(self, should_raise):

# The links in this refcycle from Thread back to self

# should be cleaned up when the thread completes.

self.should_raise = should_raise

self.thread = threading.Thread(target=self._run,

args=(self,),

kwargs={'yet_another':self})

self.thread.start()

def _run(self, other_ref, yet_another):

if self.should_raise:

raise SystemExit

restore_default_excepthook(self)

cyclic_object = RunSelfFunction(should_raise=False)

weak_cyclic_object = weakref.ref(cyclic_object)

cyclic_object.thread.join()

del cyclic_object

self.assertIsNone(weak_cyclic_object(),

msg=('%d references still around' %

sys.getrefcount(weak_cyclic_object())))

raising_cyclic_object = RunSelfFunction(should_raise=True)

weak_raising_cyclic_object = weakref.ref(raising_cyclic_object)

raising_cyclic_object.thread.join()

del raising_cyclic_object

self.assertIsNone(weak_raising_cyclic_object(),

msg=('%d references still around' %

sys.getrefcount(weak_raising_cyclic_object())))

def test_old_threading_api(self):

# Just a quick sanity check to make sure the old method names are

# still present

t = threading.Thread()

with self.assertWarnsRegex(DeprecationWarning,

r'get the daemon attribute'):

t.isDaemon()

with self.assertWarnsRegex(DeprecationWarning,

r'set the daemon attribute'):

t.setDaemon(True)

with self.assertWarnsRegex(DeprecationWarning,

r'get the name attribute'):

t.getName()

with self.assertWarnsRegex(DeprecationWarning,

r'set the name attribute'):

t.setName("name")

e = threading.Event()

with self.assertWarnsRegex(DeprecationWarning, 'use is_set()'):

e.isSet()

cond = threading.Condition()

cond.acquire()

with self.assertWarnsRegex(DeprecationWarning, 'use notify_all()'):

cond.notifyAll()

with self.assertWarnsRegex(DeprecationWarning, 'use active_count()'):

threading.activeCount()

with self.assertWarnsRegex(DeprecationWarning, 'use current_thread()'):

threading.currentThread()

def test_repr_daemon(self):

t = threading.Thread()

self.assertNotIn('daemon', repr(t))

t.daemon = True

self.assertIn('daemon', repr(t))

def test_daemon_param(self):

t = threading.Thread()

self.assertFalse(t.daemon)

t = threading.Thread(daemon=False)

self.assertFalse(t.daemon)

t = threading.Thread(daemon=True)

self.assertTrue(t.daemon)

@unittest.skipUnless(hasattr(os, 'fork'), 'needs os.fork()')

def test_fork_at_exit(self):

# bpo-42350: Calling os.fork() after threading._shutdown() must

# not log an error.

code = textwrap.dedent("""

import atexit

import os

import sys

from test.support import wait_process

# Import the threading module to register its "at fork" callback

import threading

def exit_handler():

pid = os.fork()

if not pid:

print("child process ok", file=sys.stderr, flush=True)

# child process

else:

wait_process(pid, exitcode=0)

# exit_handler() will be called after threading._shutdown()

atexit.register(exit_handler)

""")

_, out, err = assert_python_ok("-c", code)

self.assertEqual(out, b'')

self.assertEqual(err.rstrip(), b'child process ok')

@unittest.skipUnless(hasattr(os, 'fork'), 'test needs fork()')

def test_dummy_thread_after_fork(self):

# Issue #14308: a dummy thread in the active list doesn't mess up

# the after-fork mechanism.

code = """if 1:

import _thread, threading, os, time

def background_thread(evt):

# Creates and registers the _DummyThread instance

threading.current_thread()

evt.set()

time.sleep(10)

evt = threading.Event()

_thread.start_new_thread(background_thread, (evt,))

evt.wait()

assert threading.active_count() == 2, threading.active_count()

if os.fork() == 0:

assert threading.active_count() == 1, threading.active_count()

os._exit(0)

else:

os.wait()

"""

_, out, err = assert_python_ok("-c", code)

self.assertEqual(out, b'')

self.assertEqual(err, b'')

@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")

def test_is_alive_after_fork(self):

# Try hard to trigger #18418: is_alive() could sometimes be True on

# threads that vanished after a fork.

old_interval = sys.getswitchinterval()

self.addCleanup(sys.setswitchinterval, old_interval)

# Make the bug more likely to manifest.

test.support.setswitchinterval(1e-6)

for i in range(20):

t = threading.Thread(target=lambda: None)

t.start()

pid = os.fork()

if pid == 0:

os._exit(11 if t.is_alive() else 10)

else:

t.join()

support.wait_process(pid, exitcode=10)

def test_main_thread(self):

main = threading.main_thread()

self.assertEqual(main.name, 'MainThread')

self.assertEqual(main.ident, threading.current_thread().ident)

self.assertEqual(main.ident, threading.get_ident())

def f():

self.assertNotEqual(threading.main_thread().ident,

threading.current_thread().ident)

th = threading.Thread(target=f)

th.start()

th.join()

@unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()")

@unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()")

def test_main_thread_after_fork(self):

code = """if 1:

import os, threading

from test import support

pid = os.fork()

if pid == 0:

main = threading.main_thread()

print(main.name)

print(main.ident == threading.current_thread().ident)

print(main.ident == threading.get_ident())

else:

support.wait_process(pid, exitcode=0)

"""

_, out, err = assert_python_ok("-c", code)

data = out.decode().replace('\r', '')

self.assertEqual(err, b"")

self.assertEqual(data, "MainThread\nTrue\nTrue\n")

@unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")

@unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()")

@unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()")

def test_main_thread_after_fork_from_nonmain_thread(self):

code = """if 1:

import os, threading, sys

from test import support

def func():

pid = os.fork()

if pid == 0:

main = threading.main_thread()

print(main.name)

print(main.ident == threading.current_thread().ident)

print(main.ident == threading.get_ident())

# stdout is fully buffered because not a tty,

# we have to flush before exit.

sys.stdout.flush()

else:

support.wait_process(pid, exitcode=0)

th = threading.Thread(target=func)

th.start()

th.join()

"""

_, out, err = assert_python_ok("-c", code)

data = out.decode().replace('\r', '')

self.assertEqual(err, b"")

self.assertEqual(data, "Thread-1 (func)\nTrue\nTrue\n")

def test_main_thread_during_shutdown(self):

# bpo-31516: current_thread() should still point to the main thread

# at shutdown

code = """if 1:

import gc, threading

main_thread = threading.current_thread()

assert main_thread is threading.main_thread() # sanity check

class RefCycle:

def __init__(self):

self.cycle = self

def __del__(self):

print("GC:",

threading.current_thread() is main_thread,

threading.main_thread() is main_thread,

threading.enumerate() == [main_thread])

RefCycle()

gc.collect() # sanity check

x = RefCycle()

"""

_, out, err = assert_python_ok("-c", code)

data = out.decode()

self.assertEqual(err, b"")

self.assertEqual(data.splitlines(),

["GC: True True True"] * 2)

def test_finalization_shutdown(self):

# bpo-36402: Py_Finalize() calls threading._shutdown() which must wait

# until Python thread states of all non-daemon threads get deleted.

#

# Test similar to SubinterpThreadingTests.test_threads_join_2(), but

# test the finalization of the main interpreter.

code = """if 1:

import os

import threading

import time

import random

def random_sleep():

seconds = random.random() * 0.010

time.sleep(seconds)

class Sleeper:

def __del__(self):

random_sleep()

tls = threading.local()

def f():

# Sleep a bit so that the thread is still running when

# Py_Finalize() is called.

random_sleep()

tls.x = Sleeper()

random_sleep()

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

random_sleep()

"""

rc, out, err = assert_python_ok("-c", code)

self.assertEqual(err, b"")

def test_tstate_lock(self):

# Test an implementation detail of Thread objects.

started = _thread.allocate_lock()

finish = _thread.allocate_lock()

started.acquire()

finish.acquire()

def f():

started.release()

finish.acquire()

time.sleep(0.01)

# The tstate lock is None until the thread is started

t = threading.Thread(target=f)

self.assertIs(t._tstate_lock, None)

t.start()

started.acquire()

self.assertTrue(t.is_alive())

# The tstate lock can't be acquired when the thread is running

# (or suspended).

tstate_lock = t._tstate_lock

self.assertFalse(tstate_lock.acquire(timeout=0), False)

finish.release()

# When the thread ends, the state_lock can be successfully

# acquired.

self.assertTrue(tstate_lock.acquire(timeout=support.SHORT_TIMEOUT), False)

# But is_alive() is still True: we hold _tstate_lock now, which

# prevents is_alive() from knowing the thread's end-of-life C code

# is done.

self.assertTrue(t.is_alive())

# Let is_alive() find out the C code is done.

tstate_lock.release()

self.assertFalse(t.is_alive())

# And verify the thread disposed of _tstate_lock.

self.assertIsNone(t._tstate_lock)

t.join()

def test_repr_stopped(self):

# Verify that "stopped" shows up in repr(Thread) appropriately.

started = _thread.allocate_lock()

finish = _thread.allocate_lock()

started.acquire()

finish.acquire()

def f():

started.release()

finish.acquire()

t = threading.Thread(target=f)

t.start()

started.acquire()

self.assertIn("started", repr(t))

finish.release()

# "stopped" should appear in the repr in a reasonable amount of time.

# Implementation detail: as of this writing, that's trivially true

# if .join() is called, and almost trivially true if .is_alive() is

# called. The detail we're testing here is that "stopped" shows up

# "all on its own".

LOOKING_FOR = "stopped"

for i in range(500):

if LOOKING_FOR in repr(t):

break

time.sleep(0.01)

self.assertIn(LOOKING_FOR, repr(t)) # we waited at least 5 seconds

t.join()

def test_BoundedSemaphore_limit(self):

# BoundedSemaphore should raise ValueError if released too often.

for limit in range(1, 10):

bs = threading.BoundedSemaphore(limit)

threads = [threading.Thread(target=bs.acquire)

for _ in range(limit)]

for t in threads:

t.start()

for t in threads:

t.join()

threads = [threading.Thread(target=bs.release)

for _ in range(limit)]

for t in threads:

t.start()

for t in threads:

t.join()

self.assertRaises(ValueError, bs.release)

@cpython_only

def test_frame_tstate_tracing(self):

# Issue #14432: Crash when a generator is created in a C thread that is

# destroyed while the generator is still used. The issue was that a

# generator contains a frame, and the frame kept a reference to the

# Python state of the destroyed C thread. The crash occurs when a trace

# function is setup.

def noop_trace(frame, event, arg):

# no operation

return noop_trace

def generator():

while 1:

yield "generator"

def callback():

if callback.gen is None:

callback.gen = generator()

return next(callback.gen)

callback.gen = None

old_trace = sys.gettrace()

sys.settrace(noop_trace)

try:

# Install a trace function

threading.settrace(noop_trace)

# Create a generator in a C thread which exits after the call

import _testcapi

_testcapi.call_in_temporary_c_thread(callback)

# Call the generator in a different Python thread, check that the

# generator didn't keep a reference to the destroyed thread state

for test in range(3):

# The trace function is still called here

callback()

finally:

sys.settrace(old_trace)

def test_gettrace(self):

def noop_trace(frame, event, arg):

# no operation

return noop_trace

old_trace = threading.gettrace()

try:

threading.settrace(noop_trace)

trace_func = threading.gettrace()

self.assertEqual(noop_trace,trace_func)

finally:

threading.settrace(old_trace)

def test_getprofile(self):

def fn(*args): pass

old_profile = threading.getprofile()

try:

threading.setprofile(fn)

self.assertEqual(fn, threading.getprofile())

finally:

threading.setprofile(old_profile)

@cpython_only

def test_shutdown_locks(self):

for daemon in (False, True):

with self.subTest(daemon=daemon):

event = threading.Event()

thread = threading.Thread(target=event.wait, daemon=daemon)

# Thread.start() must add lock to _shutdown_locks,

# but only for non-daemon thread

thread.start()

tstate_lock = thread._tstate_lock

if not daemon:

self.assertIn(tstate_lock, threading._shutdown_locks)

else:

self.assertNotIn(tstate_lock, threading._shutdown_locks)

# unblock the thread and join it

event.set()

thread.join()

# Thread._stop() must remove tstate_lock from _shutdown_locks.

# Daemon threads must never add it to _shutdown_locks.

self.assertNotIn(tstate_lock, threading._shutdown_locks)

def test_locals_at_exit(self):

# bpo-19466: thread locals must not be deleted before destructors

# are called

rc, out, err = assert_python_ok("-c", """if 1:

import threading

class Atexit:

def __del__(self):

print("thread_dict.atexit = %r" % thread_dict.atexit)

thread_dict = threading.local()

thread_dict.atexit = "value"

atexit = Atexit()

""")

self.assertEqual(out.rstrip(), b"thread_dict.atexit = 'value'")

def test_boolean_target(self):

# bpo-41149: A thread that had a boolean value of False would not

# run, regardless of whether it was callable. The correct behaviour

# is for a thread to do nothing if its target is None, and to call

# the target otherwise.

class BooleanTarget(object):

def __init__(self):

self.ran = False

def __bool__(self):

return False

def __call__(self):

self.ran = True

target = BooleanTarget()

thread = threading.Thread(target=target)

thread.start()

thread.join()

self.assertTrue(target.ran)

def test_leak_without_join(self):

# bpo-37788: Test that a thread which is not joined explicitly

# does not leak. Test written for reference leak checks.

def noop(): pass

with threading_helper.wait_threads_exit():

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

# Thread.join() is not called

@unittest.skipUnless(Py_DEBUG, 'need debug build (Py_DEBUG)')

def test_debug_deprecation(self):

# bpo-44584: The PYTHONTHREADDEBUG environment variable is deprecated

rc, out, err = assert_python_ok("-Wdefault", "-c", "pass",

PYTHONTHREADDEBUG="1")

msg = (b'DeprecationWarning: The threading debug '

b'(PYTHONTHREADDEBUG environment variable) '

b'is deprecated and will be removed in Python 3.12')

self.assertIn(msg, err)

class ThreadJoinOnShutdown(BaseTestCase):

def _run_and_join(self, script):

script = """if 1:

import sys, os, time, threading

# a thread, which waits for the main program to terminate

def joiningfunc(mainthread):

mainthread.join()

print('end of thread')

# stdout is fully buffered because not a tty, we have to flush

# before exit.

sys.stdout.flush()

\n""" + script

rc, out, err = assert_python_ok("-c", script)

data = out.decode().replace('\r', '')

self.assertEqual(data, "end of main\nend of thread\n")

def test_1_join_on_shutdown(self):

# The usual case: on exit, wait for a non-daemon thread

script = """if 1:

import os

t = threading.Thread(target=joiningfunc,

args=(threading.current_thread(),))

t.start()

time.sleep(0.1)

print('end of main')

"""

self._run_and_join(script)

@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")

@unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")

def test_2_join_in_forked_process(self):

# Like the test above, but from a forked interpreter

script = """if 1:

from test import support

childpid = os.fork()

if childpid != 0:

# parent process

support.wait_process(childpid, exitcode=0)

sys.exit(0)

# child process

t = threading.Thread(target=joiningfunc,

args=(threading.current_thread(),))

t.start()

print('end of main')

"""

self._run_and_join(script)

@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")

@unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")

def test_3_join_in_forked_from_thread(self):

# Like the test above, but fork() was called from a worker thread

# In the forked process, the main Thread object must be marked as stopped.

script = """if 1:

from test import support

main_thread = threading.current_thread()

def worker():

childpid = os.fork()

if childpid != 0:

# parent process

support.wait_process(childpid, exitcode=0)

sys.exit(0)