Module refinery.lib.scripts.js.analysis.cfg

JavaScript's contribution to the shared control-flow substrate: which node types are which control-flow shape, and where the parts of each are stored.

Everything structural — the graph, the frontier threading, the jump-target and handler stacks, and the shapes themselves — lives in refinery.lib.scripts.analysis.cfg. What remains here is _Builder.statement, the recognition of Js* node types, and the accessors that pull a construct apart. Three of the shape parameters are answered from JavaScript semantics rather than from syntax: switch falls through from one case to the next, an unlabelled break may leave a switch as well as a loop, and a catch carries no type filter, so a throw the guarded block makes gets past it only where binding the caught value can itself throw — see _catch_may_rethrow.

"""
JavaScript's contribution to the shared control-flow substrate: which node types are which
control-flow shape, and where the parts of each are stored.

Everything structural — the graph, the frontier threading, the jump-target and handler stacks, and
the shapes themselves — lives in `refinery.lib.scripts.analysis.cfg`. What remains here is
`_Builder.statement`, the recognition of `Js*` node types, and the accessors that pull a construct
apart. Three of the shape parameters are answered from JavaScript semantics rather than from syntax:
`switch` falls through from one case to the next, an unlabelled `break` may leave a `switch` as well
as a loop, and a `catch` carries no type filter, so a throw the guarded block makes gets past it only
where binding the caught value can itself throw — see `_catch_may_rethrow`.
"""
from __future__ import annotations

from typing import Sequence

from refinery.lib.scripts import Node
from refinery.lib.scripts.analysis.cfg import (
    ArmFlow,
    CfgBuilder,
    CfgNode,
    ControlFlowGraph,
    ControlFlowModel,
    ElementLocator,
)
from refinery.lib.scripts.analysis.cfg import build_control_flow as _build_control_flow
from refinery.lib.scripts.js.analysis.model import FUNCTION_NODES
from refinery.lib.scripts.js.model import (
    JsArrayPattern,
    JsBlockStatement,
    JsBreakStatement,
    JsCatchClause,
    JsContinueStatement,
    JsDoWhileStatement,
    JsForInStatement,
    JsForOfStatement,
    JsForStatement,
    JsIfStatement,
    JsLabeledStatement,
    JsObjectPattern,
    JsReturnStatement,
    JsScript,
    JsSwitchStatement,
    JsThrowStatement,
    JsTryStatement,
    JsWhileStatement,
    JsWithStatement,
)

__all__ = [
    'CfgNode',
    'ControlFlowGraph',
    'ControlFlowModel',
    'ElementLocator',
    'build_cfg',
    'build_control_flow',
    'build_control_flow_model',
]

_LOOP_NODES = (
    JsWhileStatement,
    JsDoWhileStatement,
    JsForStatement,
    JsForInStatement,
    JsForOfStatement,
)


def _catch_may_rethrow(handler: JsCatchClause | None) -> bool:
    """
    Whether a throw offered to *handler* may still reach whatever guards the construct.

    A `catch` carries no type filter, so the clause always matches and the throw stops there — with
    one exception, which is why this is a question rather than a constant. Binding the caught value
    is a destructuring assignment when the parameter is a pattern, and destructuring evaluates:
    `catch ({ a })` over a thrown `null` throws again while binding, and that second throw leaves
    the construct. A plain identifier binds by name and an omitted parameter binds nothing, so
    neither can.
    """
    if handler is None:
        return False
    return isinstance(handler.param, (JsObjectPattern, JsArrayPattern))


def _label_name(statement: Node) -> str | None:
    label = getattr(statement, 'label', None)
    return label.name if label is not None else None


class _Builder(CfgBuilder):
    """
    The JavaScript dispatch over `refinery.lib.scripts.analysis.cfg.CfgBuilder`.
    """

    def body_blocks(self, owner: Node) -> Sequence[Sequence[Node]]:
        """
        The one block a JavaScript body is. A concise arrow function's body is an expression rather
        than a block, and it is reported as the single statement it evaluates.
        """
        if isinstance(owner, JsScript):
            return [list(owner.body)]
        body = getattr(owner, 'body', None)
        if isinstance(body, JsBlockStatement):
            return [list(body.body)]
        if isinstance(body, Node):
            return [[body]]
        return []

    def statement(self, statement: Node, frontier: list[CfgNode]) -> list[CfgNode]:
        if isinstance(statement, JsBlockStatement):
            return self.sequence(statement.body, frontier)
        if isinstance(statement, JsIfStatement):
            arms: list[Node | None] = [statement.consequent]
            if statement.alternate is not None:
                arms.append(statement.alternate)
            return self.branch_on(
                statement, arms, frontier, exhaustive=statement.alternate is not None)
        if isinstance(statement, JsWhileStatement):
            return self.loop_head_tested(statement, statement.body, frontier)
        if isinstance(statement, JsDoWhileStatement):
            return self.loop_tail_tested(statement, statement.body, frontier)
        if isinstance(statement, JsForStatement):
            return self.loop_counted(
                statement.init, statement.test, statement.update, statement.body, frontier)
        if isinstance(statement, (JsForInStatement, JsForOfStatement)):
            return self.loop_head_tested(statement, getattr(statement, 'body', None), frontier)
        if isinstance(statement, JsSwitchStatement):
            return self._switch(statement, frontier)
        if isinstance(statement, JsTryStatement):
            handler = statement.handler
            finalizer = statement.finalizer
            return self.guarded(
                statement.block,
                [(handler, handler.body)] if handler is not None else (),
                finalizer,
                list(finalizer.body) if finalizer is not None else (),
                frontier,
                escapes=_catch_may_rethrow(handler),
            )
        if isinstance(statement, JsLabeledStatement):
            return self.labelled(
                _label_name(statement),
                statement.body,
                frontier,
                binds_to_body=isinstance(statement.body, (*_LOOP_NODES, JsSwitchStatement)),
            )
        if isinstance(statement, JsReturnStatement):
            return self.terminate(statement, frontier, exceptional=False)
        if isinstance(statement, JsThrowStatement):
            return self.terminate(statement, frontier, exceptional=True)
        if isinstance(statement, JsBreakStatement):
            return self.jump_out(statement, _label_name(statement), frontier)
        if isinstance(statement, JsContinueStatement):
            return self.jump_back(statement, _label_name(statement), frontier)
        if isinstance(statement, JsWithStatement):
            node = self.node(statement)
            self.link(frontier, node)
            return self.statement(statement.body, [node]) if statement.body else [node]
        return self.opaque(statement, frontier)

    def _switch(self, statement: JsSwitchStatement, frontier: list[CfgNode]) -> list[CfgNode]:
        arms: list[Sequence[Node]] = [list(case.body) for case in statement.cases]
        exhaustive = any(case.test is None for case in statement.cases)
        return self.dispatch(
            statement, arms, frontier, arm_flow=ArmFlow.SEQUENTIAL, exhaustive=exhaustive)


def build_cfg(owner: Node) -> ControlFlowGraph:
    """
    Build the control-flow graph of *owner*, a `refinery.lib.scripts.js.model.JsScript` or a
    function node, over its own body without descending into nested function bodies.
    """
    return _Builder(owner).build()


def build_control_flow(root: JsScript) -> dict[int, ControlFlowGraph]:
    """
    Build one control-flow graph per function and for the script itself, keyed by the owner node's
    identity.
    """
    return _build_control_flow(root, _Builder, FUNCTION_NODES)


def build_control_flow_model(root: JsScript) -> ControlFlowModel:
    """
    Build the `refinery.lib.scripts.analysis.cfg.ControlFlowModel` for a script root — the shared
    control-flow layer the `DominanceModel` and `LivenessModel` consume.
    """
    return ControlFlowModel(build_control_flow(root))

Functions

def build_cfg(owner)

Build the control-flow graph of owner, a JsScript or a function node, over its own body without descending into nested function bodies.

Expand source code Browse git

def build_cfg(owner: Node) -> ControlFlowGraph:
    """
    Build the control-flow graph of *owner*, a `refinery.lib.scripts.js.model.JsScript` or a
    function node, over its own body without descending into nested function bodies.
    """
    return _Builder(owner).build()

def build_control_flow(root)

Build one control-flow graph per function and for the script itself, keyed by the owner node's identity.

Expand source code Browse git

def build_control_flow(root: JsScript) -> dict[int, ControlFlowGraph]:
    """
    Build one control-flow graph per function and for the script itself, keyed by the owner node's
    identity.
    """
    return _build_control_flow(root, _Builder, FUNCTION_NODES)

def build_control_flow_model(root)

Build the ControlFlowModel for a script root — the shared control-flow layer the DominanceModel and LivenessModel consume.

Expand source code Browse git

def build_control_flow_model(root: JsScript) -> ControlFlowModel:
    """
    Build the `refinery.lib.scripts.analysis.cfg.ControlFlowModel` for a script root — the shared
    control-flow layer the `DominanceModel` and `LivenessModel` consume.
    """
    return ControlFlowModel(build_control_flow(root))

Classes

class CfgNode (graph, element, successors=<factory>, predecessors=<factory>, is_resumption_hub=False, is_hub_bound=False)

One vertex of a control-flow graph. element is the AST node it stands for — a statement, or a loop-head expression whose reads and writes occur at this point — or None for the synthetic entry and exit. successors lists the nodes control may pass to next. graph is the body this vertex belongs to, which is what makes the questions below answerable of a node alone.

is_resumption_hub marks the synthetic fan-out standing for the over-approximate half of a resuming handler — see CfgEdge. A walk that wants only the paths going forward declines to enter one, and declines to leave one, which makes it a projection of the graph rather than a filter on its edges. It is a fact about the node and not about the edges into it, because an edge kind is keyed by the pair of nodes it joins and CfgBuilder.add_edge builds a multigraph: two edges between the same pair collapse to one entry, so a plain edge drawn beside a hub edge would make the pair read as pure hub and a walk keyed on the kind would decline the plain path too. No pair of edges can disagree about what the node is.

is_hub_bound marks a node the precise, forward-only projection does not reach: it sits inside a handler body that resumes the block around it. That block's forward edges join each guarded statement to the one it resumes at; a statement of the handler itself is on neither end of one, and everything it may reach afterwards hangs off the hub. A forward edge it does carry belongs to a block further out — one whose own resuming set guards the statement this handler is written inside — and stands for that block's resumption, not this one, so it says nothing about where this handler carries on. A forward-only walk seeded here would stop dead where the real run carries on, which for a flood is the unsound direction; reachable_forward_from_any reads this and falls back to the hub for such a source.

Both are fields rather than questions asked of graph, and that is a matter of what a walk can afford as much as of where the fact lives. Every layer that had to ask them of a graph it was separately handed could be handed the wrong one — a node lifted from a nested body reads as neither against the graph around it, and the walk that asked takes the wrong branch without anything failing — and one depth-first sweep over a large script asks tens of millions of times. graph remains because a caller naming which body it means is a contract worth checking, and because ControlFlowGraph.hub_bound answers the same question for a caller holding ids.

eq=False because every map in every layer above is keyed by id(node), and two structurally equal statements are two distinct points in the program.

repr=False because a generated repr expands successors and predecessors, and the guard against recursion only covers the node currently being formatted, so a graph re-expands at every join: a chain of five hundred nodes — a script of five hundred statements — exhausts the interpreter's stack, and a handful of branches produces megabytes. A debugger, a failing assertion, or pytest --showlocals would print one.

Expand source code Browse git

@dataclass(eq=False, repr=False)
class CfgNode:
    """
    One vertex of a control-flow graph. `element` is the AST node it stands for — a statement, or a
    loop-head expression whose reads and writes occur at this point — or `None` for the synthetic
    entry and exit. `successors` lists the nodes control may pass to next. `graph` is the body this
    vertex belongs to, which is what makes the questions below answerable of a node alone.

    `is_resumption_hub` marks the synthetic fan-out standing for the over-approximate half of a
    resuming handler — see `CfgEdge`. A walk that wants only the paths going forward declines to
    enter one, and declines to leave one, which makes it a *projection* of the graph rather than a
    filter on its edges. It is a fact about the node and not about the edges into it, because an
    edge kind is keyed by the pair of nodes it joins and `CfgBuilder.add_edge` builds a multigraph:
    two edges between the same pair collapse to one entry, so a plain edge drawn beside a hub edge
    would make the pair read as pure hub and a walk keyed on the kind would decline the plain path
    too. No pair of edges can disagree about what the node is.

    `is_hub_bound` marks a node the precise, forward-only projection does not reach: it sits inside a
    handler body that resumes the block around it. That block's forward edges join each *guarded*
    statement to the one it resumes at; a statement of the handler itself is on neither end of one,
    and everything it may reach afterwards hangs off the hub. A forward edge it does carry belongs to
    a block further out — one whose own resuming set guards the statement this handler is written
    inside — and stands for that block's resumption, not this one, so it says nothing about where
    this handler carries on. A forward-only walk seeded here would stop dead where the real run
    carries on, which for a flood is the unsound direction; `reachable_forward_from_any` reads this
    and falls back to the hub for such a source.

    Both are fields rather than questions asked of *graph*, and that is a matter of what a walk can
    afford as much as of where the fact lives. Every layer that had to ask them of a graph it was
    separately handed could be handed the wrong one — a node lifted from a nested body reads as
    neither against the graph around it, and the walk that asked takes the wrong branch without
    anything failing — and one depth-first sweep over a large script asks tens of millions of times.
    `graph` remains because a caller naming which body it means is a contract worth checking, and
    because `ControlFlowGraph.hub_bound` answers the same question for a caller holding ids.

    `eq=False` because every map in every layer above is keyed by `id(node)`, and two structurally
    equal statements are two distinct points in the program.

    `repr=False` because a generated repr expands `successors` and `predecessors`, and the guard
    against recursion only covers the node currently being formatted, so a graph re-expands at every
    join: a chain of five hundred nodes — a script of five hundred statements — exhausts the
    interpreter's stack, and a handful of branches produces megabytes. A debugger, a failing
    assertion, or `pytest --showlocals` would print one.
    """
    graph: ControlFlowGraph
    element: Node | None
    successors: list[CfgNode] = field(default_factory=list)
    predecessors: list[CfgNode] = field(default_factory=list)
    is_resumption_hub: bool = False
    is_hub_bound: bool = False

Instance variables

var graph

The type of the None singleton.

var element

The type of the None singleton.

var successors

The type of the None singleton.

var predecessors

The type of the None singleton.

var is_resumption_hub

The type of the None singleton.

var is_hub_bound

The type of the None singleton.

class ControlFlowGraph (owner)

The control-flow graph of one function or script body. entry and exit are synthetic; every other node wraps an AST element reachable through node_of.

Expand source code Browse git

class ControlFlowGraph:
    """
    The control-flow graph of one function or script body. `entry` and `exit` are synthetic; every
    other node wraps an AST element reachable through `node_of`.
    """

    def __init__(self, owner: Node):
        self.owner = owner
        self._node_of: dict[int, CfgNode] = {}
        self._edge_kinds: dict[tuple[int, int], CfgEdge] = {}
        self._hub_bound: set[int] = set()
        self._resuming = False
        self._fallback: dict[int, CfgNode] = {}
        self.entry = CfgNode(self, None)
        self.exit = CfgNode(self, None)
        self.nodes: list[CfgNode] = [self.entry, self.exit]

    def node_of(self, element: Node) -> CfgNode | None:
        """
        The graph node standing for *element*, or `None` if *element* is not part of this body, or is
        a node the graph does not represent on its own such as a plain expression inside a statement.
        """
        return self._node_of.get(id(element))

    def fallback_of(self, handler: CfgNode) -> CfgNode | None:
        """
        Where a throw offered to *handler* goes if *handler* does not take it, or `None` when
        *handler* is not a handler entry of this graph.

        This is recorded rather than read off the edges because the two are not the same claim. The
        edge is drawn only where some run may decline — a clause with a type filter, a `trap` set
        that may fail to match — and a handler certain to take the throw has none, which is what
        makes it shield whatever guards the construct. The fact holds either way, and it is what
        answers the *counterfactual*: not where the throw goes, but where it would go if this
        handler were not written at all, which is the question asked before one is deleted.
        """
        return self._fallback.get(id(handler))

    def edge_kind(self, source: CfgNode, target: CfgNode) -> CfgEdge:
        """
        What the edge from *source* to *target* says about the run that takes it — see `CfgEdge`.
        An edge the builder recorded nothing for is `CfgEdge.NORMAL`, which is what an unrelated
        pair of nodes reads as too.
        """
        return self._edge_kinds.get((id(source), id(target)), CfgEdge.NORMAL)

    def is_exceptional(self, source: CfgNode, target: CfgNode) -> bool:
        """
        Whether the error travels along the edge from *source* to *target*: the edge into a handler
        offered the throw, or outward from a set that declined it. This is the question `faults`
        asks — where an error goes — and it is *narrower* than `raise_taken`, because a handler that
        resumes swallows the error and carries on along an edge no error travels.
        """
        return bool(self.edge_kind(source, target) & CfgEdge.ERROR_CARRYING)

    def raise_taken(self, source: CfgNode, target: CfgNode) -> bool:
        """
        Whether the edge from *source* to *target* is taken only when *source* throws rather than
        completing normally. A definition *source* makes is not guaranteed to have happened along
        such an edge, so a flow-sensitive analysis must not treat it as a kill there.

        This is the question about *completion*, and it is the one nearly every consumer means. A
        resumption edge answers it although no error travels along it: the statement that resumed
        the block is precisely the one that did not finish.
        """
        return bool(self.edge_kind(source, target) & RAISE_TAKEN)

    @property
    def hub_bound(self) -> Set[int]:
        """
        The ids of the nodes `CfgNode.is_hub_bound` is set on, for a caller holding node *ids*
        rather than nodes. Written beside the field by `CfgBuilder.mark_hub_bound`, which is the one
        place either is recorded. `refinery.lib.scripts.analysis.reaching.ReachabilityQuery` is one: its
        candidate sets are ids because the layers above memoize them that way, and turning them back
        into nodes to ask each one is what those caches exist to avoid.
        """
        return self._hub_bound

    @property
    def carries_resumption(self) -> bool:
        """
        Whether any handler of this body resumes the block it guards, which is the only shape over
        which the two projections of `CfgEdge` differ. A body with none — every script that writes
        no `trap`, and every one whose traps rethrow — draws no resumption edge and marks no node
        hub-bound, so a forward walk over it answers exactly what `reachable_from_any` does and may
        be answered by the cheaper one.
        """
        return self._resuming

Instance variables

var hub_bound

The ids of the nodes CfgNode.is_hub_bound is set on, for a caller holding node ids rather than nodes. Written beside the field by CfgBuilder.mark_hub_bound, which is the one place either is recorded. ReachabilityQuery is one: its candidate sets are ids because the layers above memoize them that way, and turning them back into nodes to ask each one is what those caches exist to avoid.

Expand source code Browse git

@property
def hub_bound(self) -> Set[int]:
    """
    The ids of the nodes `CfgNode.is_hub_bound` is set on, for a caller holding node *ids*
    rather than nodes. Written beside the field by `CfgBuilder.mark_hub_bound`, which is the one
    place either is recorded. `refinery.lib.scripts.analysis.reaching.ReachabilityQuery` is one: its
    candidate sets are ids because the layers above memoize them that way, and turning them back
    into nodes to ask each one is what those caches exist to avoid.
    """
    return self._hub_bound

var carries_resumption

Whether any handler of this body resumes the block it guards, which is the only shape over which the two projections of CfgEdge differ. A body with none — every script that writes no trap, and every one whose traps rethrow — draws no resumption edge and marks no node hub-bound, so a forward walk over it answers exactly what reachable_from_any does and may be answered by the cheaper one.

Expand source code Browse git

@property
def carries_resumption(self) -> bool:
    """
    Whether any handler of this body resumes the block it guards, which is the only shape over
    which the two projections of `CfgEdge` differ. A body with none — every script that writes
    no `trap`, and every one whose traps rethrow — draws no resumption edge and marks no node
    hub-bound, so a forward walk over it answers exactly what `reachable_from_any` does and may
    be answered by the cheaper one.
    """
    return self._resuming

Methods

def node_of(self, element)

The graph node standing for element, or None if element is not part of this body, or is a node the graph does not represent on its own such as a plain expression inside a statement.

Expand source code Browse git

def node_of(self, element: Node) -> CfgNode | None:
    """
    The graph node standing for *element*, or `None` if *element* is not part of this body, or is
    a node the graph does not represent on its own such as a plain expression inside a statement.
    """
    return self._node_of.get(id(element))

def fallback_of(self, handler)

Where a throw offered to handler goes if handler does not take it, or None when handler is not a handler entry of this graph.

This is recorded rather than read off the edges because the two are not the same claim. The edge is drawn only where some run may decline — a clause with a type filter, a trap set that may fail to match — and a handler certain to take the throw has none, which is what makes it shield whatever guards the construct. The fact holds either way, and it is what answers the counterfactual: not where the throw goes, but where it would go if this handler were not written at all, which is the question asked before one is deleted.

Expand source code Browse git

def fallback_of(self, handler: CfgNode) -> CfgNode | None:
    """
    Where a throw offered to *handler* goes if *handler* does not take it, or `None` when
    *handler* is not a handler entry of this graph.

    This is recorded rather than read off the edges because the two are not the same claim. The
    edge is drawn only where some run may decline — a clause with a type filter, a `trap` set
    that may fail to match — and a handler certain to take the throw has none, which is what
    makes it shield whatever guards the construct. The fact holds either way, and it is what
    answers the *counterfactual*: not where the throw goes, but where it would go if this
    handler were not written at all, which is the question asked before one is deleted.
    """
    return self._fallback.get(id(handler))

def edge_kind(self, source, target)

What the edge from source to target says about the run that takes it — see CfgEdge. An edge the builder recorded nothing for is CfgEdge.NORMAL, which is what an unrelated pair of nodes reads as too.

Expand source code Browse git

def edge_kind(self, source: CfgNode, target: CfgNode) -> CfgEdge:
    """
    What the edge from *source* to *target* says about the run that takes it — see `CfgEdge`.
    An edge the builder recorded nothing for is `CfgEdge.NORMAL`, which is what an unrelated
    pair of nodes reads as too.
    """
    return self._edge_kinds.get((id(source), id(target)), CfgEdge.NORMAL)

def is_exceptional(self, source, target)

Whether the error travels along the edge from source to target: the edge into a handler offered the throw, or outward from a set that declined it. This is the question faults asks — where an error goes — and it is narrower than raise_taken, because a handler that resumes swallows the error and carries on along an edge no error travels.

Expand source code Browse git

def is_exceptional(self, source: CfgNode, target: CfgNode) -> bool:
    """
    Whether the error travels along the edge from *source* to *target*: the edge into a handler
    offered the throw, or outward from a set that declined it. This is the question `faults`
    asks — where an error goes — and it is *narrower* than `raise_taken`, because a handler that
    resumes swallows the error and carries on along an edge no error travels.
    """
    return bool(self.edge_kind(source, target) & CfgEdge.ERROR_CARRYING)

def raise_taken(self, source, target)

Whether the edge from source to target is taken only when source throws rather than completing normally. A definition source makes is not guaranteed to have happened along such an edge, so a flow-sensitive analysis must not treat it as a kill there.

This is the question about completion, and it is the one nearly every consumer means. A resumption edge answers it although no error travels along it: the statement that resumed the block is precisely the one that did not finish.

Expand source code Browse git

def raise_taken(self, source: CfgNode, target: CfgNode) -> bool:
    """
    Whether the edge from *source* to *target* is taken only when *source* throws rather than
    completing normally. A definition *source* makes is not guaranteed to have happened along
    such an edge, so a flow-sensitive analysis must not treat it as a kill there.

    This is the question about *completion*, and it is the one nearly every consumer means. A
    resumption edge answers it although no error travels along it: the statement that resumed
    the block is precisely the one that did not finish.
    """
    return bool(self.edge_kind(source, target) & RAISE_TAKEN)

class ControlFlowModel (graphs)

The per-body control-flow graphs of one script, paired with the ElementLocator that maps any AST node to the graph node evaluating it. Built once over the script root — the graphs are purely syntactic and need no semantic model — and shared by every solver layered on it, which would otherwise each rebuild the whole set.

Expand source code Browse git

class ControlFlowModel:
    """
    The per-body control-flow graphs of one script, paired with the `ElementLocator` that maps any
    AST node to the graph node evaluating it. Built once over the script root — the graphs are purely
    syntactic and need no semantic model — and shared by every solver layered on it, which would
    otherwise each rebuild the whole set.
    """

    def __init__(self, graphs: dict[int, ControlFlowGraph]):
        self.graphs = graphs
        self._locator = ElementLocator(graphs)

    def graph_of(self, owner: Node) -> ControlFlowGraph | None:
        """
        The control-flow graph owned by *owner* — a function node or the script root — or `None` when
        it owns none.
        """
        return self.graphs.get(id(owner))

    def node_of(self, element: Node) -> CfgNode | None:
        """
        The control-flow node standing for *element*, or `None` when the graphs do not represent it
        directly. Delegates to the shared `ElementLocator`.
        """
        return self._locator.node_of(element)

    def locate(self, element: Node) -> tuple[ControlFlowGraph, CfgNode] | None:
        """
        The graph and node that evaluate *element*, climbing out of any enclosing expression, or
        `None` when it has no enclosing graph node. Delegates to the shared `ElementLocator`.
        """
        return self._locator.locate(element)

Methods

def graph_of(self, owner)

The control-flow graph owned by owner — a function node or the script root — or None when it owns none.

Expand source code Browse git

def graph_of(self, owner: Node) -> ControlFlowGraph | None:
    """
    The control-flow graph owned by *owner* — a function node or the script root — or `None` when
    it owns none.
    """
    return self.graphs.get(id(owner))

def node_of(self, element)

The control-flow node standing for element, or None when the graphs do not represent it directly. Delegates to the shared ElementLocator.

Expand source code Browse git

def node_of(self, element: Node) -> CfgNode | None:
    """
    The control-flow node standing for *element*, or `None` when the graphs do not represent it
    directly. Delegates to the shared `ElementLocator`.
    """
    return self._locator.node_of(element)

def locate(self, element)

The graph and node that evaluate element, climbing out of any enclosing expression, or None when it has no enclosing graph node. Delegates to the shared ElementLocator.

Expand source code Browse git

def locate(self, element: Node) -> tuple[ControlFlowGraph, CfgNode] | None:
    """
    The graph and node that evaluate *element*, climbing out of any enclosing expression, or
    `None` when it has no enclosing graph node. Delegates to the shared `ElementLocator`.
    """
    return self._locator.locate(element)

class ElementLocator (graphs)

Locates an AST node among the per-body control-flow graphs of one script. Built once from the graph set, it maps an element to the graph and node that evaluate it — directly for an element a graph node stands for (node_of), or by climbing to the enclosing statement for one nested inside an expression (locate). Every flow-sensitive layer built on the graphs shares it, so the AST-to-graph mapping and its parent-climb live in one place.

Expand source code Browse git

class ElementLocator:
    """
    Locates an AST node among the per-body control-flow graphs of one script. Built once from the
    graph set, it maps an element to the graph and node that evaluate it — directly for an element a
    graph node stands for (`node_of`), or by climbing to the enclosing statement for one nested
    inside an expression (`locate`). Every flow-sensitive layer built on the graphs shares it, so the
    AST-to-graph mapping and its parent-climb live in one place.
    """

    def __init__(self, graphs: dict[int, ControlFlowGraph]):
        self._element_graph: dict[int, ControlFlowGraph] = {}
        self._owners = {id(graph.owner) for graph in graphs.values()}
        for graph in graphs.values():
            for node in graph.nodes:
                if node.element is not None:
                    self._element_graph[id(node.element)] = graph

    def node_of(self, element: Node) -> CfgNode | None:
        """
        The control-flow node standing for *element* in whichever graph owns it, or `None` when
        *element* is not itself a node the graphs represent.
        """
        graph = self._element_graph.get(id(element))
        return graph.node_of(element) if graph is not None else None

    def locate(self, element: Node) -> tuple[ControlFlowGraph, CfgNode] | None:
        """
        The graph and node that evaluate *element*, climbing out of any expression it is nested in to
        the enclosing statement or loop head, or `None` when it has no enclosing graph node.

        The climb stops at the body *element* is written in rather than continuing into the body
        around it. Something inside a body that no node of that body's graph stands for — the default
        of a parameter, an attribute on the body itself — is evaluated when that body is invoked, and
        the enclosing body's statement that mentions it is not that point. Answering with that
        statement orders the element against code the invocation may never run beside, which is the
        false claim the per-body split exists to avoid; `None` says the graphs do not place it, and a
        caller reads that as unknown.

        The body *element* is itself is not its own boundary: a block is a value written at a point
        in the body around it, so locating one climbs out to the statement that mentions it.
        """
        cursor: Node | None = element
        while cursor is not None:
            graph = self._element_graph.get(id(cursor))
            if graph is not None:
                node = graph.node_of(cursor)
                if node is not None:
                    return graph, node
            if cursor is not element and id(cursor) in self._owners:
                return None
            cursor = cursor.parent
        return None

Methods

def node_of(self, element)

The control-flow node standing for element in whichever graph owns it, or None when element is not itself a node the graphs represent.

Expand source code Browse git

def node_of(self, element: Node) -> CfgNode | None:
    """
    The control-flow node standing for *element* in whichever graph owns it, or `None` when
    *element* is not itself a node the graphs represent.
    """
    graph = self._element_graph.get(id(element))
    return graph.node_of(element) if graph is not None else None

def locate(self, element)

The graph and node that evaluate element, climbing out of any expression it is nested in to the enclosing statement or loop head, or None when it has no enclosing graph node.

The climb stops at the body element is written in rather than continuing into the body around it. Something inside a body that no node of that body's graph stands for — the default of a parameter, an attribute on the body itself — is evaluated when that body is invoked, and the enclosing body's statement that mentions it is not that point. Answering with that statement orders the element against code the invocation may never run beside, which is the false claim the per-body split exists to avoid; None says the graphs do not place it, and a caller reads that as unknown.

The body element is itself is not its own boundary: a block is a value written at a point in the body around it, so locating one climbs out to the statement that mentions it.

Expand source code Browse git

def locate(self, element: Node) -> tuple[ControlFlowGraph, CfgNode] | None:
    """
    The graph and node that evaluate *element*, climbing out of any expression it is nested in to
    the enclosing statement or loop head, or `None` when it has no enclosing graph node.

    The climb stops at the body *element* is written in rather than continuing into the body
    around it. Something inside a body that no node of that body's graph stands for — the default
    of a parameter, an attribute on the body itself — is evaluated when that body is invoked, and
    the enclosing body's statement that mentions it is not that point. Answering with that
    statement orders the element against code the invocation may never run beside, which is the
    false claim the per-body split exists to avoid; `None` says the graphs do not place it, and a
    caller reads that as unknown.

    The body *element* is itself is not its own boundary: a block is a value written at a point
    in the body around it, so locating one climbs out to the statement that mentions it.
    """
    cursor: Node | None = element
    while cursor is not None:
        graph = self._element_graph.get(id(cursor))
        if graph is not None:
            node = graph.node_of(cursor)
            if node is not None:
                return graph, node
        if cursor is not element and id(cursor) in self._owners:
            return None
        cursor = cursor.parent
    return None