Add unit tests for calculator
This commit is contained in:
1 parent
05bfcd65df
commit
2e854cbeb6
2227 files changed
+408232
-1
No files matched your search
@@ -0,0 +1,255 @@
|
||||
# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
|
||||
# For details: https://github.com/coveragepy/coveragepy/blob/main/NOTICE.txt
|
||||
|
||||
"""Bytecode analysis for coverage.py"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import collections
|
||||
import dis
|
||||
from collections.abc import Iterable, Mapping
|
||||
from types import CodeType
|
||||
from typing import Optional
|
||||
|
||||
from coverage.types import TArc, TLineNo, TOffset
|
||||
|
||||
|
||||
class ByteParser:
|
||||
"""Parse bytecode to understand the structure of code."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
code: CodeType | None = None,
|
||||
text: str | None = None,
|
||||
filename: str | None = None,
|
||||
) -> None:
|
||||
if code is None:
|
||||
assert text is not None
|
||||
code = compile(text, filename or "<string>", "exec", dont_inherit=True)
|
||||
self.code = code
|
||||
|
||||
def _child_parsers(self) -> Iterable[ByteParser]:
|
||||
"""Iterate over all the code objects nested within this one.
|
||||
|
||||
The iteration includes `self` as its first value.
|
||||
|
||||
We skip code objects named `__annotate__` since they are deferred
|
||||
annotations that usually are never run. If there are errors in the
|
||||
annotations, they will be caught by type checkers or other tools that
|
||||
use annotations.
|
||||
|
||||
"""
|
||||
return (ByteParser(code=c) for c in self.code_objects() if c.co_name != "__annotate__")
|
||||
|
||||
def code_objects(self) -> Iterable[CodeType]:
|
||||
"""Iterate over all the code objects in `code`."""
|
||||
stack = [self.code]
|
||||
while stack:
|
||||
# We're going to return the code object on the stack, but first
|
||||
# push its children for later returning.
|
||||
code = stack.pop()
|
||||
for c in code.co_consts:
|
||||
if isinstance(c, CodeType):
|
||||
stack.append(c)
|
||||
yield code
|
||||
|
||||
def _line_numbers(self) -> Iterable[TLineNo]:
|
||||
"""Yield the line numbers possible in this code object.
|
||||
|
||||
Uses co_lines() to produce a sequence: l0, l1, ...
|
||||
"""
|
||||
for _, _, line in self.code.co_lines():
|
||||
if line:
|
||||
yield line
|
||||
|
||||
def find_statements(self) -> Iterable[TLineNo]:
|
||||
"""Find the statements in `self.code`.
|
||||
|
||||
Produce a sequence of line numbers that start statements. Recurses
|
||||
into all code objects reachable from `self.code`.
|
||||
|
||||
"""
|
||||
for bp in self._child_parsers():
|
||||
# Get all of the lineno information from this code.
|
||||
yield from bp._line_numbers()
|
||||
|
||||
|
||||
def bytes_to_lines(code: CodeType) -> dict[TOffset, TLineNo]:
|
||||
"""Make a dict mapping byte code offsets to line numbers."""
|
||||
b2l = {}
|
||||
for bstart, bend, lineno in code.co_lines():
|
||||
if lineno is not None:
|
||||
for boffset in range(bstart, bend, 2):
|
||||
b2l[boffset] = lineno
|
||||
return b2l
|
||||
|
||||
|
||||
def op_set(*op_names: str) -> set[int]:
|
||||
"""Make a set of opcodes from instruction names.
|
||||
|
||||
The names might not exist in this version of Python, skip those if not.
|
||||
"""
|
||||
ops = {op for name in op_names if (op := dis.opmap.get(name))}
|
||||
assert ops, f"At least one opcode must exist: {op_names}"
|
||||
return ops
|
||||
|
||||
|
||||
# Opcodes that are unconditional jumps elsewhere.
|
||||
ALWAYS_JUMPS = op_set(
|
||||
"JUMP_BACKWARD",
|
||||
"JUMP_BACKWARD_NO_INTERRUPT",
|
||||
"JUMP_FORWARD",
|
||||
)
|
||||
|
||||
# Opcodes that exit from a function.
|
||||
RETURNS = op_set(
|
||||
"RETURN_VALUE",
|
||||
"RETURN_GENERATOR",
|
||||
)
|
||||
|
||||
# Opcodes that do nothing.
|
||||
NOPS = op_set(
|
||||
"NOP",
|
||||
"NOT_TAKEN",
|
||||
)
|
||||
|
||||
|
||||
class InstructionWalker:
|
||||
"""Utility to step through trails of instructions.
|
||||
|
||||
We have two reasons to need sequences of instructions from a code object:
|
||||
First, in strict sequence to visit all the instructions in the object.
|
||||
This is `walk(follow_jumps=False)`. Second, we want to follow jumps to
|
||||
understand how execution will flow: `walk(follow_jumps=True)`.
|
||||
"""
|
||||
|
||||
def __init__(self, code: CodeType) -> None:
|
||||
self.code = code
|
||||
self.insts: dict[TOffset, dis.Instruction] = {}
|
||||
|
||||
inst = None
|
||||
for inst in dis.get_instructions(code):
|
||||
self.insts[inst.offset] = inst
|
||||
|
||||
assert inst is not None
|
||||
self.max_offset = inst.offset
|
||||
|
||||
def walk(
|
||||
self, *, start_at: TOffset = 0, follow_jumps: bool = True
|
||||
) -> Iterable[dis.Instruction]:
|
||||
"""
|
||||
Yield instructions starting from `start_at`. Follow unconditional
|
||||
jumps if `follow_jumps` is true.
|
||||
"""
|
||||
seen = set()
|
||||
offset = start_at
|
||||
while offset < self.max_offset + 1:
|
||||
if offset in seen:
|
||||
break
|
||||
seen.add(offset)
|
||||
if inst := self.insts.get(offset):
|
||||
yield inst
|
||||
if follow_jumps and inst.opcode in ALWAYS_JUMPS:
|
||||
offset = inst.jump_target
|
||||
continue
|
||||
offset += 2
|
||||
|
||||
|
||||
TBranchTrailsOneSource = dict[Optional[TArc], set[TOffset]]
|
||||
TBranchTrails = dict[TOffset, TBranchTrailsOneSource]
|
||||
|
||||
|
||||
def branch_trails(
|
||||
code: CodeType,
|
||||
multiline_map: Mapping[TLineNo, TLineNo],
|
||||
) -> TBranchTrails:
|
||||
"""
|
||||
Calculate branch trails for `code`.
|
||||
|
||||
`multiline_map` maps line numbers to the first line number of a
|
||||
multi-line statement.
|
||||
|
||||
Instructions can have a jump_target, where they might jump to next. Some
|
||||
instructions with a jump_target are unconditional jumps (ALWAYS_JUMPS), so
|
||||
they aren't interesting to us, since they aren't the start of a branch
|
||||
possibility.
|
||||
|
||||
Instructions that might or might not jump somewhere else are branch
|
||||
possibilities. For each of those, we track a trail of instructions. These
|
||||
are lists of instruction offsets, the next instructions that can execute.
|
||||
We follow the trail until we get to a new source line. That gives us the
|
||||
arc from the original instruction's line to the new source line.
|
||||
|
||||
"""
|
||||
the_trails: TBranchTrails = collections.defaultdict(lambda: collections.defaultdict(set))
|
||||
iwalker = InstructionWalker(code)
|
||||
for inst in iwalker.walk(follow_jumps=False):
|
||||
if not inst.jump_target:
|
||||
# We only care about instructions with jump targets.
|
||||
continue
|
||||
if inst.opcode in ALWAYS_JUMPS:
|
||||
# We don't care about unconditional jumps.
|
||||
continue
|
||||
|
||||
from_line = inst.line_number
|
||||
if from_line is None:
|
||||
continue
|
||||
from_line = multiline_map.get(from_line, from_line)
|
||||
|
||||
def add_one_branch_trail(
|
||||
trails: TBranchTrailsOneSource,
|
||||
start_at: TOffset,
|
||||
) -> None:
|
||||
# pylint: disable=cell-var-from-loop
|
||||
inst_offsets: set[TOffset] = set()
|
||||
to_line = None
|
||||
for inst2 in iwalker.walk(start_at=start_at, follow_jumps=True):
|
||||
inst_offsets.add(inst2.offset)
|
||||
l2 = inst2.line_number
|
||||
if l2 is not None:
|
||||
l2 = multiline_map.get(l2, l2)
|
||||
if l2 and l2 != from_line:
|
||||
to_line = l2
|
||||
break
|
||||
elif inst2.jump_target and (inst2.opcode not in ALWAYS_JUMPS):
|
||||
break
|
||||
elif inst2.opcode in RETURNS:
|
||||
to_line = -code.co_firstlineno
|
||||
break
|
||||
if to_line is not None:
|
||||
trails[(from_line, to_line)].update(inst_offsets)
|
||||
else:
|
||||
trails[None] = set()
|
||||
|
||||
# Calculate two trails: one from the next instruction, and one from the
|
||||
# jump_target instruction.
|
||||
trails: TBranchTrailsOneSource = collections.defaultdict(set)
|
||||
add_one_branch_trail(trails, start_at=inst.offset + 2)
|
||||
add_one_branch_trail(trails, start_at=inst.jump_target)
|
||||
the_trails[inst.offset] = trails
|
||||
|
||||
# Sometimes we get BRANCH_RIGHT or BRANCH_LEFT events from instructions
|
||||
# other than the original jump possibility instruction. Register each
|
||||
# trail under all of their offsets so we can pick up in the middle of a
|
||||
# trail if need be.
|
||||
for arc, offsets in trails.items():
|
||||
for offset in offsets:
|
||||
the_trails[offset][arc].update(offsets)
|
||||
|
||||
return the_trails
|
||||
|
||||
|
||||
def always_jumps(code: CodeType) -> dict[TOffset, TOffset]:
|
||||
"""Make a map of unconditional bytecodes jumping to others.
|
||||
|
||||
Only include bytecodes that do no work and go to another bytecode.
|
||||
"""
|
||||
jumps = {}
|
||||
iwalker = InstructionWalker(code)
|
||||
for inst in iwalker.walk(follow_jumps=False):
|
||||
if inst.opcode in ALWAYS_JUMPS:
|
||||
jumps[inst.offset] = inst.jump_target
|
||||
elif inst.opcode in NOPS:
|
||||
jumps[inst.offset] = inst.offset + 2
|
||||
return jumps
|
||||
Reference in new issue
Block a user