Skip to content

Commit f3c8138

Browse files
committed
Add test for shared memory names that contain colons
1 parent ac9555d commit f3c8138

File tree

1 file changed

+51
-0
lines changed

1 file changed

+51
-0
lines changed

Lib/test/_test_multiprocessing.py

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7335,3 +7335,54 @@ def test_forkpty(self):
73357335
res = assert_python_failure("-c", code, PYTHONWARNINGS='error')
73367336
self.assertIn(b'DeprecationWarning', res.err)
73377337
self.assertIn(b'is multi-threaded, use of forkpty() may lead to deadlocks in the child', res.err)
7338+
7339+
@unittest.skipUnless(HAS_SHMEM, "requires multiprocessing.shared_memory")
7340+
class TestSharedMemoryNames(unittest.TestCase):
7341+
def test_that_shared_memory_name_with_colons_has_no_resource_tracker_errors(self):
7342+
# Test script that creates and cleans up shared memory with colon in name
7343+
test_script = textwrap.dedent("""
7344+
import sys
7345+
from multiprocessing import shared_memory
7346+
import time
7347+
7348+
# Test various patterns of colons in names
7349+
test_names = [
7350+
"a:b",
7351+
"a:b:c",
7352+
"test:name:with:many:colons",
7353+
":starts:with:colon",
7354+
"ends:with:colon:",
7355+
"::double::colons::",
7356+
]
7357+
7358+
for name in test_names:
7359+
try:
7360+
shm = shared_memory.SharedMemory(create=True, size=100, name=name)
7361+
shm.buf[:5] = b'hello' # Write something to the shared memory
7362+
shm.close()
7363+
shm.unlink()
7364+
7365+
except Exception as e:
7366+
print(f"Error with name '{name}': {e}", file=sys.stderr)
7367+
sys.exit(1)
7368+
7369+
print("SUCCESS")
7370+
""")
7371+
7372+
# Run the test script as a subprocess to capture all output
7373+
result = subprocess.run(
7374+
[sys.executable, '-c', test_script],
7375+
capture_output=True,
7376+
text=True
7377+
)
7378+
7379+
self.assertEqual(result.returncode, 0,
7380+
f"Script failed with stderr: {result.stderr}")
7381+
self.assertIn("SUCCESS", result.stdout)
7382+
7383+
stderr_lower = result.stderr.lower()
7384+
self.assertNotIn("too many values to unpack", stderr_lower,
7385+
f"Resource tracker parsing error found in stderr: {result.stderr}")
7386+
7387+
self.assertNotIn("valueerror", stderr_lower,
7388+
f"ValueError found in stderr: {result.stderr}")

0 commit comments

Comments
 (0)