diff --git a/codeanalyzer/core.py b/codeanalyzer/core.py index 4909651..48431f5 100644 --- a/codeanalyzer/core.py +++ b/codeanalyzer/core.py @@ -20,6 +20,8 @@ ) from codeanalyzer.schema.assign_ids import assign_ids from codeanalyzer.schema.l1_body import populate_l1_body +from codeanalyzer.schema.l2_callees import backfill_callees +from codeanalyzer.schema.call_graph_ids import reidentify_call_graph from codeanalyzer.schema.py_schema import PyCallEdge from codeanalyzer.semantic_analysis.call_graph import ( filter_external_edges, @@ -459,8 +461,12 @@ def analyze(self) -> Analysis: .build() ) - assign_ids(app, self.options.app_name or self.project_dir.name) + app_name = self.options.app_name or self.project_dir.name + sig_to_id = assign_ids(app, app_name) populate_l1_body(app) + if self.analysis_level >= 2: + 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+ diff --git a/codeanalyzer/neo4j/project.py b/codeanalyzer/neo4j/project.py index cf2ce6d..df76b38 100644 --- a/codeanalyzer/neo4j/project.py +++ b/codeanalyzer/neo4j/project.py @@ -57,15 +57,18 @@ def project(app: PyApplication, app_name: str, sig_to_id: dict) -> GraphRows: ["PyApplication"], "name", app_name, {"schema_version": SCHEMA_VERSION} ) + # Endpoints listed in app.external_symbols become :PyExternal ghost nodes; the + # rest are declared :PySymbol nodes emitted here (keyed by their can:// id, + # resolved through ``sig_to_id``). Both the module-body projection (for + # PY_EXTENDS / PY_RESOLVES_TO) and the PY_CALLS twin below share this split. + externals = app.external_symbols or {} + for file_key, mod in app.symbol_table.items(): mod_ref = b.node(["PyModule"], "id", mod.id, _module_props(mod, file_key)) b.edge("PY_HAS_MODULE", app_ref, mod_ref) - _project_module_body(b, file_key, mod_ref, mod) + _project_module_body(b, file_key, mod_ref, mod, externals, sig_to_id) - # The aggregated :PY_CALLS twin. Endpoints listed in app.external_symbols become - # :PyExternal ghost nodes; the rest are declared :PySymbol nodes already emitted - # (keyed by their can:// id, resolved through ``sig_to_id``). - externals = app.external_symbols or {} + # The aggregated :PY_CALLS twin. for e in app.call_graph: src = _call_endpoint(b, e.source, externals, sig_to_id) tgt = _call_endpoint(b, e.target, externals, sig_to_id) @@ -175,6 +178,16 @@ def _sym(can_id: str) -> NodeRef: return NodeRef("PySymbol", "id", can_id) +def _symbol_ref(signature: str, externals: dict, sig_to_id: dict) -> NodeRef: + """Resolve a call/inheritance target to the NodeRef under which it was (or + will be) emitted: a declared symbol by its can:// id, otherwise a + signature-keyed :PySymbol (external ghost).""" + can_id = sig_to_id.get(signature) + if can_id is not None: + return NodeRef("PySymbol", "id", can_id) + return NodeRef("PySymbol", "signature", signature) + + def _call_endpoint( b: RowBuilder, signature: str, externals: dict, sig_to_id: dict ) -> NodeRef: @@ -212,12 +225,13 @@ def _call_endpoint( def _project_module_body( - b: RowBuilder, file_key: str, mod_ref: NodeRef, mod: PyModule + b: RowBuilder, file_key: str, mod_ref: NodeRef, mod: PyModule, + externals: dict, sig_to_id: dict, ) -> None: for fn in (mod.functions or {}).values(): - _project_callable(b, file_key, mod_ref, "PY_DECLARES", fn) + _project_callable(b, file_key, mod_ref, "PY_DECLARES", fn, externals, sig_to_id) for cl in (mod.classes or {}).values(): - _project_class(b, file_key, mod_ref, "PY_DECLARES", cl) + _project_class(b, file_key, mod_ref, "PY_DECLARES", cl, externals, sig_to_id) for v in mod.variables or []: _project_variable(b, file_key, mod_ref, file_key, v) _project_imports(b, mod_ref, mod) @@ -256,7 +270,8 @@ def _project_imports(b: RowBuilder, mod_ref: NodeRef, mod: PyModule) -> None: def _project_class( - b: RowBuilder, file_key: str, parent: NodeRef, parent_rel: str, cl: PyClass + b: RowBuilder, file_key: str, parent: NodeRef, parent_rel: str, cl: PyClass, + externals: dict, sig_to_id: dict, ) -> None: ref = b.node( ["PySymbol", "PyClass"], "id", cl.id, _class_props(cl, file_key) @@ -264,18 +279,20 @@ def _project_class( b.edge(parent_rel, parent, ref) for base in cl.base_classes or []: - b.edge_to_symbol("PY_EXTENDS", ref, base) + if base: + b.edge_to_symbol("PY_EXTENDS", ref, _symbol_ref(base, externals, sig_to_id)) for m in (cl.methods or {}).values(): - _project_callable(b, file_key, ref, "PY_HAS_METHOD", m) + _project_callable(b, file_key, ref, "PY_HAS_METHOD", m, externals, sig_to_id) for a in (cl.attributes or {}).values(): _project_attribute(b, file_key, ref, cl.signature, a) for ic in (cl.inner_classes or {}).values(): - _project_class(b, file_key, ref, "PY_DECLARES", ic) + _project_class(b, file_key, ref, "PY_DECLARES", ic, externals, sig_to_id) def _project_callable( - b: RowBuilder, file_key: str, owner: NodeRef, owner_rel: str, c: PyCallable + b: RowBuilder, file_key: str, owner: NodeRef, owner_rel: str, c: PyCallable, + externals: dict, sig_to_id: dict, ) -> None: ref = b.node( ["PySymbol", "PyCallable"], @@ -296,14 +313,17 @@ def _project_callable( cs = b.node(["PyCallSite"], "id", cs_id, _call_site_props(s, file_key)) b.edge("PY_HAS_CALLSITE", ref, cs) if s.callee_signature: - b.edge_to_symbol("PY_RESOLVES_TO", cs, s.callee_signature) + b.edge_to_symbol( + "PY_RESOLVES_TO", cs, + _symbol_ref(s.callee_signature, externals, sig_to_id), + ) for v in c.local_variables or []: _project_variable(b, file_key, ref, c.signature, v) for ic in (c.inner_callables or {}).values(): - _project_callable(b, file_key, ref, "PY_DECLARES", ic) + _project_callable(b, file_key, ref, "PY_DECLARES", ic, externals, sig_to_id) for cl in (c.inner_classes or {}).values(): - _project_class(b, file_key, ref, "PY_DECLARES", cl) + _project_class(b, file_key, ref, "PY_DECLARES", cl, externals, sig_to_id) def _project_attribute( diff --git a/codeanalyzer/neo4j/rows.py b/codeanalyzer/neo4j/rows.py index cbc381f..f8a81bd 100644 --- a/codeanalyzer/neo4j/rows.py +++ b/codeanalyzer/neo4j/rows.py @@ -110,20 +110,15 @@ def edge(self, type_: str, from_ref: NodeRef, to_ref: NodeRef, props: Optional[P self._edges.append(EdgeRow(type_, from_ref, to_ref, dict(props or {}))) def edge_to_symbol( - self, type_: str, from_ref: NodeRef, target_signature: str, props: Optional[Props] = None + self, type_: str, from_ref: NodeRef, target_ref: NodeRef, props: Optional[Props] = None ) -> None: """An edge to a ``:PySymbol`` target that may be external/library code not - present in the graph. Deferred and kept only if the target signature was - actually emitted as a node — so PY_EXTENDS / PY_RESOLVES_TO never dangle (the - string fallback lives on the source node's props).""" - self._deferred.append( - EdgeRow( - type_, - from_ref, - NodeRef("PySymbol", "signature", target_signature), - dict(props or {}), - ) - ) + present in the graph. The target is an already-resolved :class:`NodeRef` + (a declared symbol by its can:// id, or a signature-keyed external ghost). + Deferred and kept only if that ``(label, value)`` was actually emitted as a + node — so PY_EXTENDS / PY_RESOLVES_TO never dangle (the string fallback lives + on the source node's props).""" + self._deferred.append(EdgeRow(type_, from_ref, target_ref, dict(props or {}))) def has_key(self, label: str, value: str) -> bool: """Whether a node with this ``(merge_label, value)`` identity was emitted.""" diff --git a/codeanalyzer/schema/call_graph_ids.py b/codeanalyzer/schema/call_graph_ids.py new file mode 100644 index 0000000..ff3a24b --- /dev/null +++ b/codeanalyzer/schema/call_graph_ids.py @@ -0,0 +1,12 @@ +"""Re-identify call-graph edge endpoints onto canonical can:// ids so the JSON +call_graph agrees with the Neo4j PY_CALLS projection. Declared endpoints map +through sig_to_id; external/library endpoints keep their dotted signature +(they have no can:// id).""" +from __future__ import annotations +from codeanalyzer.schema.py_schema import PyApplication + + +def reidentify_call_graph(app: PyApplication, sig_to_id: dict) -> None: + for edge in app.call_graph or []: + edge.source = sig_to_id.get(edge.source, edge.source) + edge.target = sig_to_id.get(edge.target, edge.target) diff --git a/codeanalyzer/schema/l2_callees.py b/codeanalyzer/schema/l2_callees.py new file mode 100644 index 0000000..8428bf9 --- /dev/null +++ b/codeanalyzer/schema/l2_callees.py @@ -0,0 +1,36 @@ +"""L2 refinement: fill each L1 `call` body node's `callee` (null→id) from the +call site's resolved signature — the one sanctioned value change. A declared +target becomes its can:// id; an external/library target keeps its dotted +signature; an unresolved call site leaves `callee` absent.""" +from __future__ import annotations +from codeanalyzer.schema.py_schema import PyApplication, PyClass, PyCallable + + +def _do_callable(c: PyCallable, sig_to_id: dict) -> None: + for cs in c.call_sites or []: + if cs.callee_signature is None: + continue + key = f"{cs.start_line}:{cs.start_column}" + node = c.body.get(key) + if node is None or node.kind != "call": + continue + node.callee = sig_to_id.get(cs.callee_signature, cs.callee_signature) + for ic in (c.inner_callables or {}).values(): + _do_callable(ic, sig_to_id) + for icl in (c.inner_classes or {}).values(): + _do_class(icl, sig_to_id) + + +def _do_class(cl: PyClass, sig_to_id: dict) -> None: + for m in (cl.methods or {}).values(): + _do_callable(m, sig_to_id) + for ic in (cl.inner_classes or {}).values(): + _do_class(ic, sig_to_id) + + +def backfill_callees(app: PyApplication, sig_to_id: dict) -> None: + for mod in app.symbol_table.values(): + for fn in (mod.functions or {}).values(): + _do_callable(fn, sig_to_id) + for cl in (mod.classes or {}).values(): + _do_class(cl, sig_to_id) diff --git a/test/conftest_v2.py b/test/conftest_v2.py index 3f8ae70..39c178b 100644 --- a/test/conftest_v2.py +++ b/test/conftest_v2.py @@ -43,8 +43,12 @@ def assert_conformant(payload: dict, max_level: int) -> None: text = mod["source"].encode("utf-8")[lo:hi].decode("utf-8") assert text.lstrip().startswith(("def ", "async def ", "@")), f"{c['id']} span mismatch" for node in c.get("body", {}).values(): - if node.get("kind") == "call" and max_level >= 2: - assert node.get("callee") is None or isinstance(node["callee"], str) + if node.get("kind") == "call" and "callee" in node: + assert isinstance(node["callee"], str), "resolved callee must be a string id" + if max_level >= 2: + for e in app.get("call_graph", []): + assert isinstance(e["source"], str), f"call_graph edge source must be string: {e}" + assert isinstance(e["target"], str), f"call_graph edge target must be string: {e}" for mod, c in _iter_callables(app): node_ids = set(c.get("body", {}).keys()) for lst in ("cfg", "cdg", "ddg", "summary"): diff --git a/test/test_v2_l2.py b/test/test_v2_l2.py new file mode 100644 index 0000000..8c966cb --- /dev/null +++ b/test/test_v2_l2.py @@ -0,0 +1,43 @@ +from codeanalyzer.schema.l2_callees import backfill_callees +from codeanalyzer.schema.call_graph_ids import reidentify_call_graph +from codeanalyzer.schema.py_schema import ( + PyApplication, PyModule, PyCallable, PyCallsite, BodyNode, PyCallEdge, +) + +def _app_with_one_call(callee_sig): + cs = PyCallsite(method_name="g", start_line=2, start_column=4, end_line=2, + end_column=7, callee_signature=callee_sig) + fn = PyCallable(name="f", path="m.py", signature="m.f", call_sites=[cs], + body={"2:4": BodyNode(kind="call", callee=None)}) + mod = PyModule(file_path="m.py", module_name="m", functions={"f": fn}) + return PyApplication(symbol_table={"m.py": mod}), fn + +def test_declared_callee_resolves_to_can_id(): + app, fn = _app_with_one_call("m.g") + backfill_callees(app, {"m.g": "can://python/app/m.py/g()"}) + assert fn.body["2:4"].callee == "can://python/app/m.py/g()" + +def test_external_callee_keeps_dotted_signature(): + app, fn = _app_with_one_call("requests.get") + backfill_callees(app, {"m.g": "can://python/app/m.py/g()"}) # requests.get not declared + assert fn.body["2:4"].callee == "requests.get" + +def test_unresolved_callsite_leaves_callee_absent(): + app, fn = _app_with_one_call(None) + backfill_callees(app, {}) + assert fn.body["2:4"].callee is None + +def test_call_graph_endpoints_reidentified(): + edge = PyCallEdge(source="m.f", target="m.g") + app = PyApplication(symbol_table={}, call_graph=[edge]) + reidentify_call_graph(app, {"m.f": "can://python/app/m.py/f()", + "m.g": "can://python/app/m.py/g()"}) + assert app.call_graph[0].source == "can://python/app/m.py/f()" + assert app.call_graph[0].target == "can://python/app/m.py/g()" + +def test_call_graph_external_target_unchanged(): + edge = PyCallEdge(source="m.f", target="requests.get") + app = PyApplication(symbol_table={}, call_graph=[edge]) + reidentify_call_graph(app, {"m.f": "can://python/app/m.py/f()"}) + assert app.call_graph[0].source == "can://python/app/m.py/f()" + assert app.call_graph[0].target == "requests.get" diff --git a/test/test_v2_superset.py b/test/test_v2_superset.py new file mode 100644 index 0000000..935b0f8 --- /dev/null +++ b/test/test_v2_superset.py @@ -0,0 +1,54 @@ +import json +import subprocess +import sys +from pathlib import Path + +from conftest_v2 import assert_conformant + + +def _run(proj, level): + out = subprocess.run( + [sys.executable, "-m", "codeanalyzer", "-i", str(proj), "-a", str(level), "--no-venv"], + capture_output=True, text=True, check=True, + ).stdout + return json.loads(out) + + +def _callables(app): + def wc(c): + yield c + for ic in (c.get("inner_callables") or {}).values(): + yield from wc(ic) + for cl in (c.get("inner_classes") or {}).values(): + yield from wcl(cl) + + def wcl(cl): + for m in (cl.get("methods") or {}).values(): + yield from wc(m) + for ic in (cl.get("inner_classes") or {}).values(): + yield from wcl(ic) + + for mod in app["symbol_table"].values(): + for fn in (mod.get("functions") or {}).values(): + yield from wc(fn) + for cl in (mod.get("classes") or {}).values(): + yield from wcl(cl) + + +def test_l1_subset_of_l2(tmp_path): + proj = tmp_path / "proj" + proj.mkdir() + (proj / "m.py").write_text("def g():\n return 1\ndef f():\n return g()\n", encoding="utf-8") + l1, l2 = _run(proj, 1), _run(proj, 2) + assert_conformant(l1, max_level=1) + assert_conformant(l2, max_level=2) + a1, a2 = l1["application"], l2["application"] + # every module + callable id present at L1 is present at L2 + ids1 = {c["id"] for c in _callables(a1)} + ids2 = {c["id"] for c in _callables(a2)} + assert ids1 <= ids2, "L2 dropped a callable present at L1" + # every L1 body node key is present at L2 (callee may refine null→id) + for c1 in _callables(a1): + 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']}" diff --git a/test/test_v2_two_projection_agreement.py b/test/test_v2_two_projection_agreement.py index 9efb017..edd9851 100644 --- a/test/test_v2_two_projection_agreement.py +++ b/test/test_v2_two_projection_agreement.py @@ -1,6 +1,8 @@ from codeanalyzer.schema.assign_ids import assign_ids from codeanalyzer.neo4j.project import project -from codeanalyzer.schema.py_schema import PyApplication, PyModule, PyCallable +from codeanalyzer.schema.py_schema import ( + PyApplication, PyModule, PyClass, PyCallable, PyCallsite, +) def test_neo4j_callable_key_equals_json_id(): @@ -13,3 +15,20 @@ def test_neo4j_callable_key_equals_json_id(): keys = {n.value for n in rows.nodes} assert fn.id in keys # the callable node is keyed by its can:// id assert app.symbol_table["m.py"].id in keys + + +def test_py_resolves_to_edge_targets_declared_callee_by_can_id(): + callee = PyCallable(name="g", path="m.py", signature="m.g", parameters=[]) + cs = PyCallsite(method_name="g", start_line=2, start_column=4, end_line=2, + end_column=7, callee_signature="m.g") + caller = PyCallable(name="f", path="m.py", signature="m.f", parameters=[], + call_sites=[cs]) + mod = PyModule(file_path="m.py", module_name="m", source="def f():\n g()\n", + functions={"f": caller, "g": callee}) + app = PyApplication(symbol_table={"m.py": mod}) + sig_to_id = assign_ids(app, "myapp") + rows = project(app, "myapp", sig_to_id) + resolves = [e for e in rows.edges if e.type == "PY_RESOLVES_TO"] + # 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"