mirror of
https://github.com/python/cpython.git
synced 2026-10-28 18:35:21 +08:00
Serhiy Storchaka
cdca5021d5
Test setting a value of an accepted type, of a wrong type and an invalid value, and deleting the attribute, for the attributes defined with PyMemberDef and PyGetSetDef which were not covered.
977 lines
29 KiB
Python
977 lines
29 KiB
Python
import copy
|
|
import operator
|
|
import re
|
|
import sys
|
|
import textwrap
|
|
import threading
|
|
import unittest
|
|
import weakref
|
|
try:
|
|
import _testcapi
|
|
except ImportError:
|
|
_testcapi = None
|
|
|
|
from collections.abc import Mapping
|
|
from test import support
|
|
from test.support import import_helper, threading_helper
|
|
from test.support.script_helper import assert_python_ok
|
|
from test import mapping_tests
|
|
|
|
|
|
class ClearTest(unittest.TestCase):
|
|
"""
|
|
Tests for frame.clear().
|
|
"""
|
|
|
|
def inner(self, x=5, **kwargs):
|
|
1/0
|
|
|
|
def outer(self, **kwargs):
|
|
try:
|
|
self.inner(**kwargs)
|
|
except ZeroDivisionError as e:
|
|
exc = e
|
|
return exc
|
|
|
|
def clear_traceback_frames(self, tb):
|
|
"""
|
|
Clear all frames in a traceback.
|
|
"""
|
|
while tb is not None:
|
|
tb.tb_frame.clear()
|
|
tb = tb.tb_next
|
|
|
|
def test_clear_locals(self):
|
|
class C:
|
|
pass
|
|
c = C()
|
|
wr = weakref.ref(c)
|
|
exc = self.outer(c=c)
|
|
del c
|
|
support.gc_collect()
|
|
# A reference to c is held through the frames
|
|
self.assertIsNot(None, wr())
|
|
self.clear_traceback_frames(exc.__traceback__)
|
|
support.gc_collect()
|
|
# The reference was released by .clear()
|
|
self.assertIs(None, wr())
|
|
|
|
def test_clear_locals_after_f_locals_access(self):
|
|
# see gh-113939
|
|
class C:
|
|
pass
|
|
|
|
wr = None
|
|
def inner():
|
|
nonlocal wr
|
|
c = C()
|
|
wr = weakref.ref(c)
|
|
1/0
|
|
|
|
try:
|
|
inner()
|
|
except ZeroDivisionError as exc:
|
|
support.gc_collect()
|
|
self.assertIsNotNone(wr())
|
|
exc.__traceback__.tb_next.tb_frame.clear()
|
|
support.gc_collect()
|
|
self.assertIsNone(wr())
|
|
|
|
def test_clear_does_not_clear_specials(self):
|
|
class C:
|
|
pass
|
|
c = C()
|
|
exc = self.outer(c=c)
|
|
del c
|
|
f = exc.__traceback__.tb_frame
|
|
f.clear()
|
|
self.assertIsNot(f.f_code, None)
|
|
self.assertIsNot(f.f_locals, None)
|
|
self.assertIsNot(f.f_builtins, None)
|
|
self.assertIsNot(f.f_globals, None)
|
|
|
|
def test_clear_generator(self):
|
|
endly = False
|
|
def g():
|
|
nonlocal endly
|
|
try:
|
|
yield
|
|
self.inner()
|
|
finally:
|
|
endly = True
|
|
gen = g()
|
|
next(gen)
|
|
self.assertFalse(endly)
|
|
|
|
# Cannot clear a suspended frame
|
|
with self.assertRaisesRegex(RuntimeError, r'suspended frame'):
|
|
gen.gi_frame.clear()
|
|
self.assertFalse(endly)
|
|
|
|
def test_clear_executing(self):
|
|
# Attempting to clear an executing frame is forbidden.
|
|
try:
|
|
1/0
|
|
except ZeroDivisionError as e:
|
|
f = e.__traceback__.tb_frame
|
|
with self.assertRaises(RuntimeError):
|
|
f.clear()
|
|
with self.assertRaises(RuntimeError):
|
|
f.f_back.clear()
|
|
|
|
def test_clear_executing_generator(self):
|
|
# Attempting to clear an executing generator frame is forbidden.
|
|
endly = False
|
|
def g():
|
|
nonlocal endly
|
|
try:
|
|
1/0
|
|
except ZeroDivisionError as e:
|
|
f = e.__traceback__.tb_frame
|
|
with self.assertRaises(RuntimeError):
|
|
f.clear()
|
|
with self.assertRaises(RuntimeError):
|
|
f.f_back.clear()
|
|
yield f
|
|
finally:
|
|
endly = True
|
|
gen = g()
|
|
f = next(gen)
|
|
self.assertFalse(endly)
|
|
# Cannot clear a suspended frame
|
|
with self.assertRaisesRegex(RuntimeError, 'suspended frame'):
|
|
f.clear()
|
|
self.assertFalse(endly)
|
|
|
|
def test_lineno_with_tracing(self):
|
|
def record_line():
|
|
f = sys._getframe(1)
|
|
lines.append(f.f_lineno-f.f_code.co_firstlineno)
|
|
|
|
def test(trace):
|
|
record_line()
|
|
if trace:
|
|
sys._getframe(0).f_trace = True
|
|
record_line()
|
|
record_line()
|
|
|
|
expected_lines = [1, 4, 5]
|
|
lines = []
|
|
test(False)
|
|
self.assertEqual(lines, expected_lines)
|
|
lines = []
|
|
test(True)
|
|
self.assertEqual(lines, expected_lines)
|
|
|
|
@support.cpython_only
|
|
def test_clear_refcycles(self):
|
|
# .clear() doesn't leave any refcycle behind
|
|
with support.disable_gc():
|
|
class C:
|
|
pass
|
|
c = C()
|
|
wr = weakref.ref(c)
|
|
exc = self.outer(c=c)
|
|
del c
|
|
self.assertIsNot(None, wr())
|
|
self.clear_traceback_frames(exc.__traceback__)
|
|
self.assertIs(None, wr())
|
|
|
|
|
|
class FrameAttrsTest(unittest.TestCase):
|
|
|
|
def make_frames(self):
|
|
def outer():
|
|
x = 5
|
|
y = 6
|
|
def inner():
|
|
z = x + 2
|
|
1/0
|
|
t = 9
|
|
return inner()
|
|
try:
|
|
outer()
|
|
except ZeroDivisionError as e:
|
|
tb = e.__traceback__
|
|
frames = []
|
|
while tb:
|
|
frames.append(tb.tb_frame)
|
|
tb = tb.tb_next
|
|
return frames
|
|
|
|
def test_clear_locals(self):
|
|
# Test f_locals after clear() (issue #21897)
|
|
f, outer, inner = self.make_frames()
|
|
outer.clear()
|
|
inner.clear()
|
|
self.assertEqual(outer.f_locals, {})
|
|
self.assertEqual(inner.f_locals, {})
|
|
|
|
def test_locals_clear_locals(self):
|
|
# Test f_locals before and after clear() (to exercise caching)
|
|
f, outer, inner = self.make_frames()
|
|
self.assertNotEqual(outer.f_locals, {})
|
|
self.assertNotEqual(inner.f_locals, {})
|
|
outer.clear()
|
|
inner.clear()
|
|
self.assertEqual(outer.f_locals, {})
|
|
self.assertEqual(inner.f_locals, {})
|
|
|
|
def test_f_lineno_del_segfault(self):
|
|
f, _, _ = self.make_frames()
|
|
with self.assertRaises(AttributeError):
|
|
del f.f_lineno
|
|
|
|
def test_f_trace(self):
|
|
f, _, _ = self.make_frames()
|
|
def tracer(*args):
|
|
pass
|
|
for value in tracer, 42, None:
|
|
f.f_trace = value
|
|
self.assertEqual(f.f_trace, value)
|
|
f.f_trace = tracer
|
|
del f.f_trace
|
|
self.assertIsNone(f.f_trace)
|
|
|
|
def test_f_trace_lines_and_opcodes(self):
|
|
f, _, _ = self.make_frames()
|
|
for name in 'f_trace_lines', 'f_trace_opcodes':
|
|
with self.subTest(name=name):
|
|
for value in False, True:
|
|
setattr(f, name, value)
|
|
self.assertEqual(getattr(f, name), value)
|
|
with self.assertRaisesRegex(TypeError,
|
|
'attribute value type must be bool'):
|
|
setattr(f, name, 1)
|
|
with self.assertRaisesRegex(TypeError,
|
|
"can't delete numeric/char attribute"):
|
|
del f.f_trace_lines
|
|
with self.assertRaisesRegex(AttributeError, 'cannot be deleted'):
|
|
del f.f_trace_opcodes
|
|
|
|
def test_f_generator(self):
|
|
# Test f_generator in different contexts.
|
|
|
|
def t0():
|
|
def nested():
|
|
frame = sys._getframe()
|
|
return frame.f_generator
|
|
|
|
def gen():
|
|
yield nested()
|
|
|
|
g = gen()
|
|
try:
|
|
return next(g)
|
|
finally:
|
|
g.close()
|
|
|
|
def t1():
|
|
frame = sys._getframe()
|
|
return frame.f_generator
|
|
|
|
def t2():
|
|
frame = sys._getframe()
|
|
yield frame.f_generator
|
|
|
|
async def t3():
|
|
frame = sys._getframe()
|
|
return frame.f_generator
|
|
|
|
# For regular functions f_generator is None
|
|
self.assertIsNone(t0())
|
|
self.assertIsNone(t1())
|
|
|
|
# For generators f_generator is equal to self
|
|
g = t2()
|
|
try:
|
|
frame_g = next(g)
|
|
self.assertIs(g, frame_g)
|
|
finally:
|
|
g.close()
|
|
|
|
# Ditto for coroutines
|
|
c = t3()
|
|
try:
|
|
c.send(None)
|
|
except StopIteration as ex:
|
|
self.assertIs(ex.value, c)
|
|
else:
|
|
raise AssertionError('coroutine did not exit')
|
|
|
|
|
|
class WeakRefTest(unittest.TestCase):
|
|
"""
|
|
Frames support weak references (gh-102960).
|
|
"""
|
|
|
|
def make_frame(self):
|
|
# Return the frame object of a finished function call. Unlike
|
|
# frames extracted from a traceback, it isn't part of a reference
|
|
# cycle, so it dies as soon as the last reference is dropped.
|
|
def func():
|
|
return sys._getframe()
|
|
return func()
|
|
|
|
def make_traceback_frames(self):
|
|
def outer():
|
|
def inner():
|
|
1/0
|
|
return inner()
|
|
try:
|
|
outer()
|
|
except ZeroDivisionError as e:
|
|
tb = e.__traceback__
|
|
frames = []
|
|
while tb:
|
|
frames.append(tb.tb_frame)
|
|
tb = tb.tb_next
|
|
return frames
|
|
|
|
def test_weakref_basic(self):
|
|
called = []
|
|
f = self.make_frame()
|
|
ref = weakref.ref(f)
|
|
cb_ref = weakref.ref(f, called.append)
|
|
self.assertIs(ref(), f)
|
|
self.assertIs(cb_ref(), f)
|
|
del f
|
|
support.gc_collect()
|
|
self.assertIsNone(ref())
|
|
self.assertIsNone(cb_ref())
|
|
self.assertEqual(called, [cb_ref])
|
|
|
|
@support.thread_unsafe("relies on gc.collect() reclaiming its cycles")
|
|
def test_weakref_live_frame(self):
|
|
refs = []
|
|
def func():
|
|
frame = sys._getframe()
|
|
refs.append(weakref.ref(frame))
|
|
self.assertIs(refs[0](), frame)
|
|
func()
|
|
support.gc_collect()
|
|
self.assertIsNone(refs[0]())
|
|
|
|
@support.thread_unsafe("relies on gc.collect() reclaiming its cycles")
|
|
def test_weak_key_dictionary(self):
|
|
wkd = weakref.WeakKeyDictionary()
|
|
def _fill():
|
|
for i, frame in enumerate(self.make_traceback_frames()):
|
|
wkd[frame] = i
|
|
self.assertEqual(len(wkd), 3)
|
|
_fill()
|
|
support.gc_collect()
|
|
self.assertEqual(len(wkd), 0)
|
|
|
|
@support.thread_unsafe("relies on gc.collect() reclaiming its cycles")
|
|
def test_weakref_traceback_frames(self):
|
|
# Frames that participate in reference cycles are cleaned up
|
|
# by the cyclic garbage collector.
|
|
refs = []
|
|
def _make():
|
|
for frame in self.make_traceback_frames():
|
|
refs.append(weakref.ref(frame))
|
|
for ref in refs:
|
|
self.assertIsNotNone(ref())
|
|
_make()
|
|
support.gc_collect()
|
|
for ref in refs:
|
|
self.assertIsNone(ref())
|
|
|
|
def test_weakref_generator_frame(self):
|
|
def gen():
|
|
yield sys._getframe()
|
|
g = gen()
|
|
frame = next(g)
|
|
ref = weakref.ref(frame)
|
|
del frame
|
|
support.gc_collect()
|
|
# The generator keeps its frame alive while suspended.
|
|
self.assertIsNotNone(ref())
|
|
g.close()
|
|
del g
|
|
support.gc_collect()
|
|
self.assertIsNone(ref())
|
|
|
|
def test_weakref_after_frame_clear(self):
|
|
f = self.make_frame()
|
|
ref = weakref.ref(f)
|
|
# Clearing the frame's contents must not affect weak references
|
|
# to the frame object itself.
|
|
f.clear()
|
|
self.assertIs(ref(), f)
|
|
del f
|
|
support.gc_collect()
|
|
self.assertIsNone(ref())
|
|
|
|
@threading_helper.requires_working_threading()
|
|
def test_weakref_concurrent(self):
|
|
# Exercise concurrent creation and destruction of weak references
|
|
# to the same frame, mainly for the free-threaded build.
|
|
def gen():
|
|
yield sys._getframe()
|
|
g = gen()
|
|
frame = next(g)
|
|
barrier = threading.Barrier(4)
|
|
# Collect failures instead of asserting in the workers: exceptions
|
|
# raised in threads don't propagate to the unittest result.
|
|
failures = []
|
|
def work():
|
|
barrier.wait()
|
|
for _ in range(1000):
|
|
ref = weakref.ref(frame)
|
|
if ref() is not frame:
|
|
failures.append('shared ref dead while frame alive')
|
|
# Callback refs are not shared, so this concurrently adds
|
|
# to and removes from the frame's weakref list.
|
|
cb_ref = weakref.ref(frame, lambda r: None)
|
|
if cb_ref() is not frame:
|
|
failures.append('callback ref dead while frame alive')
|
|
del ref, cb_ref
|
|
threads = [threading.Thread(target=work) for _ in range(4)]
|
|
with threading_helper.start_threads(threads):
|
|
pass
|
|
self.assertEqual(failures, [])
|
|
ref = weakref.ref(frame)
|
|
del frame
|
|
g.close()
|
|
del g
|
|
support.gc_collect()
|
|
self.assertIsNone(ref())
|
|
|
|
|
|
class ReprTest(unittest.TestCase):
|
|
"""
|
|
Tests for repr(frame).
|
|
"""
|
|
|
|
def test_repr(self):
|
|
def outer():
|
|
x = 5
|
|
y = 6
|
|
def inner():
|
|
z = x + 2
|
|
1/0
|
|
t = 9
|
|
return inner()
|
|
|
|
offset = outer.__code__.co_firstlineno
|
|
try:
|
|
outer()
|
|
except ZeroDivisionError as e:
|
|
tb = e.__traceback__
|
|
frames = []
|
|
while tb:
|
|
frames.append(tb.tb_frame)
|
|
tb = tb.tb_next
|
|
else:
|
|
self.fail("should have raised")
|
|
|
|
f_this, f_outer, f_inner = frames
|
|
file_repr = re.escape(repr(__file__))
|
|
self.assertRegex(repr(f_this),
|
|
r"^<frame at 0x[0-9a-fA-F]+, file %s, line %d, code test_repr>$"
|
|
% (file_repr, offset + 23))
|
|
self.assertRegex(repr(f_outer),
|
|
r"^<frame at 0x[0-9a-fA-F]+, file %s, line %d, code outer>$"
|
|
% (file_repr, offset + 7))
|
|
self.assertRegex(repr(f_inner),
|
|
r"^<frame at 0x[0-9a-fA-F]+, file %s, line %d, code inner>$"
|
|
% (file_repr, offset + 5))
|
|
|
|
class TestFrameLocals(unittest.TestCase):
|
|
def test_scope(self):
|
|
class A:
|
|
x = 1
|
|
sys._getframe().f_locals['x'] = 2
|
|
sys._getframe().f_locals['y'] = 2
|
|
|
|
self.assertEqual(A.x, 2)
|
|
self.assertEqual(A.y, 2)
|
|
|
|
def f():
|
|
x = 1
|
|
sys._getframe().f_locals['x'] = 2
|
|
sys._getframe().f_locals['y'] = 2
|
|
self.assertEqual(x, 2)
|
|
self.assertEqual(locals()['y'], 2)
|
|
f()
|
|
|
|
def test_closure(self):
|
|
x = 1
|
|
y = 2
|
|
|
|
def f():
|
|
z = x + y
|
|
d = sys._getframe().f_locals
|
|
self.assertEqual(d['x'], 1)
|
|
self.assertEqual(d['y'], 2)
|
|
d['x'] = 2
|
|
d['y'] = 3
|
|
|
|
f()
|
|
self.assertEqual(x, 2)
|
|
self.assertEqual(y, 3)
|
|
|
|
def test_closure_with_inline_comprehension(self):
|
|
lambda: k
|
|
k = 1
|
|
lst = [locals() for k in [0]]
|
|
self.assertEqual(lst[0]['k'], 0)
|
|
|
|
def test_as_dict(self):
|
|
x = 1
|
|
y = 2
|
|
d = sys._getframe().f_locals
|
|
# self, x, y, d
|
|
self.assertEqual(len(d), 4)
|
|
self.assertIs(d['d'], d)
|
|
self.assertEqual(set(d.keys()), set(['x', 'y', 'd', 'self']))
|
|
self.assertEqual(len(d.values()), 4)
|
|
self.assertIn(1, d.values())
|
|
self.assertEqual(len(d.items()), 4)
|
|
self.assertIn(('x', 1), d.items())
|
|
self.assertEqual(d.__getitem__('x'), 1)
|
|
d.__setitem__('x', 2)
|
|
self.assertEqual(d['x'], 2)
|
|
self.assertEqual(d.get('x'), 2)
|
|
self.assertIs(d.get('non_exist', None), None)
|
|
self.assertEqual(d.__len__(), 4)
|
|
self.assertEqual(set([key for key in d]), set(['x', 'y', 'd', 'self']))
|
|
self.assertIn('x', d)
|
|
self.assertTrue(d.__contains__('x'))
|
|
|
|
self.assertEqual(reversed(d), list(reversed(d.keys())))
|
|
|
|
d.update({'x': 3, 'z': 4})
|
|
self.assertEqual(d['x'], 3)
|
|
self.assertEqual(d['z'], 4)
|
|
|
|
with self.assertRaises(TypeError):
|
|
d.update([1, 2])
|
|
|
|
self.assertEqual(d.setdefault('x', 5), 3)
|
|
self.assertEqual(d.setdefault('new', 5), 5)
|
|
self.assertEqual(d['new'], 5)
|
|
|
|
with self.assertRaises(KeyError):
|
|
d['non_exist']
|
|
|
|
def test_as_number(self):
|
|
x = 1
|
|
y = 2
|
|
d = sys._getframe().f_locals
|
|
self.assertIn('z', d | {'z': 3})
|
|
d |= {'z': 3}
|
|
self.assertEqual(d['z'], 3)
|
|
d |= {'y': 3}
|
|
self.assertEqual(d['y'], 3)
|
|
with self.assertRaises(TypeError):
|
|
d |= 3
|
|
with self.assertRaises(TypeError):
|
|
_ = d | [3]
|
|
|
|
def test_non_string_key(self):
|
|
d = sys._getframe().f_locals
|
|
d[1] = 2
|
|
self.assertEqual(d[1], 2)
|
|
|
|
def test_write_with_hidden(self):
|
|
def f():
|
|
f_locals = [sys._getframe().f_locals for b in [0]][0]
|
|
f_locals['b'] = 2
|
|
f_locals['c'] = 3
|
|
self.assertEqual(b, 2)
|
|
self.assertEqual(c, 3)
|
|
b = 0
|
|
c = 0
|
|
f()
|
|
|
|
def test_local_objects(self):
|
|
o = object()
|
|
k = '.'.join(['a', 'b', 'c'])
|
|
f_locals = sys._getframe().f_locals
|
|
f_locals['o'] = f_locals['k']
|
|
self.assertEqual(o, 'a.b.c')
|
|
|
|
def test_copy(self):
|
|
x = 0
|
|
d = sys._getframe().f_locals
|
|
d_copy = d.copy()
|
|
self.assertIsInstance(d_copy, dict)
|
|
self.assertEqual(d_copy['x'], 0)
|
|
d_copy['x'] = 1
|
|
self.assertEqual(x, 0)
|
|
|
|
def test_update_with_self(self):
|
|
def f():
|
|
f_locals = sys._getframe().f_locals
|
|
f_locals.update(f_locals)
|
|
f_locals.update(f_locals)
|
|
f_locals.update(f_locals)
|
|
f()
|
|
|
|
def test_repr(self):
|
|
x = 1
|
|
# Introduce a reference cycle
|
|
frame = sys._getframe()
|
|
self.assertEqual(repr(frame.f_locals), repr(dict(frame.f_locals)))
|
|
|
|
def test_delete(self):
|
|
x = 1
|
|
d = sys._getframe().f_locals
|
|
|
|
# This needs to be tested before f_extra_locals is created
|
|
with self.assertRaisesRegex(KeyError, 'non_exist'):
|
|
del d['non_exist']
|
|
|
|
with self.assertRaises(KeyError):
|
|
d.pop('non_exist')
|
|
|
|
with self.assertRaisesRegex(ValueError, 'local variables'):
|
|
del d['x']
|
|
|
|
with self.assertRaises(AttributeError):
|
|
d.clear()
|
|
|
|
with self.assertRaises(ValueError):
|
|
d.pop('x')
|
|
|
|
with self.assertRaises(ValueError):
|
|
d.pop('x', None)
|
|
|
|
# 'm', 'n' is stored in f_extra_locals
|
|
d['m'] = 1
|
|
d['n'] = 1
|
|
|
|
with self.assertRaises(KeyError):
|
|
d.pop('non_exist')
|
|
|
|
del d['m']
|
|
self.assertEqual(d.pop('n'), 1)
|
|
|
|
self.assertNotIn('m', d)
|
|
self.assertNotIn('n', d)
|
|
|
|
self.assertEqual(d.pop('n', 2), 2)
|
|
|
|
@support.cpython_only
|
|
def test_sizeof(self):
|
|
proxy = sys._getframe().f_locals
|
|
support.check_sizeof(self, proxy, support.calcobjsize("P"))
|
|
|
|
def test_unsupport(self):
|
|
x = 1
|
|
d = sys._getframe().f_locals
|
|
with self.assertRaises(TypeError):
|
|
copy.copy(d)
|
|
|
|
with self.assertRaises(TypeError):
|
|
copy.deepcopy(d)
|
|
|
|
def test_is_mapping(self):
|
|
x = 1
|
|
d = sys._getframe().f_locals
|
|
self.assertIsInstance(d, Mapping)
|
|
match d:
|
|
case {"x": value}:
|
|
self.assertEqual(value, 1)
|
|
kind = "mapping"
|
|
case _:
|
|
kind = "other"
|
|
self.assertEqual(kind, "mapping")
|
|
|
|
def _x_stringlikes(self):
|
|
class StringSubclass(str):
|
|
pass
|
|
|
|
class ImpostorX:
|
|
def __hash__(self):
|
|
return hash('x')
|
|
|
|
def __eq__(self, other):
|
|
return other == 'x'
|
|
|
|
return StringSubclass('x'), ImpostorX(), 'x'
|
|
|
|
def test_proxy_key_stringlikes_overwrite(self):
|
|
def f(obj):
|
|
x = 1
|
|
proxy = sys._getframe().f_locals
|
|
proxy[obj] = 2
|
|
return (
|
|
list(proxy.keys()),
|
|
dict(proxy),
|
|
proxy
|
|
)
|
|
|
|
for obj in self._x_stringlikes():
|
|
with self.subTest(cls=type(obj).__name__):
|
|
|
|
keys_snapshot, proxy_snapshot, proxy = f(obj)
|
|
expected_keys = ['obj', 'x', 'proxy']
|
|
expected_dict = {'obj': 'x', 'x': 2, 'proxy': proxy}
|
|
self.assertEqual(proxy.keys(), expected_keys)
|
|
self.assertEqual(proxy, expected_dict)
|
|
self.assertEqual(keys_snapshot, expected_keys)
|
|
self.assertEqual(proxy_snapshot, expected_dict)
|
|
|
|
def test_proxy_key_stringlikes_ftrst_write(self):
|
|
def f(obj):
|
|
proxy = sys._getframe().f_locals
|
|
proxy[obj] = 2
|
|
self.assertEqual(x, 2)
|
|
x = 1
|
|
|
|
for obj in self._x_stringlikes():
|
|
with self.subTest(cls=type(obj).__name__):
|
|
f(obj)
|
|
|
|
def test_proxy_key_unhashables(self):
|
|
class StringSubclass(str):
|
|
__hash__ = None
|
|
|
|
class ObjectSubclass:
|
|
__hash__ = None
|
|
|
|
proxy = sys._getframe().f_locals
|
|
|
|
for obj in StringSubclass('x'), ObjectSubclass():
|
|
with self.subTest(cls=type(obj).__name__):
|
|
with self.assertRaises(TypeError):
|
|
proxy[obj]
|
|
with self.assertRaises(TypeError):
|
|
proxy[obj] = 0
|
|
|
|
def test_constructor(self):
|
|
FrameLocalsProxy = type([sys._getframe().f_locals
|
|
for x in range(1)][0])
|
|
self.assertEqual(FrameLocalsProxy.__name__, 'FrameLocalsProxy')
|
|
|
|
def make_frame():
|
|
x = 1
|
|
y = 2
|
|
return sys._getframe()
|
|
|
|
proxy = FrameLocalsProxy(make_frame())
|
|
self.assertEqual(proxy, {'x': 1, 'y': 2})
|
|
|
|
# constructor expects 1 frame argument
|
|
with self.assertRaises(TypeError):
|
|
FrameLocalsProxy() # no arguments
|
|
with self.assertRaises(TypeError):
|
|
FrameLocalsProxy(123) # wrong type
|
|
with self.assertRaises(TypeError):
|
|
FrameLocalsProxy(frame=sys._getframe()) # no keyword arguments
|
|
|
|
def test_overwrite_locals(self):
|
|
# Verify we do not crash if we overwrite a local passed as an argument
|
|
# from an ancestor in the call stack.
|
|
def f():
|
|
xs = [1, 2, 3]
|
|
ys = [4, 5, 6]
|
|
return g(xs)
|
|
|
|
def g(xs):
|
|
f = sys._getframe()
|
|
f.f_back.f_locals["xs"] = None
|
|
f.f_back.f_locals["ys"] = None
|
|
return xs[1]
|
|
|
|
self.assertEqual(f(), 2)
|
|
|
|
|
|
class FrameLocalsProxyMappingTests(mapping_tests.TestHashMappingProtocol):
|
|
"""Test that FrameLocalsProxy behaves like a Mapping (with exceptions)"""
|
|
|
|
def _f(*args, **kwargs):
|
|
def _f():
|
|
return sys._getframe().f_locals
|
|
return _f()
|
|
type2test = _f
|
|
|
|
@unittest.skipIf(True, 'Locals proxies for different frames never compare as equal')
|
|
def test_constructor(self):
|
|
pass
|
|
|
|
@unittest.skipIf(True, 'Unlike a mapping: del proxy[key] fails')
|
|
def test_write(self):
|
|
pass
|
|
|
|
@unittest.skipIf(True, 'Unlike a mapping: no proxy.popitem')
|
|
def test_popitem(self):
|
|
pass
|
|
|
|
@unittest.skipIf(True, 'Unlike a mapping: no proxy.pop')
|
|
def test_pop(self):
|
|
pass
|
|
|
|
@unittest.skipIf(True, 'Unlike a mapping: no proxy.clear')
|
|
def test_clear(self):
|
|
pass
|
|
|
|
@unittest.skipIf(True, 'Unlike a mapping: no proxy.fromkeys')
|
|
def test_fromkeys(self):
|
|
pass
|
|
|
|
# no del
|
|
def test_getitem(self):
|
|
mapping_tests.BasicTestMappingProtocol.test_getitem(self)
|
|
d = self._full_mapping({'a': 1, 'b': 2})
|
|
self.assertEqual(d['a'], 1)
|
|
self.assertEqual(d['b'], 2)
|
|
d['c'] = 3
|
|
d['a'] = 4
|
|
self.assertEqual(d['c'], 3)
|
|
self.assertEqual(d['a'], 4)
|
|
|
|
@unittest.skipIf(True, 'Unlike a mapping: no proxy.update')
|
|
def test_update(self):
|
|
pass
|
|
|
|
# proxy.copy returns a regular dict
|
|
def test_copy(self):
|
|
d = self._full_mapping({1:1, 2:2, 3:3})
|
|
self.assertEqual(d.copy(), {1:1, 2:2, 3:3})
|
|
d = self._empty_mapping()
|
|
self.assertEqual(d.copy(), d)
|
|
self.assertRaises(TypeError, d.copy, None)
|
|
|
|
self.assertIsInstance(d.copy(), dict)
|
|
|
|
@unittest.skipIf(True, 'Locals proxies for different frames never compare as equal')
|
|
def test_eq(self):
|
|
pass
|
|
|
|
|
|
class TestFrameCApi(unittest.TestCase):
|
|
def test_basic(self):
|
|
x = 1
|
|
ctypes = import_helper.import_module('ctypes')
|
|
import ctypes.util # noqa: F811
|
|
|
|
@ctypes.util.wrap_dll_function(ctypes.pythonapi)
|
|
def PyEval_GetFrameLocals() -> ctypes.py_object:
|
|
pass
|
|
|
|
@ctypes.util.wrap_dll_function(ctypes.pythonapi)
|
|
def PyEval_GetFrameGlobals() -> ctypes.py_object:
|
|
pass
|
|
|
|
@ctypes.util.wrap_dll_function(ctypes.pythonapi)
|
|
def PyEval_GetFrameBuiltins() -> ctypes.py_object:
|
|
pass
|
|
|
|
@ctypes.util.wrap_dll_function(ctypes.pythonapi)
|
|
def PyFrame_GetLocals(frame: ctypes.py_object) -> ctypes.py_object:
|
|
pass
|
|
|
|
frame_locals = PyEval_GetFrameLocals()
|
|
self.assertTrue(type(frame_locals), dict)
|
|
self.assertEqual(frame_locals['x'], 1)
|
|
frame_locals['x'] = 2
|
|
self.assertEqual(x, 1)
|
|
|
|
frame_globals = PyEval_GetFrameGlobals()
|
|
self.assertTrue(type(frame_globals), dict)
|
|
self.assertIs(frame_globals, globals())
|
|
|
|
frame_builtins = PyEval_GetFrameBuiltins()
|
|
self.assertEqual(frame_builtins, __builtins__)
|
|
|
|
frame = sys._getframe()
|
|
f_locals = PyFrame_GetLocals(frame)
|
|
self.assertTrue(f_locals['x'], 1)
|
|
f_locals['x'] = 2
|
|
self.assertEqual(x, 2)
|
|
|
|
|
|
class TestIncompleteFrameAreInvisible(unittest.TestCase):
|
|
|
|
def test_issue95818(self):
|
|
# See GH-95818 for details
|
|
code = textwrap.dedent(f"""
|
|
import gc
|
|
|
|
gc.set_threshold(1,1,1)
|
|
class GCHello:
|
|
def __del__(self):
|
|
print("Destroyed from gc")
|
|
|
|
def gen():
|
|
yield
|
|
|
|
fd = open({__file__!r})
|
|
l = [fd, GCHello()]
|
|
l.append(l)
|
|
del fd
|
|
del l
|
|
gen()
|
|
""")
|
|
assert_python_ok("-c", code)
|
|
|
|
@support.cpython_only
|
|
@threading_helper.requires_working_threading()
|
|
def test_sneaky_frame_object_teardown(self):
|
|
|
|
class SneakyDel:
|
|
def __del__(self):
|
|
"""
|
|
Stash a reference to the entire stack for walking later.
|
|
|
|
It may look crazy, but you'd be surprised how common this is
|
|
when using a test runner (like pytest). The typical recipe is:
|
|
ResourceWarning + -Werror + a custom sys.unraisablehook.
|
|
"""
|
|
nonlocal sneaky_frame_object
|
|
sneaky_frame_object = sys._getframe()
|
|
|
|
class SneakyThread(threading.Thread):
|
|
"""
|
|
A separate thread isn't needed to make this code crash, but it does
|
|
make crashes more consistent, since it means sneaky_frame_object is
|
|
backed by freed memory after the thread completes!
|
|
"""
|
|
|
|
def run(self):
|
|
"""Run SneakyDel.__del__ as this frame is popped."""
|
|
ref = SneakyDel()
|
|
|
|
sneaky_frame_object = None
|
|
t = SneakyThread()
|
|
t.start()
|
|
t.join()
|
|
# sneaky_frame_object can be anything, really, but it's crucial that
|
|
# SneakyThread.run's frame isn't anywhere on the stack while it's being
|
|
# torn down:
|
|
self.assertIsNotNone(sneaky_frame_object)
|
|
while sneaky_frame_object is not None:
|
|
self.assertIsNot(
|
|
sneaky_frame_object.f_code, SneakyThread.run.__code__
|
|
)
|
|
sneaky_frame_object = sneaky_frame_object.f_back
|
|
|
|
def test_entry_frames_are_invisible_during_teardown(self):
|
|
class C:
|
|
"""A weakref'able class."""
|
|
|
|
def f():
|
|
"""Try to find globals and locals as this frame is being cleared."""
|
|
ref = C()
|
|
# Ignore the fact that exec(C()) is a nonsense callback. We're only
|
|
# using exec here because it tries to access the current frame's
|
|
# globals and locals. If it's trying to get those from a shim frame,
|
|
# we'll crash before raising:
|
|
return weakref.ref(ref, exec)
|
|
|
|
with support.catch_unraisable_exception() as catcher:
|
|
# Call from C, so there is a shim frame directly above f:
|
|
weak = operator.call(f) # BOOM!
|
|
# Cool, we didn't crash. Check that the callback actually happened:
|
|
self.assertIs(catcher.unraisable.exc_type, TypeError)
|
|
self.assertIsNone(weak())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|