https://github.com/python/cpython/commit/9a1733d5712545db0703c712b158ae5e2a19e1d0
commit: 9a1733d5712545db0703c712b158ae5e2a19e1d0
branch: main
author: Victor Stinner <[email protected]>
committer: vstinner <[email protected]>
date: 2026-09-15T23:18:12+02:00
summary:
gh-157242: Add support.inject_memory_error() function (#157431)
Add inject_memory_error() and memory_error_cm() functions to
test.support to inject memory errors: make the memory allocator fail.
files:
M Lib/test/support/__init__.py
M Lib/test/test_atexit.py
M Lib/test/test_bytes.py
M Lib/test/test_class.py
M Lib/test/test_exceptions.py
M Lib/test/test_list.py
M Lib/test/test_pyexpat.py
M Lib/test/test_repl.py
M Lib/test/test_str.py
M Lib/test/test_weakref.py
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index 625898e6734aaa..13c7cec64dcf57 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -3540,3 +3540,38 @@ def built_with_c_assertions():
return False
return True
+
+
+def inject_memory_error(start=0, stop=0):
+ """
+ Memory allocation fails after 'start' allocation requests, and until 'stop'
+ allocation requests except when 'stop' is negative or equal to 0 (default)
+ in which case allocation failures never stop.
+
+ Raise SkipTest if the _testcapi extension module is missing
+ """
+ try:
+ import _testcapi
+ except ImportError:
+ raise unittest.SkipTest("_testcapi required")
+
+ _testcapi.set_nomemory(start, stop)
+
+
[email protected]
+def memory_error_cm(start=0, stop=0):
+ """
+ Similar to inject_memory_error() but can be used as a context manager.
+
+ Raise SkipTest if the _testcapi extension module is missing
+ """
+ try:
+ import _testcapi
+ except ImportError:
+ raise unittest.SkipTest("_testcapi required")
+
+ try:
+ _testcapi.set_nomemory(start, stop)
+ yield
+ finally:
+ _testcapi.remove_mem_hooks()
diff --git a/Lib/test/test_atexit.py b/Lib/test/test_atexit.py
index 33c37648da31fc..33d3eb93394781 100644
--- a/Lib/test/test_atexit.py
+++ b/Lib/test/test_atexit.py
@@ -197,14 +197,14 @@ def test_atexit_with_low_memory(self):
# callback doesn't cause an infinite loop during finalization.
code = textwrap.dedent("""
import atexit
- import _testcapi
+ from test.support import inject_memory_error
def callback():
print("hello")
atexit.register(callback)
# Simulate low memory condition
- _testcapi.set_nomemory(0)
+ inject_memory_error()
""")
with os_helper.temp_dir() as temp_dir:
diff --git a/Lib/test/test_bytes.py b/Lib/test/test_bytes.py
index 3297a53dceff00..bab6006df36a44 100644
--- a/Lib/test/test_bytes.py
+++ b/Lib/test/test_bytes.py
@@ -50,16 +50,10 @@ def __index__(self):
@contextlib.contextmanager
-def inject_memory_error(testcase, start):
- # Raise SkipTest if _testcapi extension module is missing
- _testcapi = import_helper.import_module('_testcapi')
-
+def inject_memory_error(testcase, start=0):
with testcase.assertRaises(MemoryError):
- try:
- _testcapi.set_nomemory(start)
+ with support.memory_error_cm(start):
yield
- finally:
- _testcapi.remove_mem_hooks()
class BaseBytesTest:
@@ -1585,7 +1579,7 @@ def test_resize_error(self):
del ba[:offset]
else:
expected = ba.copy()
- with inject_memory_error(self, 0):
+ with inject_memory_error(self):
ba.resize(1024)
self.assertEqual(ba, expected)
@@ -1596,7 +1590,7 @@ def test_resize_error(self):
del ba[:offset]
else:
expected = ba.copy()
- with inject_memory_error(self, 0):
+ with inject_memory_error(self):
ba.resize(1)
self.assertEqual(ba, expected)
@@ -1669,21 +1663,20 @@ def test_take_bytes_error(self):
# gh-157242: If bytearray.take_bytes() fails (MemoryError),
# the bytearray must be left unchanged.
- for logical_offset, to_take, mem_errors in (
+ for logical_offset, to_take, start_list in (
(True, 5, (0, 1)),
(False, 5, (0, 1)),
(True, None, (0,)),
):
- for mem_error in mem_errors:
- with self.subTest(logical_offset=logical_offset,
- to_take=to_take, mem_error=mem_error):
+ for start in start_list:
+ with self.subTest(logical_offset=logical_offset, start=start):
ba = bytearray(b'0123456789')
if logical_offset:
expected = ba[3:]
del ba[:3]
else:
expected = ba.copy()
- with inject_memory_error(self, mem_error):
+ with inject_memory_error(self, start):
ba.take_bytes(to_take)
self.assertEqual(ba, expected)
diff --git a/Lib/test/test_class.py b/Lib/test/test_class.py
index 7bd6d966e8536b..b8b35f68aa7a05 100644
--- a/Lib/test/test_class.py
+++ b/Lib/test/test_class.py
@@ -1016,8 +1016,6 @@ class C:
@support.nomemtest
@isolation.runInSubprocess()
def test_detach_materialized_dict_no_memory(self):
- import _testcapi
-
class A:
def __init__(self):
self.a = 1
@@ -1032,11 +1030,8 @@ def __init__(self):
d = a.__dict__
try:
with support.catch_unraisable_exception() as ex:
- _testcapi.set_nomemory(n, n + 1)
- try:
+ with support.memory_error_cm(n, n + 1):
del a
- finally:
- _testcapi.remove_mem_hooks()
exc_type = ex.unraisable and ex.unraisable.exc_type
except MemoryError:
# The failing allocation was not in the deallocation code.
diff --git a/Lib/test/test_exceptions.py b/Lib/test/test_exceptions.py
index 750e45bb164468..026c2bf6b31e73 100644
--- a/Lib/test/test_exceptions.py
+++ b/Lib/test/test_exceptions.py
@@ -1670,14 +1670,14 @@ def test_recursion_normalizing_with_no_memory(self):
# the size of the list of preallocated MemoryError instances, the
# Fatal Python error message mentions MemoryError.
code = """if 1:
- import _testcapi
+ from test import support
class C(): pass
def recurse(cnt):
cnt -= 1
if cnt:
recurse(cnt)
else:
- _testcapi.set_nomemory(0)
+ support.inject_memory_error()
C()
recurse(16)
"""
@@ -1853,9 +1853,10 @@ def test_unhandled(self):
@support.nomemtest
def test_memory_error_in_PyErr_PrintEx(self):
code = """if 1:
- import _testcapi
+ from test import support
+ stop = %d
class C(): pass
- _testcapi.set_nomemory(0, %d)
+ support.inject_memory_error(0, stop)
C()
"""
@@ -2020,8 +2021,8 @@ def test_exec_set_nomemory_hang(self):
warmup_code = "a = list(range(0, 1))\n" * 60
user_input = warmup_code + dedent("""
try:
- import _testcapi
- _testcapi.set_nomemory(0)
+ from test import support
+ support.inject_memory_error()
b = list(range(1000, 2000))
except Exception as e:
import traceback
diff --git a/Lib/test/test_list.py b/Lib/test/test_list.py
index 1b0724c2b1c99d..260f9f4f607f15 100644
--- a/Lib/test/test_list.py
+++ b/Lib/test/test_list.py
@@ -377,11 +377,12 @@ def test_tier2_invalidates_iterator(self):
def test_no_memory(self):
# gh-118331: Make sure we don't crash if list allocation fails
code = textwrap.dedent("""
- import _testcapi, sys
+ from test import support
+ import sys
# Prime the freelist
l = [None]
del l
- _testcapi.set_nomemory(0)
+ support.inject_memory_error()
l = [None]
""")
rc, _, _ = assert_python_failure("-c", code)
diff --git a/Lib/test/test_pyexpat.py b/Lib/test/test_pyexpat.py
index baa4f178427d53..23b82dc1fd2179 100644
--- a/Lib/test/test_pyexpat.py
+++ b/Lib/test/test_pyexpat.py
@@ -1077,18 +1077,9 @@ def test_error_path_no_crash(self):
parser.buffer_text = True
rc_before = sys.getrefcount(parser)
- # We avoid self.assertRaises(MemoryError) here because the
- # context manager itself needs memory allocations that fail
- # while the nomemory hook is active.
- self.testcapi.set_nomemory(1, 10)
- raised = False
- try:
- parser.ExternalEntityParserCreate(None)
- except MemoryError:
- raised = True
- finally:
- self.testcapi.remove_mem_hooks()
- self.assertTrue(raised, "MemoryError not raised")
+ with self.assertRaises(MemoryError):
+ with support.memory_error_cm(1, 10):
+ parser.ExternalEntityParserCreate(None)
rc_after = sys.getrefcount(parser)
self.assertEqual(rc_after, rc_before)
diff --git a/Lib/test/test_repl.py b/Lib/test/test_repl.py
index 372c110783bce7..ed6eb706c40d22 100644
--- a/Lib/test/test_repl.py
+++ b/Lib/test/test_repl.py
@@ -104,10 +104,11 @@ def test_no_memory(self):
# no memory. Check also that the fix does not break the interactive
# loop when an exception is raised.
user_input = """
- import sys, _testcapi
+ import sys
+ from test import support
1/0
print('After the exception.')
- _testcapi.set_nomemory(0)
+ support.inject_memory_error()
sys.exit(0)
"""
user_input = dedent(user_input)
diff --git a/Lib/test/test_str.py b/Lib/test/test_str.py
index 979bfe36fff680..17163182be08c4 100644
--- a/Lib/test/test_str.py
+++ b/Lib/test/test_str.py
@@ -613,14 +613,9 @@ def test_replace_oom(self):
s1 = "轘" * 4
s2 = "&"
s3 = "&"
- assertion = self.assertRaises(MemoryError)
- _testcapi.set_nomemory(0, 0)
- try:
- # No allocations made in the test itself:
- with assertion:
+ with self.assertRaises(MemoryError):
+ with support.memory_error_cm():
s1.replace(s2, s3) # this line used to crash before
- finally:
- _testcapi.remove_mem_hooks()
def test_repeat_id_preserving(self):
a = '123abc1@'
diff --git a/Lib/test/test_weakref.py b/Lib/test/test_weakref.py
index b9d1745592c09c..ce93481e2735f7 100644
--- a/Lib/test/test_weakref.py
+++ b/Lib/test/test_weakref.py
@@ -1029,8 +1029,8 @@ def test_no_memory_when_clearing(self):
# gh-118331: Make sure we do not raise an exception from the destructor
# when clearing weakrefs if allocating the intermediate tuple fails.
code = textwrap.dedent("""
- import _testcapi
import weakref
+ from test import support
class TestObj:
pass
@@ -1042,7 +1042,7 @@ def callback(obj):
# The choice of 50 is arbitrary, but must be large enough to ensure
# the allocation won't be serviced by the free list.
wrs = [weakref.ref(obj, callback) for _ in range(50)]
- _testcapi.set_nomemory(0)
+ support.inject_memory_error()
del obj
""").strip()
res, _ = script_helper.run_python_until_end("-c", code)
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
