Stowage/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-12-08 06:10:17 +00:00
Code Issues Projects Releases Packages Wiki Activity

gh-142206: multiprocessing.resource_tracker: Decode messages using older protocol (GH-142215)

Browse source
This commit is contained in:
Petr Viktorin 2025-12-03 13:59:14 +01:00 • committed by GitHub
parent 88cd5d9850
commit 4172644d78
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 73 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

26
Lib/test/_test_multiprocessing.py
View file

@ -39,6 +39,7 @@
from test.support import socket_helper
from test.support import threading_helper
from test.support import warnings_helper
from test.support import subTests
from test.support.script_helper import assert_python_failure, assert_python_ok
# Skip tests if _multiprocessing wasn't built.
@ -4383,6 +4384,19 @@ def test_copy(self):
self.assertEqual(bar.z, 2 ** 33)
def resource_tracker_format_subtests(func):
"""Run given test using both resource tracker communication formats"""
def _inner(self, *args, **kwargs):
tracker = resource_tracker._resource_tracker
for use_simple_format in False, True:
with (
self.subTest(use_simple_format=use_simple_format),
unittest.mock.patch.object(
tracker, '_use_simple_format', use_simple_format)
):
func(self, *args, **kwargs)
return _inner
@unittest.skipUnless(HAS_SHMEM, "requires multiprocessing.shared_memory")
@hashlib_helper.requires_hashdigest('sha256')
class _TestSharedMemory(BaseTestCase):
@ -4662,6 +4676,7 @@ def test_shared_memory_SharedMemoryServer_ignores_sigint(self):
smm.shutdown()
@unittest.skipIf(os.name != "posix", "resource_tracker is posix only")
@resource_tracker_format_subtests
def test_shared_memory_SharedMemoryManager_reuses_resource_tracker(self):
# bpo-36867: test that a SharedMemoryManager uses the
# same resource_tracker process as its parent.
@ -4913,6 +4928,7 @@ def test_shared_memory_cleaned_after_process_termination(self):
"shared_memory objects to clean up at shutdown", err)
@unittest.skipIf(os.name != "posix", "resource_tracker is posix only")
@resource_tracker_format_subtests
def test_shared_memory_untracking(self):
# gh-82300: When a separate Python process accesses shared memory
# with track=False, it must not cause the memory to be deleted
@ -4940,6 +4956,7 @@ def test_shared_memory_untracking(self):
mem.close()
@unittest.skipIf(os.name != "posix", "resource_tracker is posix only")
@resource_tracker_format_subtests
def test_shared_memory_tracking(self):
# gh-82300: When a separate Python process accesses shared memory
# with track=True, it must cause the memory to be deleted when
@ -7353,13 +7370,18 @@ def test_forkpty(self):
@unittest.skipUnless(HAS_SHMEM, "requires multiprocessing.shared_memory")
class TestSharedMemoryNames(unittest.TestCase):
def test_that_shared_memory_name_with_colons_has_no_resource_tracker_errors(self):
@subTests('use_simple_format', (True, False))
def test_that_shared_memory_name_with_colons_has_no_resource_tracker_errors(
self, use_simple_format):
# Test script that creates and cleans up shared memory with colon in name
test_script = textwrap.dedent("""
import sys
from multiprocessing import shared_memory
from multiprocessing import resource_tracker
import time
resource_tracker._resource_tracker._use_simple_format = %s
# Test various patterns of colons in names
test_names = [
"a:b",
@ -7387,7 +7409,7 @@ def test_that_shared_memory_name_with_colons_has_no_resource_tracker_errors(self
sys.exit(1)
print("SUCCESS")
""")
""" % use_simple_format)
rc, out, err = assert_python_ok("-c", test_script)
self.assertIn(b"SUCCESS", out)