aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/Lib/test/test_threading.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_threading.py')
-rw-r--r--Lib/test/test_threading.py35
1 files changed, 33 insertions, 2 deletions
diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py
index abe63c10c0a..0e51e7fc8c5 100644
--- a/Lib/test/test_threading.py
+++ b/Lib/test/test_threading.py
@@ -1352,6 +1352,35 @@ class ThreadTests(BaseTestCase):
''')
assert_python_ok("-c", script)
+ @skip_unless_reliable_fork
+ @unittest.skipUnless(hasattr(threading, 'get_native_id'), "test needs threading.get_native_id()")
+ def test_native_id_after_fork(self):
+ script = """if True:
+ import threading
+ import os
+ from test import support
+
+ parent_thread_native_id = threading.current_thread().native_id
+ print(parent_thread_native_id, flush=True)
+ assert parent_thread_native_id == threading.get_native_id()
+ childpid = os.fork()
+ if childpid == 0:
+ print(threading.current_thread().native_id, flush=True)
+ assert threading.current_thread().native_id == threading.get_native_id()
+ else:
+ try:
+ assert parent_thread_native_id == threading.current_thread().native_id
+ assert parent_thread_native_id == threading.get_native_id()
+ finally:
+ support.wait_process(childpid, exitcode=0)
+ """
+ rc, out, err = assert_python_ok('-c', script)
+ self.assertEqual(rc, 0)
+ self.assertEqual(err, b"")
+ native_ids = out.strip().splitlines()
+ self.assertEqual(len(native_ids), 2)
+ self.assertNotEqual(native_ids[0], native_ids[1])
+
class ThreadJoinOnShutdown(BaseTestCase):
def _run_and_join(self, script):
@@ -2137,8 +2166,7 @@ class CRLockTests(lock_tests.RLockTests):
]
for args, kwargs in arg_types:
with self.subTest(args=args, kwargs=kwargs):
- with self.assertWarns(DeprecationWarning):
- threading.RLock(*args, **kwargs)
+ self.assertRaises(TypeError, threading.RLock, *args, **kwargs)
# Subtypes with custom `__init__` are allowed (but, not recommended):
class CustomRLock(self.locktype):
@@ -2156,6 +2184,9 @@ class ConditionAsRLockTests(lock_tests.RLockTests):
# Condition uses an RLock by default and exports its API.
locktype = staticmethod(threading.Condition)
+ def test_constructor_noargs(self):
+ self.skipTest("Condition allows positional arguments")
+
def test_recursion_count(self):
self.skipTest("Condition does not expose _recursion_count()")