gh-157242: Add support.inject_memory_error() function

Add inject_memory_error() and with_memory_error() functions to
test.support to inject memory errors: make the memory allocator fail.

Author: vstinner

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)
+
+
+@contextlib.contextmanager
+def with_memory_error(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()

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:

Lib/test/test_bytes.py

@@ -50,16 +50,13 @@ def __index__(self):
 
 
 @contextlib.contextmanager
-def inject_memory_error(testcase, start):
+def inject_memory_error(testcase, start=0):
     # Raise SkipTest if _testcapi extension module is missing
     _testcapi = import_helper.import_module('_testcapi')
 
     with testcase.assertRaises(MemoryError):
-        try:
-            _testcapi.set_nomemory(start)
+        with support.with_memory_error(start):
             yield
-        finally:
-            _testcapi.remove_mem_hooks()
 
 
 class BaseBytesTest:
@@ -1585,7 +1582,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 +1593,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 +1666,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)
 

Lib/test/test_class.py

@@ -1032,11 +1032,8 @@ def __init__(self):
             d = a.__dict__
             try:
                 with support.catch_unraisable_exception() as ex:
-                    _testcapi.set_nomemory(n, n + 1)
-                    try:
+                    with support.with_memory_error(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.

Lib/test/test_exceptions.py

@@ -1660,14 +1660,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)
         """
@@ -1843,9 +1843,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()
         """
 
@@ -2010,8 +2011,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

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)

Lib/test/test_pyexpat.py

@@ -1080,14 +1080,12 @@ def test_error_path_no_crash(self):
         # 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)
+            with support.with_memory_error(1, 10):
+                parser.ExternalEntityParserCreate(None)
         except MemoryError:
             raised = True
-        finally:
-            self.testcapi.remove_mem_hooks()
         self.assertTrue(raised, "MemoryError not raised")
 
         rc_after = sys.getrefcount(parser)

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)

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.with_memory_error():
                 s1.replace(s2, s3)  # this line used to crash before
-        finally:
-            _testcapi.remove_mem_hooks()
 
     def test_repeat_id_preserving(self):
         a = '123abc1@'

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)