mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-27 12:16:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

gh-115942: Add locked to several multiprocessing locks (#115944)

Browse source 
Co-authored-by: mpage <mpage@cs.stanford.edu>
Co-authored-by: Hugo van Kemenade <1324225+hugovk@users.noreply.github.com>
This commit is contained in:
sobolevn 2025-04-08 11:14:12 +03:00 • committed by GitHub
parent 6cd1d6c6b1
commit f7305a06c7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 89 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

17
Lib/test/_test_multiprocessing.py
View file

@ -1486,8 +1486,10 @@ class _TestLock(BaseTestCase):
def test_lock(self):
lock = self.Lock()
self.assertEqual(lock.acquire(), True)
self.assertTrue(lock.locked())
self.assertEqual(lock.acquire(False), False)
self.assertEqual(lock.release(), None)
self.assertFalse(lock.locked())
self.assertRaises((ValueError, threading.ThreadError), lock.release)
@staticmethod
@ -1549,16 +1551,23 @@ class _TestLock(BaseTestCase):
def test_rlock(self):
lock = self.RLock()
self.assertEqual(lock.acquire(), True)
self.assertTrue(lock.locked())
self.assertEqual(lock.acquire(), True)
self.assertEqual(lock.acquire(), True)
self.assertEqual(lock.release(), None)
self.assertTrue(lock.locked())
self.assertEqual(lock.release(), None)
self.assertEqual(lock.release(), None)
self.assertFalse(lock.locked())
self.assertRaises((AssertionError, RuntimeError), lock.release)
def test_lock_context(self):
with self.Lock():
pass
with self.Lock() as locked:
self.assertTrue(locked)
def test_rlock_context(self):
with self.RLock() as locked:
self.assertTrue(locked)
class _TestSemaphore(BaseTestCase):
@ -6254,6 +6263,7 @@ class TestSyncManagerTypes(unittest.TestCase):
@classmethod
def _test_lock(cls, obj):
obj.acquire()
obj.locked()
def test_lock(self, lname="Lock"):
o = getattr(self.manager, lname)()
@ -6265,8 +6275,9 @@ class TestSyncManagerTypes(unittest.TestCase):
def _test_rlock(cls, obj):
obj.acquire()
obj.release()
obj.locked()
def test_rlock(self, lname="Lock"):
def test_rlock(self, lname="RLock"):
o = getattr(self.manager, lname)()
self.run_worker(self._test_rlock, o)