diff --git a/codeanalyzer/__main__.py b/codeanalyzer/__main__.py index fba557e..721b6ed 100644 --- a/codeanalyzer/__main__.py +++ b/codeanalyzer/__main__.py @@ -115,7 +115,7 @@ def main( "-a", "--analysis-level", help="Analysis depth: 1=symbol table+Jedi call graph, 2=+PyCG call " - "graph, 3=+native dataflow graphs (CFG/PDG/SDG).", + "graph, 3=+native intraprocedural dataflow (CFG/PDG).", min=1, max=3, ), @@ -125,10 +125,10 @@ def main( typer.Option( "--graphs", help="Level 3 only: comma-separated program-graph sections to emit " - "(cfg, dfg, pdg, sdg). Default: all. `dfg` emits the PDG's data " - "edges only; `sdg` implies the dependence edges it stitches.", + "(cfg, dfg, pdg, sdg). Default: cfg,dfg,pdg. `dfg` emits the PDG's data " + "edges only; `sdg` requires -a 4 (not yet available).", ), - ] = "cfg,dfg,pdg,sdg", + ] = "cfg,dfg,pdg", graph_field_depth: Annotated[ int, typer.Option( @@ -277,7 +277,10 @@ def main( if not selected_graphs: logger.error("--graphs requires at least one of: " + ", ".join(VALID_GRAPHS)) raise typer.Exit(code=2) - if analysis_level < 3 and graphs != "cfg,dfg,pdg,sdg": + if "sdg" in selected_graphs: + logger.error("--graphs sdg requires -a 4 (interprocedural SDG); not available yet.") + raise typer.Exit(code=2) + if analysis_level < 3 and graphs != "cfg,dfg,pdg": logger.error("--graphs is a level-3 option; pass -a 3 to emit program graphs.") raise typer.Exit(code=2) if analysis_level < 3 and graph_field_depth != 3: diff --git a/codeanalyzer/core.py b/codeanalyzer/core.py index 48431f5..7c94c20 100644 --- a/codeanalyzer/core.py +++ b/codeanalyzer/core.py @@ -468,7 +468,20 @@ def analyze(self) -> Analysis: backfill_callees(app, sig_to_id) reidentify_call_graph(app, sig_to_id) - # L3/L4 dataflow emission rebuilt on the v2 tree in Stage 3+ + # L3: intraprocedural dataflow (CFG/CDG/DDG) emitted onto the v2 tree. + if self.analysis_level >= 3: + from codeanalyzer.dataflow.builder import ( + build_function_pdgs, + emit_l3_body, + ) + from codeanalyzer.dataflow.syntactic import SyntacticOracle + + infos, _func_asts = build_function_pdgs( + app, + k=self.options.graph_field_depth, + oracle_factory=lambda c: SyntacticOracle(), + ) + emit_l3_body(app, infos, sig_to_id, set(self.options.graphs.split(","))) # Build the v2 envelope, then persist it (the cache stores the full # ``Analysis`` envelope so a reused cache round-trips schema_version). @@ -507,6 +520,16 @@ def _load_pyapplication_from_cache(self, cache_file: Path) -> Optional[Analysis] if getattr(cached, "schema_version", None) != "2.0.0": logger.info("stale/incompatible analysis cache (schema_version) — rebuilding") return None + # The cache keys only on file hash/mtime/size, not on level, so a cache + # built at a different analysis_level would leak higher-level body/edge + # content (or omit content when the cached level is lower). Reject the + # mismatch and force a full rebuild at the requested level. + if cached.max_level != self.analysis_level: + logger.info( + f"cache built at level {cached.max_level} != requested " + f"{self.analysis_level} — rebuilding" + ) + return None return cached def _save_analysis_cache(self, analysis: Analysis, cache_file: Path) -> None: diff --git a/codeanalyzer/dataflow/builder.py b/codeanalyzer/dataflow/builder.py index 01fd256..4eb2d85 100644 --- a/codeanalyzer/dataflow/builder.py +++ b/codeanalyzer/dataflow/builder.py @@ -36,7 +36,7 @@ import ast from pathlib import Path -from typing import Dict, List, Optional, Set, Tuple +from typing import Callable, Dict, List, Optional, Set, Tuple from codeanalyzer.dataflow.access_paths import _PathExtractor, _calls_in from codeanalyzer.dataflow.alias import TypeBasedAliasOracle @@ -127,14 +127,24 @@ def _match_args( return tuple(pairs) -def build_program_graphs( +def build_function_pdgs( app: PyApplication, k: int = DEFAULT_K_LIMIT, -) -> ProgramGraphsIR: - """Build CFG/PDG per callable and the whole-program SDG.""" - class_idx = _class_index(app) - callable_idx = _callable_index(app) - + *, + oracle_factory: Callable[[PyCallable], object], +) -> Tuple[Dict[str, FunctionInfo], Dict[str, ast.AST]]: + """Intraprocedural phase only: one ``FunctionInfo`` (CFG → PDG) per + callable, keyed by signature, with no SDG/summary/callsite work. + + ``oracle_factory(pycallable)`` supplies the may-alias oracle per callable — + ``TypeBasedAliasOracle`` for the L4 path, ``SyntacticOracle`` for L3. + + Returns ``(infos, func_asts)`` rather than bare PDGs so that the L4 + orchestrator (:func:`build_program_graphs`) still has both the + ``FunctionInfo`` records its callsite/summary/SDG phases mutate and the + matched def nodes its Phase 2 reads. L3 callers just read ``info.pdg`` per + signature and ignore ``func_asts``. + """ infos: Dict[str, FunctionInfo] = {} func_asts: Dict[str, ast.AST] = {} @@ -166,7 +176,7 @@ def build_program_graphs( if enclosing_ast is not None: enclosing_locals |= _locals_of(enclosing_ast) - oracle = TypeBasedAliasOracle(_base_types(pycallable)) + oracle = oracle_factory(pycallable) pdg = build_pdg( func, enclosing_locals=enclosing_locals, @@ -179,6 +189,137 @@ def build_program_graphs( ) func_asts[pycallable.signature] = func + return infos, func_asts + + +def emit_l3_body( + app: PyApplication, + infos: Dict[str, FunctionInfo], + sig_to_id: Dict[str, str], + graphs: Set[str], +) -> None: + """Project each callable's syntactic PDG onto the v2 tree at L3. + + For every callable that produced a ``FunctionInfo`` in + :func:`build_function_pdgs` (syntactic oracle), this writes onto the + matching ``PyCallable`` in ``app``'s symbol table: + + * ``body`` — one node per CFG node, keyed by its LOCAL id (``"@entry"``/ + ``"@exit"`` for the synthetic bookends, ``"line:col"`` for real + statements — the same key format L1 uses). A statement position an L1 + pass already materialized as a ``call`` node lands on the SAME local key, + so it keeps its ``call`` kind and L2-resolved ``callee`` in place (no + re-keying, no duplication) and is only given the byte-offset ``span`` L1 + could not compute. + * ``cfg`` — one ``CfgEdge`` per CFG edge, endpoints as local ids. + * ``cdg`` — the PDG's control-dependence edges. + * ``ddg`` — the PDG's syntactic def-use edges, each with ``prov=["ssa"]`` + (no points-to provenance at L3; that is the L4 delta). + + ``graphs`` scopes the edge lists exactly as the dormant + :func:`to_program_graphs` does: ``cfg`` needs ``"cfg"``; ``cdg`` needs + ``"pdg"``/``"sdg"``; ``ddg`` needs those or ``"dfg"``. ``body`` is always + populated. Callables absent from ``infos`` (unrecovered AST) are skipped. + """ + from codeanalyzer.dataflow.identity import IdentityMap + from codeanalyzer.schema.py_schema import ( + BodyNode, + CdgEdge, + CfgEdge, + DdgEdge, + Span, + byte_offsets, + ) + + want_pdg = bool({"pdg", "sdg"} & graphs) + want_cfg = "cfg" in graphs + want_ddg = want_pdg or "dfg" in graphs + + def _span_of(source: str, node) -> Optional["Span"]: + if not source or node.start_line < 1: + return None + return Span( + start=(node.start_line, node.start_column), + end=(node.end_line, node.end_column), + bytes=byte_offsets( + source, + node.start_line, + node.start_column, + node.end_line, + node.end_column, + ), + ) + + for module in app.symbol_table.values(): + source = module.source + for pycallable, _chain in _walk_callables(module): + info = infos.get(pycallable.signature) + if info is None: + continue + pdg = info.pdg + callable_id = sig_to_id.get(pycallable.signature) or pycallable.id + im = IdentityMap.for_function(callable_id, pdg) + + for node in pdg.cfg.nodes: + local = im.local(node.id) + if node.id == pdg.cfg.entry_id: + pycallable.body[local] = BodyNode(kind="entry") + continue + if node.id == pdg.cfg.exit_id: + pycallable.body[local] = BodyNode(kind="exit") + continue + span = _span_of(source, node) + # An L1 `call` node was keyed by its LOCAL "line:col"; this CFG + # node at the same position lands on the SAME key, so keep the + # node's `call` kind and L2-resolved `callee` in place and just + # fill any missing span — never re-key or duplicate it. + existing = pycallable.body.get(local) + if existing is not None: + if existing.span is None and span is not None: + existing.span = span + continue + pycallable.body[local] = BodyNode(kind=node.kind, span=span) + + if want_cfg: + pycallable.cfg = [ + CfgEdge( + src=im.local(e.source), + dst=im.local(e.target), + kind=e.kind, + ) + for e in pdg.cfg.edges + ] + if want_pdg: + pycallable.cdg = [ + CdgEdge(src=im.local(e.source), dst=im.local(e.target)) + for e in pdg.edges + if e.type == "CDG" + ] + if want_ddg: + pycallable.ddg = [ + DdgEdge( + src=im.local(e.source), + dst=im.local(e.target), + var=e.var, + prov=["ssa"], + ) + for e in pdg.edges + if e.type == "DDG" + ] + + +def build_program_graphs( + app: PyApplication, + k: int = DEFAULT_K_LIMIT, +) -> ProgramGraphsIR: + """Build CFG/PDG per callable and the whole-program SDG.""" + class_idx = _class_index(app) + callable_idx = _callable_index(app) + + infos, func_asts = build_function_pdgs( + app, k, oracle_factory=lambda c: TypeBasedAliasOracle(_base_types(c)) + ) + # Callsites and nested defs, now that every signature is known. for sig, info in infos.items(): pycallable = callable_idx[sig] diff --git a/codeanalyzer/dataflow/identity.py b/codeanalyzer/dataflow/identity.py new file mode 100644 index 0000000..328346d --- /dev/null +++ b/codeanalyzer/dataflow/identity.py @@ -0,0 +1,48 @@ +"""Bijection between internal IR node ids (ints, per function) and their +canonical ids. + +Two forms per node: + +* **local** — the intra-callable id used as the ``body`` map key and as every + ``cfg``/``cdg``/``ddg`` edge endpoint: ``"@entry"``/``"@exit"`` for the + synthetic CFG bookends, ``"line:col"`` for real statements. This matches the + key format L1 already uses for ``call`` nodes (see ``schema/l1_body.py``), so + an L1 body node and its coinciding CFG node land on the same key and L1 ⊆ L3 + holds. +* **global** — ``"@"``, the fully addressable id for + cross-callable references and the Neo4j PyCFGNode keys (a later task). +""" +from __future__ import annotations +from typing import Dict, Iterable + + +class IdentityMap: + def __init__(self, callable_id: str, id_to_local: Dict[int, str]): + self._callable_id = callable_id + self._map = id_to_local + + @classmethod + def for_function(cls, callable_id: str, pdg) -> "IdentityMap": + cfg = pdg.cfg + m: Dict[int, str] = {} + for n in cfg.nodes: + if n.id == cfg.entry_id: + m[n.id] = "@entry" + elif n.id == cfg.exit_id: + m[n.id] = "@exit" + else: + m[n.id] = f"{n.start_line}:{n.start_column}" + return cls(callable_id, m) + + def local(self, node_id: int) -> str: + """Intra-callable id: ``"@entry"``/``"@exit"`` or ``"line:col"``.""" + return self._map[node_id] + + def global_id(self, node_id: int) -> str: + """Fully addressable id: ``"@"``.""" + loc = self._map[node_id] + # local statements are "line:col"; bookends already carry the leading "@" + return f"{self._callable_id}{loc}" if loc.startswith("@") else f"{self._callable_id}@{loc}" + + def node_ids(self) -> Iterable[int]: + return self._map.keys() diff --git a/codeanalyzer/dataflow/syntactic.py b/codeanalyzer/dataflow/syntactic.py new file mode 100644 index 0000000..9d30d7f --- /dev/null +++ b/codeanalyzer/dataflow/syntactic.py @@ -0,0 +1,26 @@ +################################################################################ +# Copyright IBM Corporation 2025 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +################################################################################ + +"""The L3 (syntactic) alias oracle: two access paths alias iff they are the +identical path. Bypasses the type-based may-alias so def-use yields only +name-equality (textual) edges — the alias-derived edges are the L4 delta.""" + +from __future__ import annotations + + +class SyntacticOracle: + def may_alias(self, path_a: str, path_b: str) -> bool: + return path_a == path_b diff --git a/codeanalyzer/neo4j/project.py b/codeanalyzer/neo4j/project.py index df76b38..2961f42 100644 --- a/codeanalyzer/neo4j/project.py +++ b/codeanalyzer/neo4j/project.py @@ -76,7 +76,9 @@ def project(app: PyApplication, app_name: str, sig_to_id: dict) -> GraphRows: "PY_CALLS", src, tgt, _call_edge_props(e.weight, list(e.provenance or [])) ) - # CPG overlay projection rebuilt on the v2 tree in Stage 3 + # Level-3 CPG overlay: each callable's v2 body/cfg/cdg/ddg. Idempotent under + # MERGE — a no-op when no callable carries L3 fields (levels 1/2). + _project_program_graphs(b, app) return b.finish() @@ -86,92 +88,78 @@ def project(app: PyApplication, app_name: str, sig_to_id: dict) -> GraphRows: # ---------------------------------------------------------------------------------------------- -def _signature_modules(app: PyApplication) -> dict: - """signature → owning module file_key, for CFGNode `_module` provenance.""" - from codeanalyzer.semantic_analysis.call_graph import _walk_module_callables +def _global_ordinal(callable_id: str, local_key: str) -> str: + """The globally-unique PyCFGNode merge key for a callable's body node: the + callable's ``can://`` id joined to its LOCAL body key with a single ``@``. + The synthetic bookends already carry the leading ``@`` (``"@entry"``/ + ``"@exit"``); real statements are bare ``"line:col"`` and gain the ``@``. - out: dict = {} - for file_key, mod in app.symbol_table.items(): - for c in _walk_module_callables(mod): - out[c.signature] = file_key - return out + This MUST agree with :meth:`IdentityMap.global_id` for the same node, so the + JSON ``body``/``cfg`` projection and this Neo4j projection land on one node + identity (two-projection agreement).""" + return ( + f"{callable_id}{local_key}" + if local_key.startswith("@") + else f"{callable_id}@{local_key}" + ) -def _cfg_node_ref(b: RowBuilder, sig: str, node_id: int) -> NodeRef: - return NodeRef("PyCFGNode", "id", f"{sig}#{node_id}") +def _cfg_ref(callable_id: str, local_key: str) -> NodeRef: + return NodeRef("PyCFGNode", "id", _global_ordinal(callable_id, local_key)) def _project_program_graphs(b: RowBuilder, app: PyApplication) -> None: - """CFG/PDG/SDG rows: node label ``PyCFGNode`` (merge key ``id`` = - ``#``) and edge types ``PY_HAS_CFG_NODE`` / - ``PY_CFG_NEXT`` (prop ``kind``) / ``PY_CDG`` / ``PY_DDG`` (prop ``var``) / - ``PY_PARAM_IN`` / ``PY_PARAM_OUT`` / ``PY_SUMMARY``. The vocabulary is - cross-language in shape but PY_-namespaced like every other row family, so - a multi-language database never mingles analyzers' dependence edges. - Parameter nodes ride the same label with their HRB kinds plus - ``var``/``call_node`` props (an additive, recorded extension).""" - pg = app.program_graphs - sig_module = _signature_modules(app) - - for sig, fg in pg.functions.items(): - owner = _sym(sig) - module = sig_module.get(sig) - for n in (fg.cfg.nodes if fg.cfg else []): - ref = b.node( - ["PyCFGNode"], - "id", - f"{sig}#{n.id}", - prune( - { - "kind": n.kind, - "start_line": n.start_line, - "end_line": n.end_line, - "_module": module, - } - ), - ) - b.edge("PY_HAS_CFG_NODE", owner, ref) - for p in fg.param_nodes or []: - ref = b.node( - ["PyCFGNode"], - "id", - f"{sig}#{p.id}", - prune( - { - "kind": p.kind, - "var": p.var, - "call_node": p.call_node, - "start_line": p.start_line, - "end_line": p.end_line, - "_module": module, - } - ), - ) - b.edge("PY_HAS_CFG_NODE", owner, ref) - for e in (fg.cfg.edges if fg.cfg else []): - b.edge( - "PY_CFG_NEXT", - _cfg_node_ref(b, sig, e.source), - _cfg_node_ref(b, sig, e.target), - {"kind": e.kind}, - ) - for e in (fg.pdg.edges if fg.pdg else []): - b.edge( - f"PY_{e.type}", # PY_CDG | PY_DDG - _cfg_node_ref(b, sig, e.source), - _cfg_node_ref(b, sig, e.target), - prune({"var": e.var}), - ) + """Level-3 CPG overlay, projected off each callable's v2 ``body``/``cfg``/ + ``cdg``/``ddg`` (populated by ``emit_l3_body`` at ``-a 3``; empty otherwise). + + Node label ``PyCFGNode`` (merge key ``id`` = the GLOBAL ordinal + ``@`` — identical to the JSON body key + prefixed with the callable id, so the two projections agree). Edges: + ``PY_HAS_CFG_NODE`` from the owning callable, ``PY_CFG_NEXT`` (prop ``kind``) + over the CFG, ``PY_CDG`` over control dependence, and ``PY_DDG`` (props + ``var``/``prov``) over data dependence. The vocabulary is cross-language in + shape but PY_-namespaced like every other row family, so a multi-language + database never mingles analyzers' dependence edges. Body-node ``var``/ + ``call_node`` props are an L4 parameter-node concern and are absent here.""" + from codeanalyzer.semantic_analysis.call_graph import _walk_module_callables - for e in pg.sdg_edges: - if e.type == "CALL": - continue # the callable-level PY_CALLS twin already carries calls - b.edge( - f"PY_{e.type}", # PY_PARAM_IN | PY_PARAM_OUT | PY_SUMMARY - _cfg_node_ref(b, e.source.signature, e.source.node), - _cfg_node_ref(b, e.target.signature, e.target.node), - prune({"var": e.var}), - ) + for file_key, mod in app.symbol_table.items(): + for c in _walk_module_callables(mod): + if not c.id: + continue # unstamped callable — assign_ids must run first + owner = _sym(c.id) # the :PyCallable node, keyed by its can:// id + for local_key, node in (c.body or {}).items(): + span = node.span + ref = b.node( + ["PyCFGNode"], + "id", + _global_ordinal(c.id, local_key), + prune( + { + "kind": node.kind, + "start_line": span.start[0] if span else None, + "end_line": span.end[0] if span else None, + "_module": file_key, + } + ), + ) + b.edge("PY_HAS_CFG_NODE", owner, ref) + for e in c.cfg or []: + b.edge( + "PY_CFG_NEXT", + _cfg_ref(c.id, e.src), + _cfg_ref(c.id, e.dst), + {"kind": e.kind}, + ) + for e in c.cdg or []: + b.edge("PY_CDG", _cfg_ref(c.id, e.src), _cfg_ref(c.id, e.dst)) + for e in c.ddg or []: + b.edge( + "PY_DDG", + _cfg_ref(c.id, e.src), + _cfg_ref(c.id, e.dst), + prune({"var": e.var, "prov": list(e.prov) if e.prov else None}), + ) def _sym(can_id: str) -> NodeRef: diff --git a/test/conftest_v2.py b/test/conftest_v2.py index 39c178b..dccadaa 100644 --- a/test/conftest_v2.py +++ b/test/conftest_v2.py @@ -54,3 +54,12 @@ def assert_conformant(payload: dict, max_level: int) -> None: for lst in ("cfg", "cdg", "ddg", "summary"): for e in c.get(lst, []): assert e["src"] in node_ids and e["dst"] in node_ids, f"dangling {lst} in {c['id']}" + if max_level >= 3: + # L3 is syntactic-only: every def-use edge carries exactly ssa + # provenance (points-to provenance is the L4 delta). Dangling cfg/cdg/ddg + # endpoints are already rejected by the check above. + for mod, c in _iter_callables(app): + for e in c.get("ddg", []): + assert e.get("prov") == ["ssa"], ( + f"L3 ddg edge must have prov ['ssa'], got {e.get('prov')} in {c['id']}" + ) diff --git a/test/sample_graph_app.py b/test/sample_graph_app.py index 69f2221..1d52593 100644 --- a/test/sample_graph_app.py +++ b/test/sample_graph_app.py @@ -1,246 +1,115 @@ -"""A small, hand-built :class:`PyApplication` that exercises every Neo4j -projection path (module, class + inheritance + methods + attributes + inner -class, callable + decorators + call sites + local vars + inner callable, module -variables, imports, and a call graph with a resolved edge and a ghost edge). +"""A small v2 :class:`PyApplication` carried through the real L1 → L3 pipeline, +so the Neo4j projection tests exercise the whole overlay: a module, a class with +inheritance + a method + an attribute + an inner class, functions with +decorators + call sites + local variables, module variables, imports, a call +graph with a resolved edge and a ghost edge, and — new at level 3 — each +callable's CPG ``body``/``cfg``/``cdg``/``ddg``. -Built directly from the schema models so the Neo4j tests need neither Jedi nor a +The symbol table is built from a real (temporary) source file so +``build_function_pdgs`` can recover each callable's AST; ``assign_ids`` + +``populate_l1_body`` + ``build_function_pdgs`` (syntactic oracle) + ``emit_l3_body`` +then populate the v2 tree. The tests need neither a checked-in fixture tree nor a virtualenv — they stay fast and deterministic. + +``make_sample_app`` returns ``(app, sig_to_id)``: the projection +(``project(app, app_name, sig_to_id)``) needs the signature → ``can://`` id map +that ``assign_ids`` produced. """ from __future__ import annotations -from codeanalyzer.schema import ( - PyApplication, - PyCallable, - PyCFG, - PyCFGEdge, - PyClass, - PyClassAttribute, - PyComment, - PyExternalSymbol, - PyFunctionGraphs, - PyGraphNode, - PyImport, - PyModule, - PyParamNode, - PyPDG, - PyPDGEdge, - PyProgramGraphs, - PySDGEdge, - PySDGEndpoint, - PyVariableDeclaration, -) -from codeanalyzer.schema.py_schema import PyCallEdge, PyCallsite - - -def make_sample_app() -> PyApplication: - announce = PyCallable( - name="announce", - path="src/service.py", - signature="src.service.Service.announce", - comments=[PyComment(content="Announce something.", is_docstring=True)], - return_type="None", - code="def announce(self):\n ...", - start_line=10, - end_line=12, - code_start_line=10, - cyclomatic_complexity=1, - ) - inner = PyClass( - name="Inner", - signature="src.service.Service.Inner", - code="class Inner:\n ...", - start_line=14, - end_line=15, - ) - service = PyClass( - name="Service", - signature="src.service.Service", - comments=[PyComment(content="A service.", is_docstring=True)], - code="class Service(BaseService):\n ...", - base_classes=["src.service.BaseService"], - methods={"announce": announce}, - attributes={ - "name": PyClassAttribute(name="name", type="str", start_line=8, end_line=8) - }, - inner_classes={"Inner": inner}, - start_line=6, - end_line=15, - ) - base_service = PyClass( - name="BaseService", - signature="src.service.BaseService", - code="class BaseService:\n ...", - start_line=1, - end_line=4, - ) - helper = PyCallable( - name="helper", - path="src/service.py", - signature="src.service.helper", - decorators=["staticmethod"], - return_type="int", - code="def helper():\n Service().announce()\n requests.get(url)", - start_line=17, - end_line=20, - code_start_line=17, - cyclomatic_complexity=2, - call_sites=[ - PyCallsite( - method_name="announce", - receiver_expr="Service()", - receiver_type="src.service.Service", - callee_signature="src.service.Service.announce", - start_line=18, - start_column=4, - end_line=18, - end_column=22, - ) - ], - local_variables=[ - PyVariableDeclaration( - name="url", type="str", initializer="'x'", scope="function", - start_line=18, end_line=18, - ) - ], - ) - service_mod = PyModule( - file_path="src/service.py", - module_name="src.service", - imports=[PyImport(module="os", name="path", alias="p")], - classes={"Service": service, "BaseService": base_service}, - functions={"helper": helper}, - variables=[ - PyVariableDeclaration( - name="CONFIG", type="dict", initializer="{}", scope="module", - start_line=2, end_line=2, - ) - ], - content_hash="hash-service-v1", - last_modified=1.0, - file_size=100, - ) - util_mod = PyModule( - file_path="src/util.py", - module_name="src.util", - functions={ - "util_fn": PyCallable( - name="util_fn", - path="src/util.py", - signature="src.util.util_fn", - return_type="int", - code="def util_fn():\n return 1", - start_line=1, - end_line=2, - code_start_line=1, - cyclomatic_complexity=1, - ) - }, - content_hash="hash-util-v1", - last_modified=1.0, - file_size=40, - ) +import tempfile +from pathlib import Path +from typing import Dict, Tuple + +from codeanalyzer.dataflow.builder import build_function_pdgs, emit_l3_body +from codeanalyzer.dataflow.syntactic import SyntacticOracle +from codeanalyzer.schema import PyApplication, PyExternalSymbol +from codeanalyzer.schema.assign_ids import assign_ids +from codeanalyzer.schema.l1_body import populate_l1_body +from codeanalyzer.schema.py_schema import PyCallEdge +from codeanalyzer.syntactic_analysis.symbol_table_builder import SymbolTableBuilder + +# A tiny program with every symbol-table shape the projection walks, plus enough +# control flow (an ``if``) and def-use (``message``/``y``) that each recovered +# callable yields non-empty CFG / CDG / DDG at level 3. +_SOURCE = '''import os + +CONFIG = {} + + +def trace(fn): + return fn + + +class BaseService: + pass + + +class Service(BaseService): + name: str - call_graph = [ - # resolved edge — both endpoints live in the symbol table + def announce(self, flag): + message = build(flag) + if flag: + message = message + "!" + return message + + class Inner: + pass + + +@trace +def helper(flag): + svc = Service() + result = svc.announce(flag) + return result + + +@trace +def build(x): + y = x + return y +''' + + +def make_sample_app() -> Tuple[PyApplication, Dict[str, str]]: + """Build the v2 sample application with its level-3 CPG overlay emitted. + + Returns ``(app, sig_to_id)`` — the projection consumes both. + """ + workdir = Path(tempfile.mkdtemp(prefix="sample-graph-app-")) + source_file = workdir / "service.py" + source_file.write_text(_SOURCE, encoding="utf-8") + + module = SymbolTableBuilder(workdir, None).build_pymodule_from_file(source_file) + app = PyApplication(symbol_table={"service.py": module}) + + # A resolved call-graph edge (both endpoints declared) and a ghost edge whose + # target is a third-party member — materialized as a :PyExternal node. + app.call_graph = [ PyCallEdge( - source="src.service.helper", - target="src.service.Service.announce", + source="service.helper", + target="service.Service.announce", weight=1, provenance=["jedi"], ), - # ghost edge — target is third-party, materialized as an :PyExternal node PyCallEdge( - source="src.service.helper", - target="requests.get", + source="service.helper", + target="os.getcwd", weight=2, provenance=["jedi", "pycg"], ), ] + app.external_symbols = { + "os.getcwd": PyExternalSymbol(name="getcwd", module="os") + } - # A miniature level-3 section exercising every CPG row family: - # helper's CFG (entry → callsite stmt → exit), a CDG/DDG pair, its HRB - # parameter nodes, and PARAM_IN/PARAM_OUT/SUMMARY edges into announce. - helper_graphs = PyFunctionGraphs( - cfg=PyCFG( - nodes=[ - PyGraphNode(id=0, kind="entry", start_line=17, end_line=17), - PyGraphNode(id=1, kind="statement", start_line=18, end_line=18), - PyGraphNode(id=2, kind="exit", start_line=20, end_line=20), - ], - edges=[ - PyCFGEdge(source=0, target=1, kind="fallthrough"), - PyCFGEdge(source=1, target=2, kind="return"), - PyCFGEdge(source=1, target=2, kind="exception"), - ], - ), - pdg=PyPDG( - edges=[ - PyPDGEdge(source=0, target=1, type="CDG"), - PyPDGEdge(source=0, target=1, type="DDG", var="url"), - ] - ), - param_nodes=[ - PyParamNode(id=3, kind="formal_out", var="", start_line=20, end_line=20), - PyParamNode(id=4, kind="actual_in", var="self", call_node=1, start_line=18, end_line=18), - PyParamNode(id=5, kind="actual_out", var="", call_node=1, start_line=18, end_line=18), - ], - ) - announce_graphs = PyFunctionGraphs( - cfg=PyCFG( - nodes=[ - PyGraphNode(id=0, kind="entry", start_line=10, end_line=10), - PyGraphNode(id=1, kind="return", start_line=11, end_line=11), - PyGraphNode(id=2, kind="exit", start_line=12, end_line=12), - ], - edges=[ - PyCFGEdge(source=0, target=1, kind="fallthrough"), - PyCFGEdge(source=1, target=2, kind="return"), - ], - ), - pdg=PyPDG(edges=[PyPDGEdge(source=0, target=1, type="CDG")]), - param_nodes=[ - PyParamNode(id=3, kind="formal_in", var="self", start_line=10, end_line=10), - PyParamNode(id=4, kind="formal_out", var="", start_line=12, end_line=12), - ], - ) - program_graphs = PyProgramGraphs( - schema_version="1.0.0", - k_limit=3, - functions={ - "src.service.helper": helper_graphs, - "src.service.Service.announce": announce_graphs, - }, - sdg_edges=[ - PySDGEdge( - source=PySDGEndpoint(signature="src.service.helper", node=1), - target=PySDGEndpoint(signature="src.service.Service.announce", node=0), - type="CALL", - ), - PySDGEdge( - source=PySDGEndpoint(signature="src.service.helper", node=4), - target=PySDGEndpoint(signature="src.service.Service.announce", node=3), - type="PARAM_IN", - var="self", - ), - PySDGEdge( - source=PySDGEndpoint(signature="src.service.Service.announce", node=4), - target=PySDGEndpoint(signature="src.service.helper", node=5), - type="PARAM_OUT", - var="", - ), - PySDGEdge( - source=PySDGEndpoint(signature="src.service.helper", node=4), - target=PySDGEndpoint(signature="src.service.helper", node=5), - type="SUMMARY", - ), - ], + # Identity + L1 bodies, then the intraprocedural (syntactic) L3 overlay. + sig_to_id = assign_ids(app, "sample-app") + populate_l1_body(app) + infos, _func_asts = build_function_pdgs( + app, k=3, oracle_factory=lambda c: SyntacticOracle() ) + emit_l3_body(app, infos, sig_to_id, {"cfg", "dfg", "pdg"}) - return PyApplication( - symbol_table={"src/service.py": service_mod, "src/util.py": util_mod}, - call_graph=call_graph, - # The ghost edge's target (requests.get) is a library member, recorded as a - # first-class external symbol so the projection emits a :PyExternal for it. - external_symbols={"requests.get": PyExternalSymbol(name="get", module="requests")}, - program_graphs=program_graphs, - ) + return app, sig_to_id diff --git a/test/test_dataflow_cpg.py b/test/test_dataflow_cpg.py index 3f8d5c3..3156ce3 100644 --- a/test/test_dataflow_cpg.py +++ b/test/test_dataflow_cpg.py @@ -1,52 +1,45 @@ -"""Stage-8b gate: the CPG projection of the level-3 graphs. +"""Stage-3 (v2) gate: the CPG overlay projection of the level-3 graphs. -- PyCFGNode row count equals the JSON section's node count (CFG + parameter - nodes) — the contract's count-parity assertion; -- every PY_CFG_NEXT/PY_CDG/PY_DDG/PY_PARAM_IN/PY_PARAM_OUT/PY_SUMMARY edge - endpoint references an emitted PyCFGNode id (deferred-edge/no-dangling gate); +Projected off each callable's v2 ``body``/``cfg``/``cdg``/``ddg`` (populated by +``emit_l3_body`` at ``-a 3``), the Neo4j overlay must satisfy: + +- ``PyCFGNode`` row count equals the total number of ``body`` nodes across all + callables — the count-parity / two-projection-agreement assertion; +- every ``PY_CFG_NEXT``/``PY_CDG``/``PY_DDG`` edge endpoint that is a PyCFGNode + references an emitted PyCFGNode id (the no-dangling gate); +- every emitted PyCFGNode is owned by its callable via ``PY_HAS_CFG_NODE``; - the Cypher snapshot renders and contains the overlay's vocabulary. -Loading into a live Neo4j is exercised by the (container-gated) bolt tests; -these stay fast and deterministic. +Parameter/summary edges (``PY_PARAM_IN``/``PY_PARAM_OUT``/``PY_SUMMARY``) are an +L4/SDG concern and are intentionally absent here. Loading into a live Neo4j is +exercised by the (container-gated) bolt tests; these stay fast and deterministic. """ import pytest -pytest.skip( - "Deferred to Stage 3 v2 dataflow/neo4j test migration (see issues #73, #72). " - "Uses deleted graph models / pre-envelope analyze() shape.", - allow_module_level=True, -) - -from pathlib import Path -from codeanalyzer.core import Codeanalyzer from codeanalyzer.neo4j import project from codeanalyzer.neo4j.cypher import render_cypher -from codeanalyzer.options import AnalysisOptions +from codeanalyzer.semantic_analysis.call_graph import iter_callables_in_symbol_table -FIXTURE = Path(__file__).parent / "fixtures" / "single_functionalities" / "dataflow" +from sample_graph_app import make_sample_app -CPG_EDGE_TYPES = {"PY_CFG_NEXT", "PY_CDG", "PY_DDG", "PY_PARAM_IN", "PY_PARAM_OUT", "PY_SUMMARY"} +CPG_EDGE_TYPES = {"PY_CFG_NEXT", "PY_CDG", "PY_DDG"} @pytest.fixture(scope="module") -def level3_app(tmp_path_factory): - cache = tmp_path_factory.mktemp("dataflow-cpg-cache") - options = AnalysisOptions( - input=FIXTURE, analysis_level=3, no_venv=True, cache_dir=cache - ) - with Codeanalyzer(options) as analyzer: - return analyzer.analyze() +def sample(): + return make_sample_app() # (app, sig_to_id) @pytest.fixture(scope="module") -def rows(level3_app): - return project(level3_app, "dataflow-fixture") +def rows(sample): + app, sig_to_id = sample + return project(app, "dataflow-fixture", sig_to_id) -def test_cfg_node_count_matches_the_json_section(level3_app, rows): +def test_cfg_node_count_matches_the_body_section(sample, rows): + app, _sig_to_id = sample expected = sum( - len(fg.cfg.nodes if fg.cfg else []) + len(fg.param_nodes or []) - for fg in level3_app.program_graphs.functions.values() + len(c.body or {}) for c in iter_callables_in_symbol_table(app.symbol_table) ) emitted = [n for n in rows.nodes if "PyCFGNode" in n.labels] assert expected > 0 @@ -64,14 +57,14 @@ def test_no_dangling_cpg_edge_endpoints(rows): assert e.to_ref.value in cfg_ids, e -def test_every_callable_with_graphs_owns_its_cfg_nodes(level3_app, rows): +def test_every_callable_with_graphs_owns_its_cfg_nodes(rows): has_edges = [e for e in rows.edges if e.type == "PY_HAS_CFG_NODE"] owned = {e.to_ref.value for e in has_edges} cfg_ids = {n.value for n in rows.nodes if "PyCFGNode" in n.labels} assert owned == cfg_ids, "every CFGNode must be owned by its callable" -def test_cypher_snapshot_renders_the_overlay(level3_app, rows): +def test_cypher_snapshot_renders_the_overlay(rows): cypher = render_cypher(rows, "dataflow-fixture") assert ":PyCFGNode" in cypher for t in CPG_EDGE_TYPES: diff --git a/test/test_v2_cache.py b/test/test_v2_cache.py new file mode 100644 index 0000000..e877104 --- /dev/null +++ b/test/test_v2_cache.py @@ -0,0 +1,110 @@ +"""Cross-level cache safety. + +A cache is keyed on file hash/mtime/size only, so a run that reuses a cache +built at a *different* ``analysis_level`` must not serve stale content: an +``-a 3`` cache carries L3 ``body`` statement nodes (``@entry``/``@exit`` + +``line:col`` statements) and ``cfg``/``cdg``/``ddg`` edges, none of which +belong in an ``-a 1`` output. The loader rejects a level mismatch and rebuilds. +""" +from pathlib import Path + +from codeanalyzer.core import Codeanalyzer +from codeanalyzer.options import AnalysisOptions + + +def _all_callables(app): + """Yield every PyCallable in the application tree (pydantic objects).""" + def walk_callable(c): + yield c + for ic in (c.inner_callables or {}).values(): + yield from walk_callable(ic) + for cl in (c.inner_classes or {}).values(): + yield from walk_class(cl) + + def walk_class(cl): + for m in (cl.methods or {}).values(): + yield from walk_callable(m) + for ic in (cl.inner_classes or {}).values(): + yield from walk_class(ic) + + for mod in app.symbol_table.values(): + for fn in (mod.functions or {}).values(): + yield from walk_callable(fn) + for cl in (mod.classes or {}).values(): + yield from walk_class(cl) + + +_SRC = ( + "def f(a):\n" + " b = a\n" + " g(b)\n" + " return b\n" + "\n" + "def g(x):\n" + " return x\n" +) + + +def test_cross_level_cache_no_l3_leak(tmp_path: Path): + """Build at -a 3 (bodies + cfg), then reuse the SAME cache dir at -a 1. + The L1 result must not carry any L3-only content.""" + proj = tmp_path / "proj" + proj.mkdir() + (proj / "m.py").write_text(_SRC, encoding="utf-8") + + # Build at L3 into a shared cache dir (writes bodies + cfg into the cache). + opts3 = AnalysisOptions( + input=proj, cache_dir=tmp_path, no_venv=True, + analysis_level=3, graphs="cfg,dfg,pdg", + ) + with Codeanalyzer(opts3) as an: + res3 = an.analyze() + assert res3.max_level == 3 + # Sanity: the L3 build really did populate cfg/body statements (so the + # cross-level assertion below is meaningful, not vacuous). + assert any(c.cfg for c in _all_callables(res3.application)), "L3 build produced no cfg" + + # Reuse the SAME cache dir at L1. + opts1 = AnalysisOptions(input=proj, cache_dir=tmp_path, no_venv=True, analysis_level=1) + with Codeanalyzer(opts1) as an: + res1 = an.analyze() + assert res1.max_level == 1 + + # No L3 leak: L1 callables carry no cfg/cdg/ddg edges and their body holds + # only `call` nodes (no @entry/@exit bookends, no statement nodes). + for c in _all_callables(res1.application): + assert not c.cfg, f"L3 cfg leaked into L1 output: {c.id}" + assert not c.cdg, f"L3 cdg leaked into L1 output: {c.id}" + assert not c.ddg, f"L3 ddg leaked into L1 output: {c.id}" + for key, node in (c.body or {}).items(): + assert not key.startswith("@"), f"L3 body bookend leaked into L1: {c.id} {key}" + assert node.kind == "call", ( + f"non-call L3 body node leaked into L1: {c.id} {key} kind={node.kind}" + ) + + +def test_same_level_cache_reuse_still_works(tmp_path: Path): + """Same-level reuse is a genuine cache hit that still yields a conformant + L1 envelope (the level guard is additive; it must not break reuse).""" + proj = tmp_path / "proj" + proj.mkdir() + (proj / "m.py").write_text("def f(a):\n return a\n", encoding="utf-8") + + opts = AnalysisOptions(input=proj, cache_dir=tmp_path, no_venv=True, analysis_level=1) + with Codeanalyzer(opts) as an: + first = an.analyze() + assert first.schema_version == "2.0.0" + + an2 = Codeanalyzer(opts) + cache_file = an2.cache_dir / "analysis_cache.json" + assert cache_file.exists() + # Loader returns the cached envelope for a same-level request (cache hit, + # no rebuild forced). + loaded = an2._load_pyapplication_from_cache(cache_file) + assert loaded is not None and loaded.max_level == 1 + + # A full second run still produces a conformant L1 envelope. + with Codeanalyzer(opts) as an: + second = an.analyze() + assert second.schema_version == "2.0.0" + assert second.max_level == 1 diff --git a/test/test_v2_l3.py b/test/test_v2_l3.py new file mode 100644 index 0000000..74c05ba --- /dev/null +++ b/test/test_v2_l3.py @@ -0,0 +1,161 @@ +import textwrap +from pathlib import Path + +from codeanalyzer.__main__ import app +from codeanalyzer.dataflow.builder import build_function_pdgs, emit_l3_body +from codeanalyzer.dataflow.identity import IdentityMap +from codeanalyzer.dataflow.syntactic import SyntacticOracle +from codeanalyzer.schema.assign_ids import assign_ids +from codeanalyzer.schema.l1_body import populate_l1_body +from codeanalyzer.schema.py_schema import PyApplication +from codeanalyzer.syntactic_analysis.symbol_table_builder import SymbolTableBuilder + +class _Node: + def __init__(self, id, start_line, start_column, kind): + self.id, self.start_line, self.start_column, self.kind = id, start_line, start_column, kind + +class _CFG: + def __init__(self, nodes, entry_id, exit_id): + self._n = {n.id: n for n in nodes}; self.nodes = nodes + self.entry_id, self.exit_id = entry_id, exit_id + def node_by_id(self, i): return self._n[i] + +class _PDG: + def __init__(self, cfg): self.cfg = cfg + +def test_local_and_global_ids_for_entry_exit_and_statements(): + nodes = [_Node(0, 1, 0, "entry"), _Node(1, 2, 4, "statement"), _Node(2, 3, 4, "exit")] + pdg = _PDG(_CFG(nodes, entry_id=0, exit_id=2)) + im = IdentityMap.for_function("can://python/app/m.py/f()", pdg) + # LOCAL ids: intra-callable keys (match l1_body's "line:col" format). + assert im.local(0) == "@entry" + assert im.local(1) == "2:4" + assert im.local(2) == "@exit" + # GLOBAL id: fully addressable form for Neo4j / cross-callable use. + assert im.global_id(0) == "can://python/app/m.py/f()@entry" # entry (single @, no double) + assert im.global_id(1) == "can://python/app/m.py/f()@2:4" # statement + assert im.global_id(2) == "can://python/app/m.py/f()@exit" # exit + assert set(im.node_ids()) == {0, 1, 2} + + +def test_syntactic_oracle_only_identity_aliases(): + o = SyntacticOracle() + assert o.may_alias("x.f", "x.f") is True + assert o.may_alias("x.f", "y.f") is False + assert o.may_alias("a", "b") is False + + +def test_emit_l3_populates_body_and_cfg(tmp_path: Path): + f = tmp_path / "m.py" + f.write_text("def f(a):\n b = a\n g(b)\n return b\n", encoding="utf-8") + mod = SymbolTableBuilder(tmp_path, None).build_pymodule_from_file(f) + app = PyApplication(symbol_table={"m.py": mod}) + sig_to_id = assign_ids(app, "app") + # L1 materializes the `g(b)` call as a LOCAL "line:col" body node; simulate + # the L2 callee refinement so we can prove L3 preserves it (no re-key). + populate_l1_body(app) + fn = next(iter(mod.functions.values())) + call_key = "3:4" + assert fn.body[call_key].kind == "call" + fn.body[call_key].callee = "m.g" + + infos, _func_asts = build_function_pdgs( + app, k=3, oracle_factory=lambda c: SyntacticOracle() + ) + emit_l3_body(app, infos, sig_to_id, graphs={"cfg", "dfg", "pdg"}) + + # body keys are LOCAL: "@entry"/"@exit" bookends + bare "line:col" stmts, + # never the full "@..." form. + assert "@entry" in fn.body + assert "@exit" in fn.body + assert any(k not in ("@entry", "@exit") and ":" in k for k in fn.body) + assert not any(k.startswith("can://") for k in fn.body) + # the L1 call node is PRESERVED under its local key (not duplicated, not + # re-keyed): still kind=="call" with its L2-resolved callee. + assert fn.body[call_key].kind == "call" + assert fn.body[call_key].callee == "m.g" + + assert len(fn.cfg) > 0 + # every cfg endpoint resolves to a (local) body node id + body_ids = set(fn.body) + for e in fn.cfg: + assert e.src in body_ids and e.dst in body_ids + # ddg (if any) carries ssa provenance + for e in fn.ddg: + assert e.prov == ["ssa"] + + +def test_build_function_pdgs_returns_pdg_per_callable(tmp_path: Path): + f = tmp_path / "m.py" + f.write_text(textwrap.dedent("def f(a):\n b = a\n return b\n"), encoding="utf-8") + mod = SymbolTableBuilder(tmp_path, None).build_pymodule_from_file(f) + app = PyApplication(symbol_table={"m.py": mod}) + infos, func_asts = build_function_pdgs( + app, k=3, oracle_factory=lambda c: SyntacticOracle() + ) + sig = next(iter(mod.functions.values())).signature + assert sig in infos + assert infos[sig].pdg.cfg.entry_id is not None + + +def test_cli_graphs_sdg_requires_a4(cli_runner, tmp_path): + """CLI: requesting --graphs sdg should error with exit code 2, regardless of -a level.""" + proj = tmp_path / "proj" + proj.mkdir() + (proj / "main.py").write_text("def f():\n pass\n") + + out = tmp_path / "out1" + result = cli_runner.invoke( + app, + [ + "--input", str(proj), + "--analysis-level", "3", + "--graphs", "sdg", + "--no-venv", + "--output", str(out), + ], + env={"NO_COLOR": "1", "TERM": "dumb"}, + ) + assert result.exit_code == 2, f"Expected exit code 2, got {result.exit_code}. Output: {result.output}" + assert "sdg requires -a 4" in result.output or "not available yet" in result.output + + +def test_cli_graphs_cfg_pdg_at_a3_succeeds(cli_runner, tmp_path): + """CLI: -a 3 --graphs cfg,pdg should succeed (exit 0).""" + proj = tmp_path / "proj" + proj.mkdir() + (proj / "main.py").write_text("def f():\n pass\n") + + out = tmp_path / "out2" + result = cli_runner.invoke( + app, + [ + "--input", str(proj), + "--analysis-level", "3", + "--graphs", "cfg,pdg", + "--no-venv", + "--output", str(out), + ], + env={"NO_COLOR": "1", "TERM": "dumb"}, + ) + assert result.exit_code == 0, f"Expected exit code 0, got {result.exit_code}. Output: {result.output}" + + +def test_cli_graphs_default_at_a3_succeeds(cli_runner, tmp_path): + """CLI: -a 3 with default graphs (no --graphs flag) should succeed (exit 0).""" + proj = tmp_path / "proj" + proj.mkdir() + (proj / "main.py").write_text("def f():\n pass\n") + + out = tmp_path / "out3" + result = cli_runner.invoke( + app, + [ + "--input", str(proj), + "--analysis-level", "3", + "--no-venv", + "--output", str(out), + ], + env={"NO_COLOR": "1", "TERM": "dumb"}, + ) + assert result.exit_code == 0, f"Expected exit code 0, got {result.exit_code}. Output: {result.output}" diff --git a/test/test_v2_l3_slice.py b/test/test_v2_l3_slice.py new file mode 100644 index 0000000..5a44fcb --- /dev/null +++ b/test/test_v2_l3_slice.py @@ -0,0 +1,42 @@ +import ast +from pathlib import Path + +from codeanalyzer.dataflow.builder import build_function_pdgs +from codeanalyzer.dataflow.identity import IdentityMap +from codeanalyzer.dataflow.pdg import intraprocedural_backward_slice +from codeanalyzer.dataflow.syntactic import SyntacticOracle +from codeanalyzer.schema.py_schema import PyApplication +from codeanalyzer.syntactic_analysis.symbol_table_builder import SymbolTableBuilder + + +def test_backward_slice_equals_hand_computed_set(tmp_path: Path): + # A tiny fixture with a KNOWN def-use chain: return c <- (c = b) <- (b = a). + f = tmp_path / "m.py" + f.write_text("def f(a):\n b = a\n c = b\n return c\n", encoding="utf-8") + mod = SymbolTableBuilder(tmp_path, None).build_pymodule_from_file(f) + app = PyApplication(symbol_table={"m.py": mod}) + + infos, _func_asts = build_function_pdgs( + app, k=3, oracle_factory=lambda c: SyntacticOracle() + ) + fn = next(iter(mod.functions.values())) + callable_id = fn.signature + pdg = infos[callable_id].pdg + + # Criterion node: the `return c` statement, located by its AST node in the + # CFG (there is exactly one Return in the fixture). + criterion = next(n.id for n in pdg.cfg.nodes if isinstance(n.ast_node, ast.Return)) + + slice_ids = intraprocedural_backward_slice(pdg, criterion) + + im = IdentityMap.for_function(callable_id, pdg) + local_ids = {im.local(i) for i in slice_ids} + + # Hand-computed slice over ordinal ids, exact: + # 4:4 `return c` — the criterion itself + # 3:4 `c = b` — data dependence (reads c defined here) + # 2:4 `b = a` — data dependence (reads b defined here) + # @entry — control-region root of every unconditional statement + # AND the def site of the parameter `a` that `b = a` reads + # @exit carries no dependence into the criterion, so it is NOT in the slice. + assert local_ids == {"@entry", "2:4", "3:4", "4:4"} diff --git a/test/test_v2_superset.py b/test/test_v2_superset.py index 935b0f8..92af558 100644 --- a/test/test_v2_superset.py +++ b/test/test_v2_superset.py @@ -52,3 +52,29 @@ def test_l1_subset_of_l2(tmp_path): c2 = next(c for c in _callables(a2) if c["id"] == c1["id"]) assert set(c1.get("body", {})) <= set(c2.get("body", {})), \ f"L2 dropped a body node from {c1['id']}" + + +def test_l2_subset_of_l3(tmp_path): + proj = tmp_path / "proj" + proj.mkdir() + # Multi-statement fixture with a bare call: the `g(b)` expression statement + # is materialized at L1/L2 as a `call` body node keyed by its local + # "line:col"; L3 lands its CFG statement node on the SAME key, so the fact + # must survive (not be dropped or re-keyed). + (proj / "m.py").write_text( + "def g(x):\n return x\ndef f(a):\n b = a\n g(b)\n return b\n", + encoding="utf-8", + ) + l2, l3 = _run(proj, 2), _run(proj, 3) + assert_conformant(l2, max_level=2) + assert_conformant(l3, max_level=3) + a2, a3 = l2["application"], l3["application"] + # every callable id present at L2 is present at L3 + ids2 = {c["id"] for c in _callables(a2)} + ids3 = {c["id"] for c in _callables(a3)} + assert ids2 <= ids3, "L3 dropped a callable present at L2" + # L3 only ADDS body statements (+ @entry/@exit); every L2 body key survives. + for c2 in _callables(a2): + c3 = next(c for c in _callables(a3) if c["id"] == c2["id"]) + assert set(c2.get("body", {})) <= set(c3.get("body", {})), \ + f"L3 dropped a body node from {c2['id']}" diff --git a/test/test_v2_two_projection_agreement.py b/test/test_v2_two_projection_agreement.py index edd9851..237c4db 100644 --- a/test/test_v2_two_projection_agreement.py +++ b/test/test_v2_two_projection_agreement.py @@ -1,8 +1,12 @@ from codeanalyzer.schema.assign_ids import assign_ids -from codeanalyzer.neo4j.project import project +from codeanalyzer.neo4j.project import project, _global_ordinal from codeanalyzer.schema.py_schema import ( PyApplication, PyModule, PyClass, PyCallable, PyCallsite, ) +from codeanalyzer.dataflow.builder import build_function_pdgs, emit_l3_body +from codeanalyzer.dataflow.syntactic import SyntacticOracle +from codeanalyzer.schema.l1_body import populate_l1_body +from codeanalyzer.syntactic_analysis.symbol_table_builder import SymbolTableBuilder def test_neo4j_callable_key_equals_json_id(): @@ -32,3 +36,58 @@ def test_py_resolves_to_edge_targets_declared_callee_by_can_id(): # the callsite must resolve to g's can:// id — edge kept, not dropped assert any(e.to_ref.value == sig_to_id["m.g"] for e in resolves), \ "PY_RESOLVES_TO must target the declared callee by can:// id" + + +# ---------------------------------------------------------------------------------------------- +# Level-3 CPG overlay: PyCFGNode merge keys are the callable can:// id + local body key, +# proving the Neo4j projection and the JSON `body`/`cfg` land on the same node identity. +# ---------------------------------------------------------------------------------------------- + + +def _build_l3_app(tmp_path): + """A one-function project carried through the real L1 → L3 pipeline (mirrors + test_v2_l3) so each callable has a populated ``body``/``cfg``/``cdg``/``ddg``.""" + f = tmp_path / "m.py" + f.write_text("def f(a):\n b = a\n g(b)\n return b\n", encoding="utf-8") + mod = SymbolTableBuilder(tmp_path, None).build_pymodule_from_file(f) + app = PyApplication(symbol_table={"m.py": mod}) + sig_to_id = assign_ids(app, "app") + populate_l1_body(app) + infos, _func_asts = build_function_pdgs( + app, k=3, oracle_factory=lambda c: SyntacticOracle() + ) + emit_l3_body(app, infos, sig_to_id, graphs={"cfg", "dfg", "pdg"}) + return app, sig_to_id, mod + + +def test_cpg_overlay_pycfgnode_keys_equal_callable_id_plus_body_key(tmp_path): + app, sig_to_id, mod = _build_l3_app(tmp_path) + c = next(iter(mod.functions.values())) + assert c.body, "precondition: L3 must populate the callable body" + + rows = project(app, "app", sig_to_id) + + # (a) two-projection agreement: the set of PyCFGNode merge values for this + # callable == { @ }. + expected = {_global_ordinal(c.id, k) for k in c.body} + emitted = { + n.value + for n in rows.nodes + if n.labels[0] == "PyCFGNode" and n.value.startswith(c.id + "@") + } + assert emitted == expected, f"PyCFGNode keys {emitted} != body ordinals {expected}" + + # (b) a PY_CFG_NEXT edge whose endpoints are two of those global ids. + cfg_next = [e for e in rows.edges if e.type == "PY_CFG_NEXT"] + assert any( + e.from_ref.value in expected and e.to_ref.value in expected for e in cfg_next + ), "expected a PY_CFG_NEXT edge over the callable's CFG-node ids" + + # (c) PY_HAS_CFG_NODE from the callable to its @entry CFG node. + entry_id = _global_ordinal(c.id, "@entry") + assert any( + e.type == "PY_HAS_CFG_NODE" + and e.from_ref.value == c.id + and e.to_ref.value == entry_id + for e in rows.edges + ), "expected PY_HAS_CFG_NODE from the callable to its @entry CFG node"