pypy.interpreter.test.apptest_with_leak:test_shutil_pattern
def test_shutil_pattern():
# Reproduce the shutil.copyfile failure.
# Multiple early returns inside the inner with cause duplicate_exits_without_lineno
# to copy the with-exit code, which confuses the linear exception table scan and
# leaves the fallback path (copyfileobj) with no exception table coverage.
# Structure mirrors shutil.copyfile:
# outer with (suppresses) -> try/except -> inner with (raises in __exit__)
# -> fast paths with early returns -> fallback that raises via None return
class InnerCM:
def __enter__(self):
return None # intentional -- fallback body raises AttributeError
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is not None:
raise OSError("dest close failed")
return False
class OuterCM:
exited_with = 'not called'
def __enter__(self): return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.exited_with = exc_type
return True # suppress everything
class GiveupOnFastCopy(Exception):
pass
def copyfile(use_fast1, use_fast2):
result = []
outer = OuterCM()
with outer:
try:
with InnerCM() as f:
if use_fast1:
try:
result.append('fast1')
return result
except GiveupOnFastCopy:
pass
elif use_fast2:
try:
result.append('fast2')
return result
except GiveupOnFastCopy:
pass
# fallback: f is None, raises AttributeError
result.append(f.read)
except IsADirectoryError:
pass
return outer
import dis
> dis.dis(copyfile)
../build/pypy/interpreter/test/apptest_with_leak.py:131:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../build/lib-python/3/dis.py:114: in dis
_disassemble_recursive(x, file=file, depth=depth, show_caches=show_caches, adaptive=adaptive)
../build/lib-python/3/dis.py:555: in _disassemble_recursive
disassemble(co, file=file, show_caches=show_caches, adaptive=adaptive)
../build/lib-python/3/dis.py:548: in disassemble
_disassemble_bytes(_get_code_array(co, adaptive),
../build/lib-python/3/dis.py:586: in _disassemble_bytes
for instr in _get_instructions_bytes(code, varname_from_oparg, names,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
code = b"g\x00}\x02\x89\x02\x83\x00}\x03|\x035\x00\x01\x00\t\x00\x89\x01\x83\x005\x00}\x04|\x00r/\t\x00|\x02\xa0\x00d\x01\xa1...\x00\x08\x00\x08\x00\x83\x03\x01\x00\x99\x86#\x001\x00s\x82w\x02&\x03Y\x00w\x01\x01\x00Y\x00\x01\x00\x01\x00|\x03S\x00"
varname_from_oparg = <bound method code._varname_from_oparg of <code object copyfile at 0x00007f43ae93bd00, file "/build_dir/pypy-c-jit-linux-x86-64/build/pypy/interpreter/test/apptest_with_leak.py", line 106>>
names = ('append', 'read', 'IsADirectoryError')
co_consts = (None, 'fast1', 'fast2')
linestarts = {0: 107, 4: 108, 10: 109, 16: 110, ...}, line_offset = 0
exception_entries = [_ExceptionTableEntry(start=14, end=18, target=246, depth=1, lasti=True), _ExceptionTableEntry(start=18, end=24, targe...end=56, target=210, depth=1, lasti=False), _ExceptionTableEntry(start=70, end=80, target=88, depth=3, lasti=True), ...]
co_positions = <sequenceiterator object at 0x0000000002483898>
show_caches = False
def _get_instructions_bytes(code, varname_from_oparg=None,
names=None, co_consts=None,
linestarts=None, line_offset=0,
exception_entries=(), co_positions=None,
show_caches=False):
"""Iterate over the instructions in a bytecode string.
Generates a sequence of Instruction namedtuples giving the details of each
opcode. Additional information about the code's runtime environment
(e.g. variable names, co_consts) can be specified using optional
arguments.
"""
co_positions = co_positions or iter(())
get_name = None if names is None else names.__getitem__
labels = set(findlabels(code))
for start, end, target, _, _ in exception_entries:
for i in range(start, end):
labels.add(target)
starts_line = None
for offset, op, arg in _unpack_opargs(code):
if linestarts is not None:
starts_line = linestarts.get(offset, None)
if starts_line is not None:
starts_line += line_offset
is_jump_target = offset in labels
argval = None
argrepr = ''
positions = Positions(*next(co_positions, ()))
deop = _deoptop(op)
caches = _inline_cache_entries[deop]
if arg is not None:
# Set argval to the dereferenced value of the argument when
# available, and argrepr to the string representation of argval.
# _disassemble_bytes needs the string repr of the
# raw name index for LOAD_GLOBAL, LOAD_CONST, etc.
argval = arg
if deop in hasconst:
argval, argrepr = _get_const_info(deop, arg, co_consts)
elif deop in hasname:
if deop == LOAD_GLOBAL:
argval, argrepr = _get_name_info(arg//2, get_name)
if (arg & 1) and argrepr:
argrepr = "NULL + " + argrepr
elif deop == LOAD_ATTR:
argval, argrepr = _get_name_info(arg//2, get_name)
if (arg & 1) and argrepr:
argrepr = "NULL|self + " + argrepr
elif deop == LOAD_SUPER_ATTR:
argval, argrepr = _get_name_info(arg//4, get_name)
if (arg & 1) and argrepr:
argrepr = "NULL|self + " + argrepr
else:
argval, argrepr = _get_name_info(arg, get_name)
elif deop in hasjabs:
argval = arg*2
argrepr = "to " + repr(argval)
elif deop in hasjrel:
signed_arg = -arg if _is_backward_jump(deop) else arg
argval = offset + 2 + signed_arg*2
argval += 2 * caches
argrepr = "to " + repr(argval)
elif deop in haslocal or deop in hasfree:
argval, argrepr = _get_name_info(arg, varname_from_oparg)
elif deop in hascompare:
argval = cmp_op[arg>>4]
argrepr = argval
elif deop == FORMAT_VALUE:
argval, argrepr = FORMAT_VALUE_CONVERTERS[arg & 0x3]
argval = (argval, bool(arg & 0x4))
if argval[1]:
if argrepr:
argrepr += ', '
argrepr += 'with format'
elif deop == MAKE_FUNCTION:
argrepr = ', '.join(s for i, s in enumerate(MAKE_FUNCTION_FLAGS)
if arg & (1<<i))
> elif deop == BINARY_OP:
E NameError: name 'BINARY_OP' is not defined
../build/lib-python/3/dis.py:512: NameError
builder: pypy-c-jit-linux-x86-64 build #11848+
test: pypy.interpreter.test.apptest_with_leak:test_shutil_pattern