Skip to content

Update test_{dict,weakref}.py from 3.13.5 #5963

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
youknowone merged 2 commits into from
Jul 14, 2025
Merged

Update test_{dict,weakref}.py from 3.13.5 #5963

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1c9eac3
Update test_dict.py from 3.13.5
ShaharNaveh Jul 13, 2025
61e54c5
Update `test_weakref.py` from 3.13.5
ShaharNaveh Jul 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Update test_weakref.py from 3.13.5
  • Loading branch information
@ShaharNaveh
ShaharNaveh committed Jul 13, 2025
commit 61e54c5d5fb01ed6560c326869e53a942fdce403
209 changes: 156 additions & 53 deletions Lib/test/test_weakref.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import gc
import sys
import doctest
import unittest
import collections
import weakref
Expand All @@ -9,10 +10,14 @@
import threading
import time
import random
import textwrap

from test import support
from test.support import script_helper, ALWAYS_EQ
from test.support import script_helper, ALWAYS_EQ, suppress_immortalization
from test.support import gc_collect
from test.support import import_helper
from test.support import threading_helper
from test.support import is_wasi, Py_DEBUG

# Used in ReferencesTestCase.test_ref_created_during_del() .
ref_from_del = None
Expand Down Expand Up @@ -77,7 +82,7 @@ def callback(self, ref):


@contextlib.contextmanager
def collect_in_thread(period=0.0001):
def collect_in_thread(period=0.005):
"""
Ensure GC collections happen in a different thread, at a high frequency.
"""
Expand Down Expand Up @@ -114,14 +119,57 @@ def test_basic_ref(self):
del o
repr(wr)

@support.cpython_only
def test_ref_repr(self):
obj = C()
ref = weakref.ref(obj)
regex = (
rf"<weakref at 0x[0-9a-fA-F]+; "
rf"to '{'' if __name__ == '__main__' else C.__module__ + '.'}{C.__qualname__}' "
rf"at 0x[0-9a-fA-F]+>"
)
self.assertRegex(repr(ref), regex)

obj = None
gc_collect()
self.assertRegex(repr(ref),
rf'<weakref at 0x[0-9a-fA-F]+; dead>')

# test type with __name__
class WithName:
@property
def __name__(self):
return "custom_name"

obj2 = WithName()
ref2 = weakref.ref(obj2)
regex = (
rf"<weakref at 0x[0-9a-fA-F]+; "
rf"to '{'' if __name__ == '__main__' else WithName.__module__ + '.'}"
rf"{WithName.__qualname__}' "
rf"at 0x[0-9a-fA-F]+ +\(custom_name\)>"
)
self.assertRegex(repr(ref2), regex)

def test_repr_failure_gh99184(self):
class MyConfig(dict):
def __getattr__(self, x):
return self[x]

obj = MyConfig(offset=5)
obj_weakref = weakref.ref(obj)

self.assertIn('MyConfig', repr(obj_weakref))
self.assertIn('MyConfig', str(obj_weakref))

def test_basic_callback(self):
self.check_basic_callback(C)
self.check_basic_callback(create_function)
self.check_basic_callback(create_bound_method)

@support.cpython_only
def test_cfunction(self):
import _testcapi
_testcapi = import_helper.import_module("_testcapi")
create_cfunction = _testcapi.create_cfunction
f = create_cfunction()
wr = weakref.ref(f)
Expand Down Expand Up @@ -182,6 +230,22 @@ def check(proxy):
self.assertRaises(ReferenceError, bool, ref3)
self.assertEqual(self.cbcalled, 2)

@support.cpython_only
def test_proxy_repr(self):
obj = C()
ref = weakref.proxy(obj, self.callback)
regex = (
rf"<weakproxy at 0x[0-9a-fA-F]+; "
rf"to '{'' if __name__ == '__main__' else C.__module__ + '.'}{C.__qualname__}' "
rf"at 0x[0-9a-fA-F]+>"
)
self.assertRegex(repr(ref), regex)

obj = None
gc_collect()
self.assertRegex(repr(ref),
rf'<weakproxy at 0x[0-9a-fA-F]+; dead>')

def check_basic_ref(self, factory):
o = factory()
ref = weakref.ref(o)
Expand Down Expand Up @@ -613,7 +677,8 @@ class C(object):
# deallocation of c2.
del c2

def test_callback_in_cycle_1(self):
@suppress_immortalization()
def test_callback_in_cycle(self):
import gc

class J(object):
Expand Down Expand Up @@ -653,40 +718,11 @@ def acallback(self, ignore):
del I, J, II
gc.collect()

def test_callback_in_cycle_2(self):
def test_callback_reachable_one_way(self):
import gc

# This is just like test_callback_in_cycle_1, except that II is an
# old-style class. The symptom is different then: an instance of an
# old-style class looks in its own __dict__ first. 'J' happens to
# get cleared from I.__dict__ before 'wr', and 'J' was never in II's
# __dict__, so the attribute isn't found. The difference is that
# the old-style II doesn't have a NULL __mro__ (it doesn't have any
# __mro__), so no segfault occurs. Instead it got:
# test_callback_in_cycle_2 (__main__.ReferencesTestCase) ...
# Exception exceptions.AttributeError:
# "II instance has no attribute 'J'" in <bound method II.acallback
# of <?.II instance at 0x00B9B4B8>> ignored

class J(object):
pass

class II:
def acallback(self, ignore):
self.J

I = II()
I.J = J
I.wr = weakref.ref(J, I.acallback)

del I, J, II
gc.collect()

def test_callback_in_cycle_3(self):
import gc

# This one broke the first patch that fixed the last two. In this
# case, the objects reachable from the callback aren't also reachable
# This one broke the first patch that fixed the previous test. In this case,
# the objects reachable from the callback aren't also reachable
# from the object (c1) *triggering* the callback: you can get to
# c1 from c2, but not vice-versa. The result was that c2's __dict__
# got tp_clear'ed by the time the c2.cb callback got invoked.
Expand All @@ -706,10 +742,10 @@ def cb(self, ignore):
del c1, c2
gc.collect()

def test_callback_in_cycle_4(self):
def test_callback_different_classes(self):
import gc

# Like test_callback_in_cycle_3, except c2 and c1 have different
# Like test_callback_reachable_one_way, except c2 and c1 have different
# classes. c2's class (C) isn't reachable from c1 then, so protecting
# objects reachable from the dying object (c1) isn't enough to stop
# c2's class (C) from getting tp_clear'ed before c2.cb is invoked.
Expand All @@ -736,6 +772,7 @@ class D:

# TODO: RUSTPYTHON
@unittest.expectedFailure
@suppress_immortalization()
def test_callback_in_cycle_resurrection(self):
import gc

Expand Down Expand Up @@ -879,6 +916,7 @@ def test_init(self):
# No exception should be raised here
gc.collect()

@suppress_immortalization()
def test_classes(self):
# Check that classes are weakrefable.
class A(object):
Expand Down Expand Up @@ -958,6 +996,7 @@ def test_hashing(self):
self.assertEqual(hash(a), hash(42))
self.assertRaises(TypeError, hash, b)

@unittest.skipIf(is_wasi and Py_DEBUG, "requires deep stack")
def test_trashcan_16602(self):
# Issue #16602: when a weakref's target was part of a long
# deallocation chain, the trashcan mechanism could delay clearing
Expand Down Expand Up @@ -1015,6 +1054,31 @@ def __del__(self): pass
del x
support.gc_collect()

@support.cpython_only
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

class TestObj:
pass

def callback(obj):
pass

obj = TestObj()
# 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)
del obj
""").strip()
res, _ = script_helper.run_python_until_end("-c", code)
stderr = res.err.decode("ascii", "backslashreplace")
self.assertNotRegex(stderr, "_Py_Dealloc: Deallocator of type 'TestObj'")


class SubclassableWeakrefTestCase(TestBase):

Expand Down Expand Up @@ -1267,6 +1331,12 @@ class MappingTestCase(TestBase):

COUNT = 10

if support.check_sanitizer(thread=True) and support.Py_GIL_DISABLED:
# Reduce iteration count to get acceptable latency
NUM_THREADED_ITERATIONS = 1000
else:
NUM_THREADED_ITERATIONS = 100000

def check_len_cycles(self, dict_type, cons):
N = 20
items = [RefCycle() for i in range(N)]
Expand Down Expand Up @@ -1898,34 +1968,56 @@ def test_make_weak_keyed_dict_repr(self):
dict = weakref.WeakKeyDictionary()
self.assertRegex(repr(dict), '<WeakKeyDictionary at 0x.*>')

@threading_helper.requires_working_threading()
def test_threaded_weak_valued_setdefault(self):
d = weakref.WeakValueDictionary()
with collect_in_thread():
for i in range(100000):
for i in range(self.NUM_THREADED_ITERATIONS):
x = d.setdefault(10, RefCycle())
self.assertIsNot(x, None) # we never put None in there!
del x

@threading_helper.requires_working_threading()
def test_threaded_weak_valued_pop(self):
d = weakref.WeakValueDictionary()
with collect_in_thread():
for i in range(100000):
for i in range(self.NUM_THREADED_ITERATIONS):
d[10] = RefCycle()
x = d.pop(10, 10)
self.assertIsNot(x, None) # we never put None in there!

@threading_helper.requires_working_threading()
def test_threaded_weak_valued_consistency(self):
# Issue #28427: old keys should not remove new values from
# WeakValueDictionary when collecting from another thread.
d = weakref.WeakValueDictionary()
with collect_in_thread():
for i in range(200000):
for i in range(2 * self.NUM_THREADED_ITERATIONS):
o = RefCycle()
d[10] = o
# o is still alive, so the dict can't be empty
self.assertEqual(len(d), 1)
o = None # lose ref

@support.cpython_only
def test_weak_valued_consistency(self):
# A single-threaded, deterministic repro for issue #28427: old keys
# should not remove new values from WeakValueDictionary. This relies on
# an implementation detail of CPython's WeakValueDictionary (its
# underlying dictionary of KeyedRefs) to reproduce the issue.
d = weakref.WeakValueDictionary()
with support.disable_gc():
d[10] = RefCycle()
# Keep the KeyedRef alive after it's replaced so that GC will invoke
# the callback.
wr = d.data[10]
# Replace the value with something that isn't cyclic garbage
o = RefCycle()
d[10] = o
# Trigger GC, which will invoke the callback for `wr`
gc.collect()
self.assertEqual(len(d), 1)

def check_threaded_weak_dict_copy(self, type_, deepcopy):
# `type_` should be either WeakKeyDictionary or WeakValueDictionary.
# `deepcopy` should be either True or False.
Expand Down Expand Up @@ -1987,22 +2079,28 @@ def pop_and_collect(lst):
if exc:
raise exc[0]

@threading_helper.requires_working_threading()
def test_threaded_weak_key_dict_copy(self):
# Issue #35615: Weakref keys or values getting GC'ed during dict
# copying should not result in a crash.
self.check_threaded_weak_dict_copy(weakref.WeakKeyDictionary, False)

@threading_helper.requires_working_threading()
@support.requires_resource('cpu')
def test_threaded_weak_key_dict_deepcopy(self):
# Issue #35615: Weakref keys or values getting GC'ed during dict
# copying should not result in a crash.
self.check_threaded_weak_dict_copy(weakref.WeakKeyDictionary, True)

@unittest.skip("TODO: RUSTPYTHON; occasionally crash (Exit code -6)")
@threading_helper.requires_working_threading()
def test_threaded_weak_value_dict_copy(self):
# Issue #35615: Weakref keys or values getting GC'ed during dict
# copying should not result in a crash.
self.check_threaded_weak_dict_copy(weakref.WeakValueDictionary, False)

@threading_helper.requires_working_threading()
@support.requires_resource('cpu')
def test_threaded_weak_value_dict_deepcopy(self):
# Issue #35615: Weakref keys or values getting GC'ed during dict
# copying should not result in a crash.
Expand Down Expand Up @@ -2195,6 +2293,19 @@ def test_atexit(self):
self.assertTrue(b'ZeroDivisionError' in err)


class ModuleTestCase(unittest.TestCase):
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_names(self):
for name in ('ReferenceType', 'ProxyType', 'CallableProxyType',
'WeakMethod', 'WeakSet', 'WeakKeyDictionary', 'WeakValueDictionary'):
obj = getattr(weakref, name)
if name != 'WeakSet':
self.assertEqual(obj.__module__, 'weakref')
self.assertEqual(obj.__name__, name)
self.assertEqual(obj.__qualname__, name)


libreftest = """ Doctest for examples in the library reference: weakref.rst

>>> from test.support import gc_collect
Expand Down Expand Up @@ -2283,19 +2394,11 @@ def test_atexit(self):

__test__ = {'libreftest' : libreftest}

def test_main():
support.run_unittest(
ReferencesTestCase,
WeakMethodTestCase,
MappingTestCase,
WeakValueDictionaryTestCase,
WeakKeyDictionaryTestCase,
SubclassableWeakrefTestCase,
FinalizeTestCase,
)
def load_tests(loader, tests, pattern):
# TODO: RUSTPYTHON
# support.run_doctest(sys.modules[__name__])
# tests.addTest(doctest.DocTestSuite())
return tests


if __name__ == "__main__":
test_main()
unittest.main()
Loading