Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 43 additions & 19 deletions Lib/test/test_threading_local.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,18 @@
from doctest import DocTestSuite
from test import support
from test.support import threading_helper
from test.support.import_helper import import_module
import weakref
import gc

# Modules under test
import _thread
import threading
import _threading_local


threading_helper.requires_working_threading(module=True)


class Weak(object):
pass

Expand All @@ -23,7 +26,7 @@ def target(local, weaklist):

class BaseLocalTest:

@unittest.skip("TODO: RUSTPYTHON, flaky test")
@unittest.skip('TODO: RUSTPYTHON; flaky test')
def test_local_refs(self):
self._local_refs(20)
self._local_refs(50)
Expand Down Expand Up @@ -182,8 +185,7 @@ class LocalSubclass(self._local):
"""To test that subclasses behave properly."""
self._test_dict_attribute(LocalSubclass)

# TODO: RUSTPYTHON, cycle detection/collection
@unittest.expectedFailure
@unittest.expectedFailure # TODO: RUSTPYTHON; cycle detection/collection
def test_cycle_collection(self):
class X:
pass
Expand All @@ -197,35 +199,57 @@ class X:
self.assertIsNone(wr())


def test_threading_local_clear_race(self):
# See https://github.com/python/cpython/issues/100892

_testcapi = import_module('_testcapi')
_testcapi.call_in_temporary_c_thread(lambda: None, False)

for _ in range(1000):
_ = threading.local()

_testcapi.join_temporary_c_thread()

@support.cpython_only
def test_error(self):
class Loop(self._local):
attr = 1

# Trick the "if name == '__dict__':" test of __setattr__()
# to always be true
class NameCompareTrue:
def __eq__(self, other):
return True

loop = Loop()
with self.assertRaisesRegex(AttributeError, 'Loop.*read-only'):
loop.__setattr__(NameCompareTrue(), 2)


class ThreadLocalTest(unittest.TestCase, BaseLocalTest):
_local = _thread._local

# TODO: RUSTPYTHON, __new__ vs __init__ cooperation
@unittest.expectedFailure
def test_arguments():
super().test_arguments()

@unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: TypeError not raised by _local
def test_arguments(self):
return super().test_arguments()

class PyThreadingLocalTest(unittest.TestCase, BaseLocalTest):
_local = _threading_local.local


def test_main():
suite = unittest.TestSuite()
suite.addTest(DocTestSuite('_threading_local'))
suite.addTest(unittest.makeSuite(ThreadLocalTest))
suite.addTest(unittest.makeSuite(PyThreadingLocalTest))
def load_tests(loader, tests, pattern):
tests.addTest(DocTestSuite('_threading_local'))

local_orig = _threading_local.local
def setUp(test):
_threading_local.local = _thread._local
def tearDown(test):
_threading_local.local = local_orig
suite.addTest(DocTestSuite('_threading_local',
setUp=setUp, tearDown=tearDown)
)
tests.addTests(DocTestSuite('_threading_local',
setUp=setUp, tearDown=tearDown)
)
return tests

support.run_unittest(suite)

if __name__ == '__main__':
test_main()
unittest.main()
Loading