mirror of
https://github.com/python/cpython.git
synced 2025-08-03 16:39:00 +00:00
45bc120d45
The `free_work_item()` function in QSBR may call arbitrary code via Python object destructors, which may reenter the QSBR code. Reorder the processing of work items to be robust to reentrancy. Also fix the TODO for the out of memory situation.
227 lines
7.9 KiB
Python
227 lines
7.9 KiB
Python
import enum
|
|
import textwrap
|
|
import unittest
|
|
from test import support
|
|
from test.support import import_helper
|
|
from test.support import os_helper
|
|
from test.support import threading_helper
|
|
from test.support.script_helper import assert_python_failure
|
|
|
|
|
|
_testlimitedcapi = import_helper.import_module('_testlimitedcapi')
|
|
_testcapi = import_helper.import_module('_testcapi')
|
|
_testinternalcapi = import_helper.import_module('_testinternalcapi')
|
|
|
|
|
|
class Constant(enum.IntEnum):
|
|
Py_CONSTANT_NONE = 0
|
|
Py_CONSTANT_FALSE = 1
|
|
Py_CONSTANT_TRUE = 2
|
|
Py_CONSTANT_ELLIPSIS = 3
|
|
Py_CONSTANT_NOT_IMPLEMENTED = 4
|
|
Py_CONSTANT_ZERO = 5
|
|
Py_CONSTANT_ONE = 6
|
|
Py_CONSTANT_EMPTY_STR = 7
|
|
Py_CONSTANT_EMPTY_BYTES = 8
|
|
Py_CONSTANT_EMPTY_TUPLE = 9
|
|
|
|
INVALID_CONSTANT = Py_CONSTANT_EMPTY_TUPLE + 1
|
|
|
|
|
|
class GetConstantTest(unittest.TestCase):
|
|
def check_get_constant(self, get_constant):
|
|
self.assertIs(get_constant(Constant.Py_CONSTANT_NONE), None)
|
|
self.assertIs(get_constant(Constant.Py_CONSTANT_FALSE), False)
|
|
self.assertIs(get_constant(Constant.Py_CONSTANT_TRUE), True)
|
|
self.assertIs(get_constant(Constant.Py_CONSTANT_ELLIPSIS), Ellipsis)
|
|
self.assertIs(get_constant(Constant.Py_CONSTANT_NOT_IMPLEMENTED), NotImplemented)
|
|
|
|
for constant_id, constant_type, value in (
|
|
(Constant.Py_CONSTANT_ZERO, int, 0),
|
|
(Constant.Py_CONSTANT_ONE, int, 1),
|
|
(Constant.Py_CONSTANT_EMPTY_STR, str, ""),
|
|
(Constant.Py_CONSTANT_EMPTY_BYTES, bytes, b""),
|
|
(Constant.Py_CONSTANT_EMPTY_TUPLE, tuple, ()),
|
|
):
|
|
with self.subTest(constant_id=constant_id):
|
|
obj = get_constant(constant_id)
|
|
self.assertEqual(type(obj), constant_type, obj)
|
|
self.assertEqual(obj, value)
|
|
|
|
with self.assertRaises(SystemError):
|
|
get_constant(Constant.INVALID_CONSTANT)
|
|
|
|
def test_get_constant(self):
|
|
self.check_get_constant(_testlimitedcapi.get_constant)
|
|
|
|
def test_get_constant_borrowed(self):
|
|
self.check_get_constant(_testlimitedcapi.get_constant_borrowed)
|
|
|
|
|
|
class PrintTest(unittest.TestCase):
|
|
def testPyObjectPrintObject(self):
|
|
|
|
class PrintableObject:
|
|
|
|
def __repr__(self):
|
|
return "spam spam spam"
|
|
|
|
def __str__(self):
|
|
return "egg egg egg"
|
|
|
|
obj = PrintableObject()
|
|
output_filename = os_helper.TESTFN
|
|
self.addCleanup(os_helper.unlink, output_filename)
|
|
|
|
# Test repr printing
|
|
_testcapi.call_pyobject_print(obj, output_filename, False)
|
|
with open(output_filename, 'r') as output_file:
|
|
self.assertEqual(output_file.read(), repr(obj))
|
|
|
|
# Test str printing
|
|
_testcapi.call_pyobject_print(obj, output_filename, True)
|
|
with open(output_filename, 'r') as output_file:
|
|
self.assertEqual(output_file.read(), str(obj))
|
|
|
|
def testPyObjectPrintNULL(self):
|
|
output_filename = os_helper.TESTFN
|
|
self.addCleanup(os_helper.unlink, output_filename)
|
|
|
|
# Test repr printing
|
|
_testcapi.pyobject_print_null(output_filename)
|
|
with open(output_filename, 'r') as output_file:
|
|
self.assertEqual(output_file.read(), '<nil>')
|
|
|
|
def testPyObjectPrintNoRefObject(self):
|
|
output_filename = os_helper.TESTFN
|
|
self.addCleanup(os_helper.unlink, output_filename)
|
|
|
|
# Test repr printing
|
|
correct_output = _testcapi.pyobject_print_noref_object(output_filename)
|
|
with open(output_filename, 'r') as output_file:
|
|
self.assertEqual(output_file.read(), correct_output)
|
|
|
|
def testPyObjectPrintOSError(self):
|
|
output_filename = os_helper.TESTFN
|
|
self.addCleanup(os_helper.unlink, output_filename)
|
|
|
|
open(output_filename, "w+").close()
|
|
with self.assertRaises(OSError):
|
|
_testcapi.pyobject_print_os_error(output_filename)
|
|
|
|
|
|
class ClearWeakRefsNoCallbacksTest(unittest.TestCase):
|
|
"""Test PyUnstable_Object_ClearWeakRefsNoCallbacks"""
|
|
def test_ClearWeakRefsNoCallbacks(self):
|
|
"""Ensure PyUnstable_Object_ClearWeakRefsNoCallbacks works"""
|
|
import weakref
|
|
import gc
|
|
class C:
|
|
pass
|
|
obj = C()
|
|
messages = []
|
|
ref = weakref.ref(obj, lambda: messages.append("don't add this"))
|
|
self.assertIs(ref(), obj)
|
|
self.assertFalse(messages)
|
|
_testcapi.pyobject_clear_weakrefs_no_callbacks(obj)
|
|
self.assertIsNone(ref())
|
|
gc.collect()
|
|
self.assertFalse(messages)
|
|
|
|
def test_ClearWeakRefsNoCallbacks_no_weakref_support(self):
|
|
"""Don't fail on objects that don't support weakrefs"""
|
|
import weakref
|
|
obj = object()
|
|
with self.assertRaises(TypeError):
|
|
ref = weakref.ref(obj)
|
|
_testcapi.pyobject_clear_weakrefs_no_callbacks(obj)
|
|
|
|
|
|
@threading_helper.requires_working_threading()
|
|
class EnableDeferredRefcountingTest(unittest.TestCase):
|
|
"""Test PyUnstable_Object_EnableDeferredRefcount"""
|
|
@support.requires_resource("cpu")
|
|
def test_enable_deferred_refcount(self):
|
|
from threading import Thread
|
|
|
|
self.assertEqual(_testcapi.pyobject_enable_deferred_refcount("not tracked"), 0)
|
|
foo = []
|
|
self.assertEqual(_testcapi.pyobject_enable_deferred_refcount(foo), int(support.Py_GIL_DISABLED))
|
|
|
|
# Make sure reference counting works on foo now
|
|
self.assertEqual(foo, [])
|
|
if support.Py_GIL_DISABLED:
|
|
self.assertTrue(_testinternalcapi.has_deferred_refcount(foo))
|
|
|
|
# Make sure that PyUnstable_Object_EnableDeferredRefcount is thread safe
|
|
def silly_func(obj):
|
|
self.assertIn(
|
|
_testcapi.pyobject_enable_deferred_refcount(obj),
|
|
(0, 1)
|
|
)
|
|
|
|
silly_list = [1, 2, 3]
|
|
threads = [
|
|
Thread(target=silly_func, args=(silly_list,)) for _ in range(4)
|
|
]
|
|
|
|
with threading_helper.start_threads(threads):
|
|
for i in range(10):
|
|
silly_list.append(i)
|
|
|
|
if support.Py_GIL_DISABLED:
|
|
self.assertTrue(_testinternalcapi.has_deferred_refcount(silly_list))
|
|
|
|
|
|
class CAPITest(unittest.TestCase):
|
|
def check_negative_refcount(self, code):
|
|
# bpo-35059: Check that Py_DECREF() reports the correct filename
|
|
# when calling _Py_NegativeRefcount() to abort Python.
|
|
code = textwrap.dedent(code)
|
|
rc, out, err = assert_python_failure('-c', code)
|
|
self.assertRegex(err,
|
|
br'object\.c:[0-9]+: '
|
|
br'_Py_NegativeRefcount: Assertion failed: '
|
|
br'object has negative ref count')
|
|
|
|
@unittest.skipUnless(hasattr(_testcapi, 'negative_refcount'),
|
|
'need _testcapi.negative_refcount()')
|
|
def test_negative_refcount(self):
|
|
code = """
|
|
import _testcapi
|
|
from test import support
|
|
|
|
with support.SuppressCrashReport():
|
|
_testcapi.negative_refcount()
|
|
"""
|
|
self.check_negative_refcount(code)
|
|
|
|
@unittest.skipUnless(hasattr(_testcapi, 'decref_freed_object'),
|
|
'need _testcapi.decref_freed_object()')
|
|
@support.skip_if_sanitizer("use after free on purpose",
|
|
address=True, memory=True, ub=True)
|
|
def test_decref_freed_object(self):
|
|
code = """
|
|
import _testcapi
|
|
from test import support
|
|
|
|
with support.SuppressCrashReport():
|
|
_testcapi.decref_freed_object()
|
|
"""
|
|
self.check_negative_refcount(code)
|
|
|
|
def test_decref_delayed(self):
|
|
# gh-130519: Test that _PyObject_XDecRefDelayed() and QSBR code path
|
|
# handles destructors that are possibly re-entrant or trigger a GC.
|
|
import gc
|
|
|
|
class MyObj:
|
|
def __del__(self):
|
|
gc.collect()
|
|
|
|
for _ in range(1000):
|
|
obj = MyObj()
|
|
_testinternalcapi.incref_decref_delayed(obj)
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|