diff --git a/.claude/FACADE_DECISIONS.md b/.claude/FACADE_DECISIONS.md index 9f93cfc..6757841 100644 --- a/.claude/FACADE_DECISIONS.md +++ b/.claude/FACADE_DECISIONS.md @@ -3,6 +3,21 @@ Decision log for SDK-surface design (per designing-cldk-changes, sdk-facade-design-loop). One line per locked decision; newest section first. +## 2026-07-27 — L3/L4 verb wiring is staged per release + +- **Staging:** rc.1 wires the five slice/flow verbs on the Python facade only (local + Neo4j); + rc.2 adds TypeScript; rc.3 adds the Java honest-degrade leg; rc.4 (Go) and rc.5 (C++) are + new-language legs entering via designing-cldk-changes (the C++ leg absorbs the existing C + facade); 2.0.0 final swaps the Rust query core (#279) in under the verbs, restores the + all-language DoD, and closes #270. +- **URI minting:** providers mint body-vertex URIs as `@`, + matching the analyzers' own param_in/param_out vocabulary (the real 1.0.2 Neo4j emitter stores + these can:// ids directly on PyCFGNode). Local keys never escape a provider. +- **source_slice contract:** existing span-less (synthetic) vertex → `(module_path, None)`; + unknown vertex → `(None, None)`; never derive a line from key shape. Both backends identical. +- **Rust hand-off unchanged:** the Rust query core (#279) replaces the Python engine underneath + these verbs post-M1; the facade surface added here is the stable contract it must satisfy. + ## 2026-07-21 — Rust query engine (fluent API core) - **M1 scope:** Epic B success criteria — the two Odoo PoE audit queries (#155), single-language, diff --git a/cldk/analysis/__init__.py b/cldk/analysis/__init__.py index 7735d3c..9ff4598 100644 --- a/cldk/analysis/__init__.py +++ b/cldk/analysis/__init__.py @@ -28,3 +28,26 @@ class AnalysisLevel(str, Enum): call_graph = "call graph" program_dependency_graph = "program dependency graph" system_dependency_graph = "system dependency graph" + + +def to_analysis_level(value) -> AnalysisLevel: + """Normalize a level given as an AnalysisLevel, its value ("call graph"), + or its name ("call_graph") — both spellings are in the wild.""" + if isinstance(value, AnalysisLevel): + return value + try: + return AnalysisLevel(value) + except ValueError: + try: + return AnalysisLevel[value] + except KeyError: + raise ValueError(f"unknown analysis level: {value!r}") from None + + +#: Facade-level vocabulary → the analyzers' integer analysis level (schema 2.0 ``max_level``). +ANALYSIS_LEVEL_TO_INT = { + AnalysisLevel.symbol_table: 1, + AnalysisLevel.call_graph: 2, + AnalysisLevel.program_dependency_graph: 3, + AnalysisLevel.system_dependency_graph: 4, +} diff --git a/cldk/analysis/python/backend.py b/cldk/analysis/python/backend.py index c0c64ad..70ea452 100644 --- a/cldk/analysis/python/backend.py +++ b/cldk/analysis/python/backend.py @@ -34,6 +34,7 @@ import networkx as nx +from cldk.graph.provider import ProgramGraphProvider from cldk.models.python import ( PyApplication, PyCallable, @@ -45,7 +46,7 @@ ) -class PythonAnalysisBackend(ABC): +class PythonAnalysisBackend(ProgramGraphProvider, ABC): """Abstract base every Python analysis backend implements. A backend owns all indexing and query logic for a Python application; the diff --git a/cldk/analysis/python/codeanalyzer/codeanalyzer.py b/cldk/analysis/python/codeanalyzer/codeanalyzer.py index 4f896a8..e242b9b 100644 --- a/cldk/analysis/python/codeanalyzer/codeanalyzer.py +++ b/cldk/analysis/python/codeanalyzer/codeanalyzer.py @@ -60,8 +60,9 @@ from codeanalyzer.options import AnalysisOptions from codeanalyzer.schema import model_dump_json -from cldk.analysis import AnalysisLevel +from cldk.analysis import ANALYSIS_LEVEL_TO_INT, AnalysisLevel, to_analysis_level from cldk.analysis.python.backend import PythonAnalysisBackend +from cldk.graph._cpg_local import CpgLocalProviderMixin from cldk.utils.exceptions import CldkSchemaMismatchException from cldk.models.python import ( PyApplication, @@ -104,7 +105,7 @@ def _overview(c: PyCallable, class_signature: str | None, kind: str) -> PyCallab ) -class PyCodeanalyzer(PythonAnalysisBackend): +class PyCodeanalyzer(CpgLocalProviderMixin, PythonAnalysisBackend): """In-process driver for the ``codeanalyzer-python`` analysis backend. This class serves as the primary interface to the codeanalyzer-python @@ -194,6 +195,7 @@ def __init__( if not self.project_dir.is_dir(): raise ValueError(f"project_dir does not exist or is not a directory: {self.project_dir}") self.analysis_level = analysis_level + self._level_int = ANALYSIS_LEVEL_TO_INT[to_analysis_level(analysis_level)] self.eager_analysis = eager_analysis self.target_files = target_files self.use_ray = use_ray @@ -211,7 +213,7 @@ def __init__( for class_sig in module.types: self._class_to_file[class_sig] = file_path - if analysis_level == AnalysisLevel.call_graph: + if self._level_int >= 2: self.call_graph: nx.DiGraph | None = self._build_call_graph(self.application.call_graph, self._id_to_signature()) else: self.call_graph = None @@ -246,18 +248,27 @@ def _run_analyzer(self) -> PyApplication: logger.warning("codeanalyzer-python supports only a single target file; using the first.") target_file = Path(self.target_files[0]) - options = AnalysisOptions( - input=self.project_dir, - output=self.analysis_json_path, - format=OutputFormat.JSON, - using_ray=self.use_ray, - rebuild_analysis=self.eager_analysis, - skip_tests=True, - file_name=target_file, - cache_dir=self.cache_dir, - clear_cache=False, - verbosity=0, - ) + opts = { + "input": self.project_dir, + "output": self.analysis_json_path, + "format": OutputFormat.JSON, + "using_ray": self.use_ray, + "rebuild_analysis": self.eager_analysis, + "skip_tests": True, + "file_name": target_file, + "cache_dir": self.cache_dir, + "clear_cache": False, + "verbosity": 0, + } + # Add analysis_level if _level_int is available (set in __init__). The guard IS + # load-bearing, not dead defensiveness: tests/analysis/python/test_python_schema_contract.py + # exercises this method on a `PyCodeanalyzer.__new__` instance that never ran `__init__` + # (and so never set `_level_int`) to isolate the schema-envelope gate below from analyzer + # construction — removing the guard would make those tests raise AttributeError. + if hasattr(self, "_level_int"): + opts["analysis_level"] = self._level_int + + options = AnalysisOptions(**opts) with Codeanalyzer(options) as analyzer: analysis = analyzer.analyze() @@ -266,8 +277,18 @@ def _run_analyzer(self) -> PyApplication: f"analysis schema {analysis.schema_version!r} from codeanalyzer-python, this SDK speaks " f"{self.SUPPORTED_ANALYSIS_SCHEMA!r} — align the pinned codeanalyzer-python with the SDK" ) + # The envelope reports what the analyzer actually computed; the capability gate + # (cldk/graph/capability.py) reads it via max_level() — recorded, never sniffed. + # Schema 2.0.0+'s Analysis envelope always carries max_level (default 1), so capturing + # it is unconditional here; an envelope that somehow lacks the attribute should fail + # loudly rather than silently fabricate a level. + self._max_level: int = analysis.max_level return analysis.application + def max_level(self) -> int: + """The analysis level of the underlying run (1-4), as reported by the analyzer.""" + return self._max_level + def _id_to_signature(self) -> Dict[str, str]: """Map every symbol-table callable's ``can://`` id to its dotted signature. diff --git a/cldk/analysis/python/neo4j/neo4j_backend.py b/cldk/analysis/python/neo4j/neo4j_backend.py index 21afc14..b0cec19 100644 --- a/cldk/analysis/python/neo4j/neo4j_backend.py +++ b/cldk/analysis/python/neo4j/neo4j_backend.py @@ -17,9 +17,12 @@ """Neo4j-backed Python analysis backend (read-only Cypher client). A drop-in alternative to :class:`~cldk.analysis.python.codeanalyzer.PyCodeanalyzer`: it exposes the -**same query method surface** (the 21 methods of :class:`PythonAnalysisBackend`) so the -:class:`~cldk.analysis.python.PythonAnalysis` facade can delegate to either one, but every method -answers by running **Cypher over a live Neo4j graph** instead of walking the in-memory +**same query method surface** — the 21 :class:`PythonAnalysisBackend` accessors plus the six +:class:`~cldk.graph.provider.ProgramGraphProvider` primitives (``program_graph``, ``sdg_edges``, +``resolve_location``, ``source_slice``, ``callable_of``, ``max_level``) that ABC now also requires +(#270) — so the :class:`~cldk.analysis.python.PythonAnalysis` facade and the slice/flow +:class:`~cldk.graph.engine.Engine` can both delegate to either backend interchangeably. Every +method answers by running **Cypher over a live Neo4j graph** instead of walking the in-memory pydantic / NetworkX structures. Mirrors :class:`~cldk.analysis.typescript.neo4j.TSNeo4jBackend`. This class is purely a **query client**: it never builds the graph and has no dependency on the @@ -71,13 +74,14 @@ from __future__ import annotations import logging -from typing import Any, Dict, List, Tuple +from typing import Any, Dict, Iterable, List, Optional, Tuple import networkx as nx from codeanalyzer.schema import model_dump_json from cldk.analysis.python.backend import PythonAnalysisBackend from cldk.analysis.python.neo4j import reconstruct as R +from cldk.models.cpg import Edge as CpgEdge from cldk.models.python import ( PyApplication, PyCallEdge, @@ -209,6 +213,215 @@ def _load_module_keys(self) -> List[str]: ) return [r["k"] for r in rows] + # ===================================================================================== + # ProgramGraphProvider primitives (#270) — same app-scoping idiom as every other accessor + # in this file: anchor on (:PyApplication {name: $app})-[:PY_HAS_MODULE]->(:PyModule) and + # filter node-carrying queries by that module set. Folded into a single ``WITH ... AS mods`` + # prelude per query (rather than reusing the cached ``self._modules`` list other accessors + # build in ``__init__``) so every primitive stays a single round trip. + # ===================================================================================== + _MODULES_CTE = "MATCH (:PyApplication {name: $app})-[:PY_HAS_MODULE]->(m:PyModule) WITH collect(m.file_key) AS mods " + + def _sig_to_can(self) -> Dict[str, str]: + """Lazy, cached dotted-signature -> can:// id map, built from this app's ``PyCallable`` + nodes (each carries its minted ``can://...`` id in ``.id``). Only feeds ``_to_uri``'s + defensive ``#``-form fallback (see its docstring) — the real emitter's ``PyCFGNode.id`` + needs no such translation, but ``PyCallable.signature`` (dotted) -> ``PyCallable.id`` + (can://) is also how ``program_graph``/``source_slice`` resolve a caller-supplied can:// + callable id back to the ``signature`` a Cypher ``MATCH`` needs. + """ + m = getattr(self, "_sig_can_map", None) + if m is None: + rows = self._run( + self._MODULES_CTE + "MATCH (c:PyCallable) WHERE c._module IN mods RETURN c.signature AS sig, c.id AS id", + app=self.application_name, + ) + m = {r["sig"]: r["id"] for r in rows if r.get("id")} + self._sig_can_map = m + self._can_sig_map = {v: k for k, v in m.items()} + return m + + def _to_uri(self, cfg_node_id: str) -> str: + """Translate a raw ``PyCFGNode.id`` into the minted ``can://...@`` vertex id + the local backend's mixin would produce. + + Handles two shapes defensively: + + * The **real** codeanalyzer-python 1.0.2 emitter (verified against a live Neo4j + instance populated by the real analyzer+emitter — see #295) stores ``PyCFGNode.id`` + as the already-minted ``can://...@`` URI directly, with no ``#`` anywhere + in it. That id *is* the answer; no translation is needed or possible (a dotted-sig + lookup would only ever miss). + * The dotted ``"#"`` form documented in the analyzer repo's + ``schema.py`` comment (and this file's original brief) has not been observed on any + real graph. Kept as a defensive fallback in case a future/older emitter version + actually produces it. + """ + if "#" not in cfg_node_id: + return cfg_node_id + sig, _, key = cfg_node_id.partition("#") + can = self._sig_to_can().get(sig, sig) + return f"{can}@{key.removeprefix('@')}" + + def max_level(self) -> int: + """The deepest overlay actually present in the graph for this application (see the + module docstring's :class:`PyNeo4jBackend` — persisted ``max_level`` is upstream gap + codellm-devkit/codeanalyzer-python, filed in #270's Task 5).""" + scope = self._MODULES_CTE + app = self.application_name + if self._run( + scope + "MATCH (a)-[r:PY_PARAM_IN|PY_PARAM_OUT|PY_SUMMARY]->() WHERE a._module IN mods RETURN 1 AS one LIMIT 1", + app=app, + ): + return 4 + if self._run(scope + "MATCH (n:PyCFGNode) WHERE n._module IN mods RETURN 1 AS one LIMIT 1", app=app): + return 3 + if self._run(scope + "MATCH (s:PySymbol)-[r:PY_CALLS]->() WHERE s._module IN mods RETURN 1 AS one LIMIT 1", app=app): + return 2 + return 1 + + def program_graph(self, callable_uri: str) -> nx.MultiDiGraph: + """The per-callable CFG/CDG/DDG overlay, translated to can:// vertex ids. Parallel + cfg/cdg/ddg edges between the same vertex pair stay distinct (MultiDiGraph auto-keys + each ``add_edge``), matching :class:`~cldk.graph.provider.ProgramGraphProvider`'s + contract and the local backend's mixin. + """ + self._sig_to_can() + sig = self._can_sig_map.get(callable_uri) + g = nx.MultiDiGraph() + if sig is None: + return g + scope = self._MODULES_CTE + app = self.application_name + rows = self._run( + scope + "MATCH (c:PyCallable {signature:$sig})-[:PY_HAS_CFG_NODE]->(n:PyCFGNode) " + "WHERE c._module IN mods " + "RETURN n.id AS id, n.kind AS kind, n.start_line AS sl, n.end_line AS el " + "ORDER BY n.id", + app=app, + sig=sig, + ) + for r in rows: + g.add_node(self._to_uri(r["id"]), kind=r["kind"], span=(r["sl"], r["el"])) + for fam, rel in (("cfg", "PY_CFG_NEXT"), ("cdg", "PY_CDG"), ("ddg", "PY_DDG")): + edge_rows = self._run( + scope + f"MATCH (c:PyCallable {{signature:$sig}})-[:PY_HAS_CFG_NODE]->(a)-[r:{rel}]->(b) " + "WHERE c._module IN mods AND (c)-[:PY_HAS_CFG_NODE]->(b) " + "RETURN a.id AS src, b.id AS dst, r.kind AS kind, r.var AS var, r.prov AS prov " + "ORDER BY a.id, b.id", + app=app, + sig=sig, + ) + for r in edge_rows: + g.add_edge( + self._to_uri(r["src"]), + self._to_uri(r["dst"]), + family=fam, + kind=r.get("kind"), + var=r.get("var"), + prov=list(r.get("prov") or []), + ) + return g + + def sdg_edges(self) -> Iterable[CpgEdge]: + """Every application-level interprocedural edge (``PY_PARAM_IN``/``PY_PARAM_OUT``/ + ``PY_SUMMARY``), translated to can:// vertex ids and kinded (real graph edges carry no + ``kind`` of their own — the family name doubles as the kind, mirroring the local + backend's mixin so a boundary hop never surfaces as opaque "sdg").""" + self._sig_to_can() + scope = self._MODULES_CTE + app = self.application_name + out: List[CpgEdge] = [] + for kind, rel in (("param_in", "PY_PARAM_IN"), ("param_out", "PY_PARAM_OUT"), ("summary", "PY_SUMMARY")): + rows = self._run( + scope + f"MATCH (a:PyCFGNode)-[r:{rel}]->(b:PyCFGNode) " + "WHERE a._module IN mods AND b._module IN mods " + "RETURN a.id AS src, b.id AS dst, r.var AS var " + "ORDER BY a.id, b.id", + app=app, + ) + out.extend( + CpgEdge(src=self._to_uri(r["src"]), dst=self._to_uri(r["dst"]), kind=kind, var=r.get("var"), prov=[]) for r in rows + ) + return out + + def resolve_location(self, file: str, line: int, col: Optional[int] = None) -> List[str]: + """Vertex ids at a source location, ordered by parsed column — the local key encodes + ``line:col`` (e.g. ``"3:8"``), so column is recovered by parsing the key rather than + from a stored property, matching the local backend's ordering exactly. + + Parses the key out of ``_to_uri``'s *output* (always ``"@"``, no + ``#``), not the raw ``r["id"]`` — the real emitter's raw ``PyCFGNode.id`` already *is* + that translated form (see ``_to_uri``), so partitioning the raw id on ``#`` (as a first + version of this method did) would never find a separator and always parse an empty key. + + Accepts ``file`` as either the full stored ``_module`` key or just its basename/suffix + (``"mod.py"`` matching ``"pkg/mod.py"``) — the same latitude the local backend's mixin + gives a caller, so an identical seed string behaves identically on both backends. + """ + scope = self._MODULES_CTE + rows = self._run( + scope + "MATCH (n:PyCFGNode) WHERE n._module IN mods " + "AND (n._module = $file OR n._module ENDS WITH $suffix) AND n.start_line = $line " + "RETURN n.id AS id", + app=self.application_name, + file=file, + suffix="/" + file, + line=line, + ) + hits = [] + for r in rows: + uri = self._to_uri(r["id"]) + key = uri.partition("@")[2] + head = key.split("/")[0] + c = int(head.split(":")[1]) if ":" in head else -1 + if col is None or c == col: + hits.append(((line, c), uri)) + return [u for _, u in sorted(hits)] + + def source_slice(self, vertex_uri: str) -> Tuple[Optional[str], Optional[str]]: + """Lossy by contract: ``PyCFGNode`` carries no byte offsets and modules carry no source + in the graph, so ``code`` is always ``None`` (Task 6's parity harness treats this as the + documented exception). The location half follows the same three-way contract as the + local backend's mixin — never fabricated by parsing the vertex key's shape: + + * the vertex doesn't exist (unknown callable, or no matching ``PyCFGNode``) → ``(None, None)``; + * it exists but carries no ``start_line`` (a synthetic ``@entry``/``@exit``/ + ``@formal_in:N``/``@actual_*`` port) → ``(module_path, None)``; + * it exists with a ``start_line`` → ``(f"{module_path}:{start_line}", None)``. + + Matches the EXACT vertex by comparing ``_to_uri(n.id)`` (which bridges both the real + emitter's already-minted can:// ``PyCFGNode.id`` and the defensive dotted-sig ``#`` form) + against ``vertex_uri`` — scoped to the owning callable so this stays a single, cheap + round trip rather than a whole-application node scan. + """ + self._sig_to_can() + cid = self.callable_of(vertex_uri) + sig = self._can_sig_map.get(cid) + if sig is None: + return (None, None) + scope = self._MODULES_CTE + rows = self._run( + scope + "MATCH (c:PyCallable {signature:$sig})-[:PY_HAS_CFG_NODE]->(n:PyCFGNode) " + "WHERE c._module IN mods " + "RETURN n.id AS id, n.start_line AS sl, n._module AS mod", + app=self.application_name, + sig=sig, + ) + for r in rows: + if self._to_uri(r["id"]) != vertex_uri: + continue + if r.get("sl") is None: + return (r["mod"], None) + return (f"{r['mod']}:{r['sl']}", None) + return (None, None) + + def callable_of(self, vertex_uri: str) -> Optional[str]: + """The owning callable's can:// id — vertex ids are ``"@"``, so this + is a partition at the first ``@`` (can:// ids never contain one themselves).""" + head, sep, _ = vertex_uri.partition("@") + return head if sep else vertex_uri + # ===================================================================================== # Reconstruction helpers — fetch a node's children over Cypher, then assemble via R. # ===================================================================================== diff --git a/cldk/analysis/python/python_analysis.py b/cldk/analysis/python/python_analysis.py index 3262309..41dd732 100644 --- a/cldk/analysis/python/python_analysis.py +++ b/cldk/analysis/python/python_analysis.py @@ -57,6 +57,7 @@ from cldk.analysis.python.backend import PythonAnalysisBackend from cldk.analysis.python.codeanalyzer import PyCodeanalyzer from cldk.analysis.python.neo4j import PyNeo4jBackend +from cldk.graph import Engine from cldk.models.python import ( PyApplication, PyCallable, @@ -234,6 +235,31 @@ def get_raw_ast(self, source_code: str) -> Tree: """ return self.treesitter_python.get_raw_ast(source_code) + # -----[ L3/L4 slice & flow verbs (#270) ]----- + def slice_backward(self, seed, *, edges=("cfg", "cdg", "ddg"), + interprocedural=None, strict=False): + """Backward program slice from ``seed`` ('file:line[:col]', can:// URI, or body node).""" + return Engine(self.backend).slice_backward( + seed, edges=edges, interprocedural=interprocedural, strict=strict) + + def slice_forward(self, seed, *, edges=("cfg", "cdg", "ddg"), + interprocedural=None, strict=False): + """Forward program slice from ``seed``.""" + return Engine(self.backend).slice_forward( + seed, edges=edges, interprocedural=interprocedural, strict=strict) + + def flows_to(self, source_seed, sink_seed, *, strict=False): + """Dataflow reachability with witness paths (interprocedural at L4).""" + return Engine(self.backend).flows_to(source_seed, sink_seed, strict=strict) + + def def_use(self, seed, *, strict=False): + """Definitions/uses reachable from ``seed`` over the dataflow graph.""" + return Engine(self.backend).def_use(seed, strict=strict) + + def control_deps(self, seed, *, strict=False): + """The control-dependence ancestors of ``seed`` (intraprocedural by design).""" + return Engine(self.backend).control_deps(seed, strict=strict) + # -----[ application view ]----- def get_application_view(self) -> PyApplication: """Return the complete analyzed application model. diff --git a/cldk/graph/__init__.py b/cldk/graph/__init__.py index e69de29..ce47511 100644 --- a/cldk/graph/__init__.py +++ b/cldk/graph/__init__.py @@ -0,0 +1,5 @@ +from cldk.graph.capability import CapabilityError +from cldk.graph.engine import Engine +from cldk.graph.result import FlowPath, FlowResult, SliceResult + +__all__ = ["CapabilityError", "Engine", "FlowPath", "FlowResult", "SliceResult"] diff --git a/cldk/graph/_cpg_local.py b/cldk/graph/_cpg_local.py new file mode 100644 index 0000000..8d9b6bb --- /dev/null +++ b/cldk/graph/_cpg_local.py @@ -0,0 +1,163 @@ +# cldk/graph/_cpg_local.py +from __future__ import annotations +from typing import Any, Dict, Iterable, List, Optional, Tuple +import networkx as nx +from cldk.models.cpg import Edge as CpgEdge + + +def _qualify(callable_id: str, local_key: str) -> str: + """A body node's dict key is local to its callable ('3:8', '@entry', '@formal_in:0', ...). + Application-level param_in/param_out already cross-reference these bodies as + '@', with the local key's own leading '@' (entry/exit/formal_*) + doubling as the join separator (never '...)@@entry'). Reproducing that exact scheme is what + lets provider-synthesized vertex ids agree with the ids the Application already uses.""" + return f"{callable_id}{local_key}" if local_key.startswith("@") else f"{callable_id}@{local_key}" + + +class CpgLocalProviderMixin: + """Implements ProgramGraphProvider's five read primitives over an in-memory cpg Application + on self.application. Language-neutral: the cpg models are shared across every analyzer, so + one mixin serves every local backend. Concrete backends supply max_level() (Task 8). + + A body node is keyed by local position in Node.body and normally carries no `.id` of its own + — only durable nodes (types/callables/functions/fields) are required to have one (see + Node's docstring and tests/models/cpg/test_node.py::test_body_node_missing_id_parses). This + mixin's canonical vertex id for a body node is `bn.id` when the analyzer did set one, else + the synthesized `_qualify(callable_id, local_key)` — matching how real param_in/param_out + already reference these nodes. + """ + + application: Any # cpg Application + + # --- internal index (built lazily, cached on the instance) --- + def _index(self) -> Dict[str, Any]: + idx = getattr(self, "_cpg_idx", None) + if idx is not None: + return idx + callables: Dict[str, Any] = {} # callable id -> (callable Node, Module, path) + canon_of: Dict[str, Dict[str, str]] = {} # callable id -> {local key: canonical vertex id} + n2c: Dict[str, str] = {} # canonical vertex id -> owning callable id + nodes: Dict[str, Any] = {} # canonical vertex id -> (body Node, Module, path) + + def _add(c, mod, path): + # bn.id: only durable body nodes (rare) carry an explicit id; the upstream + # codeanalyzer-python BodyNode has no `.id` attribute at all, so probe with + # getattr first — `bn.id` below is only ever reached once that's confirmed present. + canon = {k: (bn.id if getattr(bn, "id", None) is not None else _qualify(c.id, k)) + for k, bn in c.body.items()} + canon_of[c.id] = canon + callables[c.id] = (c, mod, path) + for k, bn in c.body.items(): + vid = canon[k] + n2c[vid] = c.id + nodes[vid] = (bn, mod, path) + # Recurse into this callable's own closures (a function/method can declare further + # nested callables, e.g. a factory's inner helper) and any classes it declares + # locally (e.g. a function defining a small helper class) — getattr-defensive since + # the slimmer upstream models may omit these containers entirely. This mirrors + # codeanalyzer-python's own recursive walk (semantic_analysis/call_graph.py's + # _walk_callable/_walk_class_callables) that the Neo4j emitter also follows + # (neo4j/project.py's _project_callable/_project_class), so every callable that owns + # a body — however deeply nested — ends up indexed here too. + for inner_c in (getattr(c, "callables", None) or {}).values(): + _add(inner_c, mod, path) + for inner_t in (getattr(c, "types", None) or {}).values(): + _add_type(inner_t, mod, path) + + def _add_type(t, mod, path): + # A class contributes no body of its own — only its methods and any nested classes + # (which may themselves nest further, arbitrarily deep) do. + for c in (getattr(t, "callables", None) or {}).values(): + _add(c, mod, path) + for inner_t in (getattr(t, "types", None) or {}).values(): + _add_type(inner_t, mod, path) + + for path, mod in self.application.symbol_table.items(): + for t in mod.types.values(): + _add_type(t, mod, path) + for f in mod.functions.values(): + _add(f, mod, path) + + idx = {"callables": callables, "canon": canon_of, "n2c": n2c, "nodes": nodes} + self._cpg_idx = idx + return idx + + def program_graph(self, callable_uri: str) -> nx.MultiDiGraph: + g = nx.MultiDiGraph() + entry = self._index()["callables"].get(callable_uri) + if entry is None: + # An unknown callable (e.g. the engine's callable_of() passed an id straight + # through because it wasn't a body vertex) is a structural non-match, not an + # error — return the empty graph rather than raising KeyError. + return g + c, _, _ = entry + canon = self._index()["canon"][callable_uri] + for k, bn in c.body.items(): + g.add_node(canon[k], kind=bn.kind, span=bn.span) + # No explicit edge key: 'family' alone does not identify a parallel edge uniquely (e.g. a + # callable can have several ddg edges between the same pair, one per var, or two cfg + # edges to the same successor with different kinds). provider.py's ABC docstring requires + # such edges to stay distinct, so let MultiDiGraph auto-assign a fresh key per edge + # instead of colliding same-family parallels onto one. + # kind/var/prov are duck-typed via getattr: the upstream codeanalyzer-python edge models + # are slimmer than the cldk cpg Edge (e.g. CdgEdge has no .kind, CfgEdge has no .var/.prov) + # — absent attributes surface as None/[] rather than raising AttributeError. + for fam, edges in (("cfg", c.cfg), ("cdg", c.cdg), ("ddg", c.ddg)): + for e in edges: + g.add_edge(canon.get(e.src, e.src), canon.get(e.dst, e.dst), + family=fam, kind=getattr(e, "kind", None), var=getattr(e, "var", None), + prov=list(getattr(e, "prov", None) or [])) + return g + + def sdg_edges(self) -> Iterable[Any]: + # param_in/param_out are already '@'-qualified at the Application + # level; summary is per-callable and LOCAL like cfg/cdg/ddg, so it needs the same + # qualification program_graph applies. All three are stamped with their own kind + # ("param_in"/"param_out"/"summary"): real edges carry kind=None in the raw analysis, and + # Engine.flows_to reports a boundary hop's bare family ("sdg") whenever kind is unset, so + # leaving these untagged would surface every interprocedural hop as opaque "sdg". + # + # Built as fresh cldk.models.cpg.Edge objects rather than model_copy: the upstream + # ParamEdge/SummaryEdge models don't declare `kind`/`var`/`prov` at all, so + # model_copy(update={...}) on them would set undeclared fields (undefined behavior). + idx = self._index() + out = [CpgEdge(src=e.src, dst=e.dst, kind="param_in", + var=getattr(e, "var", None), prov=list(getattr(e, "prov", None) or [])) + for e in self.application.param_in] + out += [CpgEdge(src=e.src, dst=e.dst, kind="param_out", + var=getattr(e, "var", None), prov=list(getattr(e, "prov", None) or [])) + for e in self.application.param_out] + for c, _, _ in idx["callables"].values(): + canon = idx["canon"][c.id] + out += [CpgEdge(src=canon.get(e.src, e.src), dst=canon.get(e.dst, e.dst), + kind="summary", var=getattr(e, "var", None), + prov=list(getattr(e, "prov", None) or [])) + for e in c.summary] + return out + + def resolve_location(self, file: str, line: int, col: Optional[int] = None) -> List[str]: + hits = [] + for vid, (bn, _mod, path) in self._index()["nodes"].items(): + if path != file and not path.endswith("/" + file): + continue + if bn.span is None or bn.span.start[0] != line: + continue + if col is not None and bn.span.start[1] != col: + continue + hits.append((bn.span.start, vid)) + # Deterministic order — a backend-agnostic tie-break the Neo4j provider must + # reproduce too — rather than whatever order the body dict happens to iterate in. + return [vid for _, vid in sorted(hits)] + + def source_slice(self, vertex_uri: str) -> Tuple[Optional[str], Optional[str]]: + node = self._index()["nodes"].get(vertex_uri) + if node is None: + return (None, None) + bn, mod, path = node + if bn.span is None: + return (path, None) + code = mod.source[bn.span.bytes[0]:bn.span.bytes[1]] if mod.source else None + return (f"{path}:{bn.span.start[0]}", code) + + def callable_of(self, vertex_uri: str) -> Optional[str]: + return self._index()["n2c"].get(vertex_uri, vertex_uri) diff --git a/tests/analysis/python/test_python_l34_levels.py b/tests/analysis/python/test_python_l34_levels.py new file mode 100644 index 0000000..8c691e0 --- /dev/null +++ b/tests/analysis/python/test_python_l34_levels.py @@ -0,0 +1,140 @@ +"""Level plumbing for L3/L4 (#270): the facade's AnalysisLevel maps to the analyzer's integer +``analysis_level`` option, and the backend reports the envelope's ``max_level`` — captured from +the SAME in-process run, never re-derived or sniffed. codeanalyzer-python >= 1.0.2 models are +schema-2.0-shaped (PyCallable.body/cfg/cdg/ddg/summary, PyApplication.param_in/param_out), so +``self.application`` doubles as the cpg source the graph provider mixin walks — no second parse. +""" +import json +from pathlib import Path + +import pytest + +from cldk.analysis import ANALYSIS_LEVEL_TO_INT, AnalysisLevel, to_analysis_level + +RES = Path(__file__).parent.parent.parent / "resources" / "cpg" + + +def test_analysis_level_map_is_total(): + assert ANALYSIS_LEVEL_TO_INT == { + AnalysisLevel.symbol_table: 1, + AnalysisLevel.call_graph: 2, + AnalysisLevel.program_dependency_graph: 3, + AnalysisLevel.system_dependency_graph: 4, + } + + +def test_to_analysis_level_accepts_enum(): + """to_analysis_level passes through enum values unchanged.""" + assert to_analysis_level(AnalysisLevel.call_graph) is AnalysisLevel.call_graph + + +def test_to_analysis_level_accepts_value_form(): + """to_analysis_level accepts the enum's value (space-separated).""" + assert to_analysis_level("call graph") is AnalysisLevel.call_graph + + +def test_to_analysis_level_accepts_name_form(): + """to_analysis_level accepts the enum's name (underscore form).""" + assert to_analysis_level("call_graph") is AnalysisLevel.call_graph + assert to_analysis_level("symbol_table") is AnalysisLevel.symbol_table + assert to_analysis_level("program_dependency_graph") is AnalysisLevel.program_dependency_graph + assert to_analysis_level("system_dependency_graph") is AnalysisLevel.system_dependency_graph + + +def test_to_analysis_level_rejects_garbage(): + """to_analysis_level raises on unknown values.""" + with pytest.raises(ValueError, match="unknown analysis level"): + to_analysis_level("garbage") + + +class _FakeEnvelope: + """Stands in for the analyzer's schema-v2 Analysis envelope.""" + + def __init__(self, payload_dict): + from cldk.models.python import PyApplication + + self.schema_version = payload_dict["schema_version"] + self.max_level = payload_dict["max_level"] + self.application = PyApplication(**payload_dict["application"]) + + +@pytest.fixture +def backend(monkeypatch, tmp_path): + payload = json.loads((RES / "py-a4.json").read_text()) + captured = {} + + class _FakeCodeanalyzer: + def __init__(self, options): + captured["options"] = options + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def analyze(self): + return _FakeEnvelope(payload) + + import cldk.analysis.python.codeanalyzer.codeanalyzer as mod + + monkeypatch.setattr(mod, "Codeanalyzer", _FakeCodeanalyzer) + from cldk.analysis.python.codeanalyzer.codeanalyzer import PyCodeanalyzer + + b = PyCodeanalyzer( + project_dir=tmp_path, + analysis_level=AnalysisLevel.system_dependency_graph, + analysis_json_path=None, + eager_analysis=False, + ) + return b, captured + + +def test_max_level_captured_from_same_run(backend): + b, _ = backend + assert b.max_level() == 4 + + +def test_analyzer_receives_int_level(backend): + _, captured = backend + assert captured["options"].analysis_level == 4 + + +def test_call_graph_built_at_level_ge_2(backend): + b, _ = backend + assert b.call_graph is not None + # the pyfix sample's three internal callables all appear + assert b.call_graph.number_of_edges() >= 3 + + +def test_backend_accepts_underscore_analysis_level(monkeypatch, tmp_path): + """PyCodeanalyzer accepts analysis_level as underscore form (e.g. "call_graph").""" + payload = json.loads((RES / "py-a4.json").read_text()) + + class _FakeCodeanalyzer: + def __init__(self, options): + self.captured_options = options + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def analyze(self): + return _FakeEnvelope(payload) + + import cldk.analysis.python.codeanalyzer.codeanalyzer as mod + + monkeypatch.setattr(mod, "Codeanalyzer", _FakeCodeanalyzer) + from cldk.analysis.python.codeanalyzer.codeanalyzer import PyCodeanalyzer + + # Construct with underscore form + b = PyCodeanalyzer( + project_dir=tmp_path, + analysis_level="call_graph", + analysis_json_path=None, + eager_analysis=False, + ) + assert b.max_level() == 4 + assert b.call_graph is not None diff --git a/tests/analysis/python/test_python_schema_contract.py b/tests/analysis/python/test_python_schema_contract.py index 26c725f..7fec0f2 100644 --- a/tests/analysis/python/test_python_schema_contract.py +++ b/tests/analysis/python/test_python_schema_contract.py @@ -72,9 +72,12 @@ def test_neo4j_schema_version_absent_fails_fast(monkeypatch): # -----[ in-process envelope gate ]----- class _StubAnalysis: - def __init__(self, schema_version, application=None): + def __init__(self, schema_version, application=None, max_level=1): self.schema_version = schema_version self.application = application + # Real schema 2.0.0+ envelopes always carry max_level (pydantic default 1) — + # _run_analyzer now captures it unconditionally, so the stub must too. + self.max_level = max_level class _StubCodeanalyzer: diff --git a/tests/graph/test_cpg_local.py b/tests/graph/test_cpg_local.py new file mode 100644 index 0000000..1c26ef7 --- /dev/null +++ b/tests/graph/test_cpg_local.py @@ -0,0 +1,412 @@ +"""CpgLocalProviderMixin: Step 1's hand-built Application (kept verbatim) plus the same five +primitives exercised end-to-end against a REAL committed L4 golden sample +(tests/resources/cpg/py-a4.json) — including driving the merged Engine on real data.""" +import json +from pathlib import Path +import networkx as nx +from cldk.graph._cpg_local import CpgLocalProviderMixin +from cldk.graph.engine import Engine +from cldk.models.cpg.models import Application, Module, Node, Edge, Span +from cldk.models.cpg import AnalysisPayload +from cldk.models.cpg import Edge as CpgEdge + + +def _app(): + call = Node(id="can://x/m.py/f", kind="function", signature="f", + span=Span(start=(1, 0), end=(4, 0), bytes=(0, 40)), + body={"f@1:0": Node(id="can://x/m.py/f@1:0", kind="statement", + span=Span(start=(1, 0), end=(1, 8), bytes=(0, 8))), + "f@2:0": Node(id="can://x/m.py/f@2:0", kind="statement", + span=Span(start=(2, 0), end=(2, 8), bytes=(9, 17)))}, + cfg=[Edge(src="can://x/m.py/f@1:0", dst="can://x/m.py/f@2:0", kind="fallthrough")], + ddg=[Edge(src="can://x/m.py/f@1:0", dst="can://x/m.py/f@2:0", var="a", prov=["ssa"])], + cdg=[], summary=[]) + mod = Module(id="can://x/m.py", source="a = 1\nb = a\n", functions={"f": call}, types={}) + return Application(id="can://x", symbol_table={"m.py": mod}, call_graph=[], param_in=[], param_out=[]) + + +class Backend(CpgLocalProviderMixin): + def __init__(self): self.application = _app(); self._level = 3 + def max_level(self): return self._level + + +def test_program_graph_has_body_and_edges(): + g = Backend().program_graph("can://x/m.py/f") + assert set(g.nodes) == {"can://x/m.py/f@1:0", "can://x/m.py/f@2:0"} + fams = {d["family"] for _, _, d in g.edges(data=True)} + assert fams == {"cfg", "ddg"} + + +def test_resolve_location_hits_line(): + assert Backend().resolve_location("m.py", 2) == ["can://x/m.py/f@2:0"] + + +def test_source_slice_reads_module_source(): + fl, code = Backend().source_slice("can://x/m.py/f@1:0") + assert fl == "m.py:1" and code == "a = 1\nb " + + +# --- Golden fixture: a REAL, conformant L4 codeanalyzer-python sample ------------------------ +# Unlike the hand-built Application above (whose body nodes carry an explicit .id matching the +# brief's own convention), a real analyzer leaves body-node .id unset — they're keyed by LOCAL +# position ("3:8", "@entry", ...) per Node's own docstring and tests/models/cpg/test_node.py's +# test_body_node_missing_id_parses. Application-level param_in/param_out already reference these +# as "@" (verified against the raw JSON below), which is the scheme the +# mixin must reproduce so provider-synthesized ids agree with the application's own. +RES = Path(__file__).parent.parent / "resources" / "cpg" +GOLDEN = AnalysisPayload(**json.loads((RES / "py-a4.json").read_text())).application + +MOD_PATH = "pkg/mod.py" +ENTRY = "can://python/pyfix/pkg/mod.py/entry()" +RESET_PW = "can://python/pyfix/pkg/mod.py/ResUsers/reset_password(self,login)" +ACTION = "can://python/pyfix/pkg/mod.py/ResUsers/_action_reset_password(self,ids)" + +# reset_password's own body keys, straight from py-a4.json (11 total: 2 spanned source +# positions, entry/exit, 2 formal_in/2 formal_out ports, 3 actual_in/out ports for its one +# call site). +RESET_PW_BODY_KEYS = ["3:15", "@entry", "3:8", "@exit", "@formal_in:0", "@formal_in:1", + "@formal_out:0", "@formal_out:1", "3:8/actual_in:0", "3:8/actual_in:1", + "3:8/actual_out"] + + +def _qid(callable_id: str, local_key: str) -> str: + # Independent re-derivation of the qualification scheme (not imported from the mixin under + # test) so a bug in the mixin's own _qualify can't silently validate itself. + return callable_id + local_key if local_key.startswith("@") else f"{callable_id}@{local_key}" + + +class GoldenBackend(CpgLocalProviderMixin): + application = GOLDEN + def max_level(self): return 4 + + +def test_golden_program_graph_preserves_parallel_edges(): + # reset_password has THREE ddg edges between the same pair (@entry -> 3:8, one per var) and + # TWO cfg edges between another same pair (3:8 -> @exit, kinds 'exception' and 'return'). + # provider.py's ABC docstring requires these to stay distinct parallel edges, each with its + # own family/var/prov/kind — a naive MultiDiGraph key of just the family name would silently + # collapse each trio/pair into one edge. + g = GoldenBackend().program_graph(RESET_PW) + assert isinstance(g, nx.MultiDiGraph) + assert set(g.nodes) == {_qid(RESET_PW, k) for k in RESET_PW_BODY_KEYS} + assert g.number_of_edges() == 7 # 3 cfg + 1 cdg + 3 ddg (summary excluded) + + fams = [d["family"] for _, _, d in g.edges(data=True)] + assert sorted(fams) == sorted(["cfg", "cfg", "cfg", "cdg", "ddg", "ddg", "ddg"]) + + ddg_vars = {d["var"] for _, _, d in g.edges(data=True) if d["family"] == "ddg"} + assert ddg_vars == {"login", "self", "self._action_reset_password"} # none collapsed away + + exit_kinds = {d["kind"] for _, _, d in g.edges(data=True) + if d["family"] == "cfg" and d["kind"] in ("exception", "return")} + assert exit_kinds == {"exception", "return"} # both 3:8 -> @exit edges survive + + assert g.has_edge(_qid(RESET_PW, "@entry"), _qid(RESET_PW, "3:8")) + + +def test_golden_resolve_location_hits_real_lines(): + b = GoldenBackend() + # line 3 has TWO spanned body nodes in reset_password: the call (col 15) and the return + # (col 8) — col=None must return both; a col narrows to exactly one. + assert set(b.resolve_location(MOD_PATH, 3)) == {_qid(RESET_PW, "3:15"), _qid(RESET_PW, "3:8")} + assert b.resolve_location(MOD_PATH, 3, 8) == [_qid(RESET_PW, "3:8")] + assert b.resolve_location(MOD_PATH, 3, 15) == [_qid(RESET_PW, "3:15")] + assert b.resolve_location("mod.py", 3, 8) == [_qid(RESET_PW, "3:8")] # basename suffix match + assert b.resolve_location(MOD_PATH, 2) == [] # the `def` line has no body node + + +def test_golden_source_slice_reads_real_module_source(): + b = GoldenBackend() + fl, code = b.source_slice(_qid(RESET_PW, "3:8")) + assert fl == "pkg/mod.py:3" and code == "return self._action_reset_password([login])" + fl2, code2 = b.source_slice(_qid(RESET_PW, "3:15")) + assert fl2 == "pkg/mod.py:3" and code2 == "self._action_reset_password([login])" + # @entry has no span (it's a synthetic CFG node, not a source position) — degrades to + # (path, None) rather than raising. + fl3, code3 = b.source_slice(_qid(RESET_PW, "@entry")) + assert fl3 == "pkg/mod.py" and code3 is None + + +def test_golden_callable_of_maps_body_node_and_is_identity_on_callable(): + b = GoldenBackend() + assert b.callable_of(_qid(RESET_PW, "3:8")) == RESET_PW + assert b.callable_of(_qid(RESET_PW, "@entry")) == RESET_PW + assert b.callable_of(_qid(ACTION, "5:8")) == ACTION + assert b.callable_of(RESET_PW) == RESET_PW # passthrough: not a body vertex + assert b.callable_of("can://nonexistent") == "can://nonexistent" + + +def test_golden_sdg_edges_include_param_in_out_and_summary_tagged_by_kind(): + # param_in/param_out are already @ qualified at the Application level; + # summary is per-callable and LOCAL like cfg/cdg/ddg, so it needs the same qualification the + # mixin applies to program_graph. All three must be tagged with their OWN kind + # ("param_in"/"param_out"/"summary") — real param_in/param_out/summary edges carry kind=None + # in the raw JSON, and engine.flows_to reports a boundary hop's family ("sdg") only when kind + # is unset (test_engine_interproc.py::test_flow_boundary_hop_reports_sdg_kind, already + # merged), so an untagged edge here would surface as the opaque "sdg" instead. + got = {(e.src, e.dst, e.kind) for e in GoldenBackend().sdg_edges()} + assert got == { + (_qid(RESET_PW, "3:8/actual_in:0"), _qid(ACTION, "formal_in:0"), "param_in"), + (_qid(RESET_PW, "3:8/actual_in:1"), _qid(ACTION, "formal_in:1"), "param_in"), + (_qid(ENTRY, "7:4/actual_in:0"), _qid(RESET_PW, "formal_in:1"), "param_in"), + (_qid(ACTION, "formal_out:0"), _qid(RESET_PW, "3:8/actual_out"), "param_out"), + (_qid(RESET_PW, "formal_out:0"), _qid(ENTRY, "7:4/actual_out"), "param_out"), + (_qid(RESET_PW, "3:8/actual_in:1"), _qid(RESET_PW, "3:8/actual_out"), "summary"), + (_qid(ENTRY, "7:4/actual_in:0"), _qid(ENTRY, "7:4/actual_out"), "summary"), + } + + +def test_golden_engine_slice_backward_on_real_line(): + # Feeds the mixin to the real merged Engine and drives it with a plain 'file:line[:col]' + # seed string, exactly as an SDK caller would — proving the mixin's five primitives compose + # correctly through resolve_vertex -> callable_of -> program_graph -> sdg overlay -> subgraph. + r = Engine(GoldenBackend()).slice_backward(f"{MOD_PATH}:3:8") + assert bool(r) and len(r) > 0 + # only @entry precedes 3:8 (via cfg/cdg/ddg); the sdg overlay is engaged (level 4, ddg + # requested) but nothing points INTO the bare '3:8' return node from another callable in + # this sample — the interprocedural wiring attaches at the unspanned actual/formal ports. + assert set(r.uris()) == {_qid(RESET_PW, "3:8"), _qid(RESET_PW, "@entry")} + assert r.explain()["interprocedural"] is True + assert r.explain()["level"] == 4 + + ev = {e["uri"]: e for e in r.evidence} + assert ev[_qid(RESET_PW, "3:8")]["code"] == "return self._action_reset_password([login])" + assert ev[_qid(RESET_PW, "3:8")]["role"] == "seed" + assert ev[_qid(RESET_PW, "@entry")]["code"] is None + assert ev[_qid(RESET_PW, "@entry")]["role"] == "def" + + +def test_golden_engine_flows_to_crosses_callable_boundary(): + # A real cross-callable dataflow hop, driven end to end through the merged Engine: the value + # passed as reset_password's 2nd call argument (3:8/actual_in:1) flows via param_in into + # _action_reset_password's 2nd formal parameter — proving the mixin's sdg_edges() actually + # drives Engine.flows_to's interprocedural path on real data, not just synthetic fixtures. + src = _qid(RESET_PW, "3:8/actual_in:1") + dst = _qid(ACTION, "formal_in:1") + r = Engine(GoldenBackend()).flows_to(src, dst) + assert len(r.paths) == 1 + hop = r.paths[0].hops[0] + assert hop["from"] == src and hop["to"] == dst + assert hop["kind"] == "param_in" # the specific sdg kind, not the opaque family "sdg" + # this param_in edge carries no ssa/points-to provenance in the raw sample, so the honest + # confidence is 'unresolved' rather than a stronger tier it can't back up. + assert r.paths[0].confidence == "unresolved" + + +# --- Slim-model hardening: the mixin must duck-type BOTH the cldk cpg models AND the slimmer +# upstream codeanalyzer-python 1.0.2 models. These stand-ins carry ONLY the fields the upstream +# BodyNode/CfgEdge/CdgEdge/DdgEdge/ParamEdge/SummaryEdge models declare (verified against +# codeanalyzer.schema.py_schema): no .id on body nodes, no .var/.prov on cfg edges, no .kind at +# all on cdg/param/summary edges. Accessing a field the real upstream model lacks must raise +# AttributeError here too, or this test would validate nothing. +class _SlimSpan: + def __init__(self, start): + self.start = start + + +class _SlimBodyNode: + """Mirrors upstream BodyNode: {kind, span, callee, of, parent} — no `.id`.""" + + def __init__(self, kind, span=None, callee=None, of=None, parent=None): + self.kind = kind + self.span = span + self.callee = callee + self.of = of + self.parent = parent + + +class _SlimCfgEdge: + """Mirrors upstream CfgEdge: {src, dst, kind} — no `.var`/`.prov`.""" + + def __init__(self, src, dst, kind="fallthrough"): + self.src = src + self.dst = dst + self.kind = kind + + +class _SlimCdgEdge: + """Mirrors upstream CdgEdge: {src, dst} — no `.kind`/`.var`/`.prov`.""" + + def __init__(self, src, dst): + self.src = src + self.dst = dst + + +class _SlimDdgEdge: + """Mirrors upstream DdgEdge: {src, dst, var, prov} — no `.kind`.""" + + def __init__(self, src, dst, var=None, prov=None): + self.src = src + self.dst = dst + self.var = var + self.prov = prov or [] + + +class _SlimParamEdge: + """Mirrors upstream ParamEdge: {src, dst} only.""" + + def __init__(self, src, dst): + self.src = src + self.dst = dst + + +class _SlimSummaryEdge: + """Mirrors upstream SummaryEdge: {src, dst} only.""" + + def __init__(self, src, dst): + self.src = src + self.dst = dst + + +class _SlimCallable: + def __init__(self, id, body, cfg=None, cdg=None, ddg=None, summary=None, callables=None, types=None): + self.id = id + self.body = body + self.cfg = cfg or [] + self.cdg = cdg or [] + self.ddg = ddg or [] + self.summary = summary or [] + # A callable can declare further nested callables (closures) and/or locally-defined + # classes — both absent by default (upstream PyCallable.callables/.types, empty dicts). + self.callables = callables or {} + self.types = types or {} + + +class _SlimClass: + """Mirrors upstream PyClass's type-facet containers: {callables, types} — methods and any + classes nested inside this one. No `.body`/`.cfg`/etc — a class contributes no body of its + own, only its members do.""" + + def __init__(self, callables=None, types=None): + self.callables = callables or {} + self.types = types or {} + + +class _SlimModule: + def __init__(self, functions, types=None): + self.types = types or {} + self.functions = functions + + +class _SlimApplication: + def __init__(self, symbol_table, param_in=None, param_out=None): + self.symbol_table = symbol_table + self.param_in = param_in or [] + self.param_out = param_out or [] + + +SLIM_CID = "can://slim/m.py/f" + + +# Ids for the nested-recursion fixture (#270 final review Finding 1): a closure declared +# inside the top-level function `f`, and a method reachable only through a class nested inside +# another class — both must be discoverable by _index()'s recursive walk, mirroring +# codeanalyzer-python's own _walk_callable/_walk_class_callables and the Neo4j emitter's +# _project_callable/_project_class. +CLOSURE_CID = f"{SLIM_CID}/g" +OUTER_METHOD_CID = "can://slim/m.py/Outer.m" +INNER_METHOD_CID = "can://slim/m.py/Outer.Inner.n" + + +def _slim_app(): + # Body dict deliberately inserts the col-15 key BEFORE the col-8 key, so a resolve_location + # that merely preserved dict insertion order would return them in the wrong order. + body = { + "3:15": _SlimBodyNode(kind="call", span=_SlimSpan(start=(3, 15))), + "3:8": _SlimBodyNode(kind="statement", span=_SlimSpan(start=(3, 8))), + "@entry": _SlimBodyNode(kind="entry", span=None), + } + # A closure nested inside `f` — its own callable, with its own body, reachable only via + # f.callables (never listed at module/class top level). + g = _SlimCallable(id=CLOSURE_CID, body={"5:0": _SlimBodyNode(kind="statement", span=_SlimSpan(start=(5, 0)))}) + f = _SlimCallable( + id=SLIM_CID, + body=body, + cfg=[_SlimCfgEdge(src="@entry", dst="3:8", kind="fallthrough")], + cdg=[_SlimCdgEdge(src="@entry", dst="3:15")], + ddg=[_SlimDdgEdge(src="@entry", dst="3:8", var="x", prov=["ssa"])], + summary=[_SlimSummaryEdge(src="3:8", dst="3:15")], + callables={"g": g}, + ) + # Outer.m is an ordinary method; Outer.Inner.n is a method on a class nested TWO levels deep + # (a class inside a class) — both must surface through mod.types' recursive walk, not just + # a single top-level pass over each class's own .callables. + inner_method = _SlimCallable(id=INNER_METHOD_CID, + body={"30:0": _SlimBodyNode(kind="statement", span=_SlimSpan(start=(30, 0)))}) + inner_cls = _SlimClass(callables={"n": inner_method}) + outer_method = _SlimCallable(id=OUTER_METHOD_CID, + body={"20:0": _SlimBodyNode(kind="statement", span=_SlimSpan(start=(20, 0)))}) + outer_cls = _SlimClass(callables={"m": outer_method}, types={"Inner": inner_cls}) + mod = _SlimModule(functions={"f": f}, types={"Outer": outer_cls}) + return _SlimApplication( + symbol_table={"m.py": mod}, + param_in=[_SlimParamEdge(src="can://slim/other@formal_in:0", dst=f"{SLIM_CID}@3:8")], + param_out=[_SlimParamEdge(src=f"{SLIM_CID}@3:8", dst="can://slim/other@formal_out:0")], + ) + + +class SlimBackend(CpgLocalProviderMixin): + application = _slim_app() + + def max_level(self): + return 4 + + +def test_slim_program_graph_duck_types_absent_fields_to_none_or_empty(): + g = SlimBackend().program_graph(SLIM_CID) + assert isinstance(g, nx.MultiDiGraph) + by_family = {d["family"]: d for _, _, d in g.edges(data=True)} + # every edge datum has family/kind/var/prov regardless of what the source model declared + assert by_family["cfg"]["kind"] == "fallthrough" + assert by_family["cfg"]["var"] is None and by_family["cfg"]["prov"] == [] + assert by_family["cdg"]["kind"] is None # CdgEdge has no .kind at all + assert by_family["cdg"]["var"] is None and by_family["cdg"]["prov"] == [] + assert by_family["ddg"]["kind"] is None # DdgEdge has no .kind at all + assert by_family["ddg"]["var"] == "x" and by_family["ddg"]["prov"] == ["ssa"] + + +def test_slim_sdg_edges_are_cpg_edge_instances_with_stamped_kind(): + edges = SlimBackend().sdg_edges() + got = {(e.src, e.dst, e.kind) for e in edges} + assert got == { + ("can://slim/other@formal_in:0", f"{SLIM_CID}@3:8", "param_in"), + (f"{SLIM_CID}@3:8", "can://slim/other@formal_out:0", "param_out"), + (f"{SLIM_CID}@3:8", f"{SLIM_CID}@3:15", "summary"), # summary re-qualified like cfg/cdg/ddg + } + assert all(isinstance(e, CpgEdge) for e in edges) + + +def test_slim_resolve_location_orders_by_start_line_and_col_not_insertion(): + hits = SlimBackend().resolve_location("m.py", 3) + assert hits == [f"{SLIM_CID}@3:8", f"{SLIM_CID}@3:15"] # col 8 first, despite 3:15 inserted first + + +def test_slim_program_graph_unknown_callable_returns_empty_graph_not_keyerror(): + g = SlimBackend().program_graph("can://nowhere") + assert isinstance(g, nx.MultiDiGraph) + assert g.number_of_nodes() == 0 and g.number_of_edges() == 0 + + +def test_index_recurses_into_closures_and_doubly_nested_class_methods(): + # #270 final review Finding 1 (Critical): _index() used to walk only mod.functions and + # mod.types[*].callables, one level — a closure declared inside a function, or a method on a + # class nested inside another class, was silently invisible (empty program_graph, no owning + # callable, no resolve_location hit) even though the Neo4j emitter walks these recursively. + b = SlimBackend() + + g = b.program_graph(CLOSURE_CID) + assert set(g.nodes) == {f"{CLOSURE_CID}@5:0"} + + m = b.program_graph(OUTER_METHOD_CID) + assert set(m.nodes) == {f"{OUTER_METHOD_CID}@20:0"} + + n = b.program_graph(INNER_METHOD_CID) + assert set(n.nodes) == {f"{INNER_METHOD_CID}@30:0"} + + assert b.callable_of(f"{CLOSURE_CID}@5:0") == CLOSURE_CID + assert b.callable_of(f"{OUTER_METHOD_CID}@20:0") == OUTER_METHOD_CID + assert b.callable_of(f"{INNER_METHOD_CID}@30:0") == INNER_METHOD_CID + + assert b.resolve_location("m.py", 5) == [f"{CLOSURE_CID}@5:0"] + assert b.resolve_location("m.py", 20) == [f"{OUTER_METHOD_CID}@20:0"] + assert b.resolve_location("m.py", 30) == [f"{INNER_METHOD_CID}@30:0"] diff --git a/tests/graph/test_facade_delegates.py b/tests/graph/test_facade_delegates.py new file mode 100644 index 0000000..8000c99 --- /dev/null +++ b/tests/graph/test_facade_delegates.py @@ -0,0 +1,49 @@ +from unittest.mock import MagicMock + +import networkx as nx + +from cldk.analysis.python.python_analysis import PythonAnalysis +from cldk.graph import SliceResult, FlowResult + + +def _facade_with_backend(backend): + pa = PythonAnalysis.__new__(PythonAnalysis) + pa.backend = backend + return pa + + +def _mock_backend(): + b = MagicMock() + b.max_level.return_value = 3 + b.callable_of.side_effect = lambda u: "c" + g = nx.MultiDiGraph() + # Edge direction: def→use; the definition at c@1:0 flows into the use at c@2:0. + # A backward slice from the use (c@2:0) walks incoming edges to reach the def (c@1:0). + g.add_edge("c@1:0", "c@2:0", family="ddg", prov=["ssa"], var="x", kind=None) + b.program_graph.return_value = g + b.sdg_edges.return_value = [] + b.resolve_location.return_value = ["c@2:0"] + b.source_slice.side_effect = lambda u: (u, u) + return b + + +def test_slice_backward_delegates(): + r = _facade_with_backend(_mock_backend()).slice_backward("m.py:2", edges=("ddg",)) + assert isinstance(r, SliceResult) + assert set(r.uris()) == {"c@2:0", "c@1:0"} + + +def test_flows_to_delegates(): + r = _facade_with_backend(_mock_backend()).flows_to("c@1:0", "c@2:0") + assert isinstance(r, FlowResult) + + +def test_all_five_verbs_exist(): + for verb in ("slice_backward", "slice_forward", "flows_to", "def_use", "control_deps"): + assert callable(getattr(PythonAnalysis, verb, None)), verb + + +def test_public_exports(): + import cldk.graph as gr + for name in ("Engine", "SliceResult", "FlowResult", "FlowPath", "CapabilityError"): + assert hasattr(gr, name), name diff --git a/tests/graph/test_golden_pyfix.py b/tests/graph/test_golden_pyfix.py new file mode 100644 index 0000000..5863448 --- /dev/null +++ b/tests/graph/test_golden_pyfix.py @@ -0,0 +1,106 @@ +"""Engine goldens over the pyfix L4 sample + the REAL duck-typed backend end-to-end: +PyCodeanalyzer's own upstream application (codeanalyzer-python 1.0.2 models) drives the +shared Engine through the mixin — the exact object graph production uses.""" +import json +from pathlib import Path + +import pytest + +from cldk.analysis import AnalysisLevel +from cldk.graph import CapabilityError +from cldk.graph._cpg_local import CpgLocalProviderMixin +from cldk.graph.engine import Engine +from cldk.models.cpg import AnalysisPayload + +RES = Path(__file__).parent.parent / "resources" / "cpg" +MOD = "pkg/mod.py" +ENTRY = "can://python/pyfix/pkg/mod.py/entry()" +C2 = "can://python/pyfix/pkg/mod.py/ResUsers/reset_password(self,login)" +C3 = "can://python/pyfix/pkg/mod.py/ResUsers/_action_reset_password(self,ids)" + + +class _Local(CpgLocalProviderMixin): + def __init__(self, application, level=4): + self.application = application + self._level = level + + def max_level(self): + return self._level + + +@pytest.fixture(scope="module") +def eng(): + app = AnalysisPayload(**json.loads((RES / "py-a4.json").read_text())).application + return Engine(_Local(app)) + + +def test_backward_slice_from_location_multi_seed(eng): + r = eng.slice_backward(f"{MOD}:5") + assert set(r.uris()) == {f"{C3}@entry", f"{C3}@5:8", f"{C3}@5:15"} + assert r.explain()["level"] == 4 and "degraded" not in r.explain() + + +def test_control_deps_exact(eng): + r = eng.control_deps(f"{C2}@3:8") + assert set(r.uris()) == {f"{C2}@entry", f"{C2}@3:8"} + + +def test_def_use_exact(eng): + r = eng.def_use(f"{C2}@entry") + assert set(r.uris()) == {f"{C2}@entry", f"{C2}@3:8"} + + +def test_flows_to_via_summary(eng): + r = eng.flows_to(f"{C2}@3:8/actual_in:1", f"{C2}@3:8/actual_out") + assert len(r.paths) == 1 + assert [h["kind"] for h in r.paths[0].hops] == ["summary"] + + +def test_flows_to_no_route_is_empty_not_error(eng): + r = eng.flows_to(f"{C3}@5:8", f"{ENTRY}@7:4") + assert len(r.paths) == 0 and not r + + +def test_capability_degrade_below_l4(eng): + # A level-2 provider over the SAME application object as `eng` (no re-analysis, no + # subclassing): flows_to needs L4 for its sdg (param_in/param_out/summary) overlay, so a + # cross-callable flow degrades to "no route" rather than crashing or silently completing. + low = Engine(_Local(eng.p.application, level=2)) + r = low.flows_to(f"{ENTRY}@7:4/actual_in:0", f"{C2}@formal_in:1") + assert len(r.paths) == 0 # no sdg overlay below L4 + assert r.explain()["degraded"]["requested"] == 4 + with pytest.raises(CapabilityError): + low.flows_to(f"{ENTRY}@7:4/actual_in:0", f"{C2}@formal_in:1", strict=True) + + +def test_real_backend_is_a_provider_end_to_end(monkeypatch, tmp_path): + # Build a REAL PyCodeanalyzer (upstream models, not cldk cpg models) via the fake-analyzer + # pattern from tests/analysis/python/test_python_l34_levels.py, then drive the Engine + # through it — this is the production object graph, and it is what catches upstream-model + # slimness (BodyNode without id, CfgEdge without var/prov). + import cldk.analysis.python.codeanalyzer.codeanalyzer as mod + from cldk.models.python import PyApplication + + payload = json.loads((RES / "py-a4.json").read_text()) + + class _Env: + schema_version = payload["schema_version"] + max_level = payload["max_level"] + application = PyApplication(**payload["application"]) + + class _Fake: + def __init__(self, options): ... + def __enter__(self): return self + def __exit__(self, *a): return False + def analyze(self): return _Env() + + monkeypatch.setattr(mod, "Codeanalyzer", _Fake) + b = mod.PyCodeanalyzer(project_dir=tmp_path, + analysis_level=AnalysisLevel.system_dependency_graph, + analysis_json_path=None, eager_analysis=False) + assert isinstance(b, CpgLocalProviderMixin) and b.max_level() == 4 + # exact same golden as the cpg-model path: + r = Engine(b).slice_backward(f"{MOD}:5") + assert set(r.uris()) == {f"{C3}@entry", f"{C3}@5:8", f"{C3}@5:15"} + flows = Engine(b).flows_to(f"{C2}@3:8/actual_in:1", f"{C3}@formal_in:1") + assert len(flows.paths) == 1 and flows.paths[0].hops[0]["kind"] == "param_in" diff --git a/tests/graph/test_py_neo4j_provider.py b/tests/graph/test_py_neo4j_provider.py new file mode 100644 index 0000000..b2992ab --- /dev/null +++ b/tests/graph/test_py_neo4j_provider.py @@ -0,0 +1,133 @@ +"""ProgramGraphProvider conformance for the read-only Neo4j Python backend (#270). + +Stub-based: fakes ``_run`` so no live Neo4j server is required. Verifies the identity +translation and that every query stays scoped to ``application_name`` the same way the file's +existing accessors do. + +``PyCFGNode.id`` rows below mirror the REAL codeanalyzer-python 1.0.2 emitter, confirmed against +a live Neo4j instance in #270 Task 6: ``n.id`` is already the fully-qualified ``can://...@key`` +vertex id, with no ``#`` separator anywhere (see issue #295, which this stub previously masked by +hard-coding the dotted-sig ``"#"``-separated form the analyzer repo's ``schema.py`` comment +describes but that has never actually been observed on the wire). ``test_to_uri_...hash_form`` +below is the one deliberately-kept case exercising ``_to_uri``'s defensive ``#`` fallback branch. +""" + +import pytest + +from cldk.analysis.python.neo4j.neo4j_backend import PyNeo4jBackend +from cldk.graph.provider import ProgramGraphProvider + +SIG2 = "pkg.mod.ResUsers.reset_password" +C2 = "can://python/pyfix/pkg/mod.py/ResUsers/reset_password(self,login)" +SIG3 = "pkg.mod.ResUsers._action_reset_password" +C3 = "can://python/pyfix/pkg/mod.py/ResUsers/_action_reset_password(self,ids)" + + +@pytest.fixture +def backend(monkeypatch): + b = PyNeo4jBackend.__new__(PyNeo4jBackend) + b.application_name = "pyfix" # real attribute name (neo4j_backend.py:134) + + def fake_run(query, **params): + q = " ".join(query.split()) + if "PY_PARAM_IN|PY_PARAM_OUT|PY_SUMMARY" in q and "LIMIT 1" in q: + return [{"one": 1}] + if "MATCH (c:PyCallable)" in q and "RETURN c.signature" in q: + # Both endpoints resolved — a param_in/out edge's dst (a @formal_in/@formal_out + # vertex on the CALLEE) must not silently fall through the sig_is_None guard in + # source_slice/callable_of; that would mask a parse bug on the synthetic-key path. + return [{"sig": SIG2, "id": C2}, {"sig": SIG3, "id": C3}] + if "PY_HAS_CFG_NODE" in q and "n._module AS mod" in q: + # source_slice's per-callable node scan (checked BEFORE the generic + # program_graph node query below, since both match "PY_HAS_CFG_NODE"/"RETURN n.id"). + if params.get("sig") == SIG2: + return [{"id": f"{C2}@entry", "sl": None, "mod": "pkg/mod.py"}, + {"id": f"{C2}@3:8", "sl": 3, "mod": "pkg/mod.py"}] + if params.get("sig") == SIG3: + return [{"id": f"{C3}@formal_in:1", "sl": None, "mod": "pkg/mod.py"}] + return [] + if "PY_HAS_CFG_NODE" in q and "RETURN n.id" in q: + # Real emitter form: n.id is already the minted can:// URI, no "#". + return [{"id": f"{C2}@entry", "kind": "entry", "sl": None, "el": None}, + {"id": f"{C2}@3:8", "kind": "return", "sl": 3, "el": 3}] + if "r:PY_CFG_NEXT" in q: + return [{"src": f"{C2}@entry", "dst": f"{C2}@3:8", "kind": "fallthrough", + "var": None, "prov": None}] + if "r:PY_CDG" in q or "r:PY_DDG" in q: + return [] + if "PY_PARAM_IN" in q: + return [{"src": f"{C2}@3:8/actual_in:1", + "dst": f"{C3}@formal_in:1", "var": None}] + if "PY_PARAM_OUT" in q or "PY_SUMMARY" in q: + return [] + if "n.start_line = $line" in q: + return [{"id": f"{C2}@3:8", "mod": "pkg/mod.py", "sl": 3}] + raise AssertionError(f"unstubbed query: {q}") + + monkeypatch.setattr(b, "_run", fake_run) + return b + + +def test_is_a_provider(backend): + assert isinstance(backend, ProgramGraphProvider) + + +def test_max_level_derived_from_overlay(backend): + assert backend.max_level() == 4 + + +def test_program_graph_translates_to_can_uris(backend): + g = backend.program_graph(C2) + assert set(g.nodes) == {f"{C2}@entry", f"{C2}@3:8"} + (d,) = g.get_edge_data(f"{C2}@entry", f"{C2}@3:8").values() + assert d["family"] == "cfg" and d["kind"] == "fallthrough" + + +def test_sdg_edges_translated_and_kinded(backend): + edges = list(backend.sdg_edges()) + assert edges and edges[0].kind == "param_in" + assert edges[0].src == f"{C2}@3:8/actual_in:1" + + +def test_resolve_location_orders_by_parsed_col(backend): + assert backend.resolve_location("pkg/mod.py", 3) == [f"{C2}@3:8"] + + +def test_resolve_location_basename_suffix_match(backend): + # A basename/suffix seed must behave identically to the fully-qualified module path — the + # same latitude the local backend's mixin already gives (#270 final review Finding 2). + assert backend.resolve_location("mod.py", 3) == [f"{C2}@3:8"] + + +def test_source_slice_lossy_code_none(backend): + fl, code = backend.source_slice(f"{C2}@3:8") + assert fl == "pkg/mod.py:3" and code is None + + +def test_source_slice_degrades_to_module_path_on_synthetic_formal_in_vertex(backend): + # A resolved callable (sig_is_None guard doesn't hide it) whose vertex is a synthetic + # @formal_in:N body node — it EXISTS but carries no start_line of its own, so per the + # adjudicated contract (#270 final review Finding 3) this degrades to (module_path, None), + # matching the local backend's mixin exactly — never (None, None) and never a fabricated + # line parsed out of the "formal_in" key shape. + assert backend.source_slice(f"{C3}@formal_in:1") == ("pkg/mod.py", None) + + +def test_source_slice_unknown_vertex_is_none_none(backend): + # A vertex whose callable can't be resolved at all (not a can:// id this app knows about) + # is a structural non-match, not a synthetic-vertex degrade — (None, None), never fabricated. + assert backend.source_slice("can://nonexistent@0:0") == (None, None) + + +def test_callable_of_partitions_at_first_at(backend): + assert backend.callable_of(f"{C2}@3:8/actual_in:1") == C2 + + +def test_to_uri_translates_dotted_sig_hash_form_defensively(backend): + # The analyzer repo's schema.py comment (and this file's original brief) documents a + # "#"-separated dotted-sig PyCFGNode.id form; never observed on a real graph (issue #295 — + # the real emitter stores the can:// id directly, exercised by every other test above), but + # _to_uri keeps translating it correctly as a defensive fallback should some emitter version + # ever actually produce it. + assert backend._to_uri(f"{SIG2}#3:8") == f"{C2}@3:8" + assert backend._to_uri(f"{SIG2}#@entry") == f"{C2}@entry" diff --git a/tests/graph/test_py_parity_live.py b/tests/graph/test_py_parity_live.py new file mode 100644 index 0000000..b4ca729 --- /dev/null +++ b/tests/graph/test_py_parity_live.py @@ -0,0 +1,203 @@ +################################################################################ +# Copyright IBM Corporation 2026 +# +# 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. +################################################################################ + +"""Live dual-backend parity for the five verbs' data seam (env-gated, mirrors +tests/analysis/python/test_python_neo4j_backend.py). Writes the pyfix sources from the +py-a4.json fixture to a temp project, runs the real analyzer at level 4, emits to Neo4j +in-process from that SAME analysis, then asserts each provider primitive agrees modulo +documented lossiness (Neo4j source_slice code is None). + +The whole module is skipped unless CLDK_TEST_NEO4J_URI is set. Point it at a server with: + + CLDK_TEST_NEO4J_URI=bolt://localhost:7687 \ + CLDK_TEST_NEO4J_USER=neo4j \ + CLDK_TEST_NEO4J_PASSWORD=testpassword \ + uv run pytest tests/graph/test_py_parity_live.py -v + +(e.g. `podman run -d -p 7687:7687 -e NEO4J_AUTH=neo4j/testpassword neo4j:5` — a DEDICATED +container is still advised, even though the analyzer's bolt writer scopes its orphan-module +prune to THIS application's own app_name (codeanalyzer/neo4j/bolt.py's full-run prune reads +`WHERE ... app=app_name`, never touching other applications' data) — a fresh container just +keeps this suite free of any stale state left over from a previous run.) +""" + +import json +import os +from pathlib import Path + +import pytest + +RES = Path(__file__).parent.parent / "resources" / "cpg" +NEO4J_URI = os.environ.get("CLDK_TEST_NEO4J_URI") +pytestmark = pytest.mark.skipif(not NEO4J_URI, reason="set CLDK_TEST_NEO4J_URI to run") + + +@pytest.fixture(scope="module") +def both_backends(tmp_path_factory): + """(local, remote): the real in-process PyCodeanalyzer and a PyNeo4jBackend over the + SAME analysis, projected to Neo4j out of band.""" + payload = json.loads((RES / "py-a4.json").read_text()) + proj = tmp_path_factory.mktemp("pyfix") + for path, mod in payload["application"]["symbol_table"].items(): + f = proj / path + f.parent.mkdir(parents=True, exist_ok=True) + f.write_text(mod["source"] or "") + + from cldk.analysis import AnalysisLevel + from cldk.analysis.python.codeanalyzer.codeanalyzer import PyCodeanalyzer + + local = PyCodeanalyzer( + project_dir=proj, + analysis_level=AnalysisLevel.system_dependency_graph, + analysis_json_path=None, + eager_analysis=True, + ) + + # Populate Neo4j from the SAME analysis `local` already ran — never a second analyze(): + # jedi call-resolution is nondeterministic across runs and a re-analyze would masquerade + # as a backend diff. codeanalyzer-python 1.0.2's emit_neo4j takes the v2 `Analysis` + # envelope (schema_version/max_level/analyzer/application), not a bare PyApplication — + # confirmed against its real signature and tests/analysis/python/test_python_neo4j_backend.py + # (the emit-API sketch in the task brief predates that envelope). `analyzer` lives only on + # the envelope Codeanalyzer.analyze() built (not on PyApplication itself), and `local` + # doesn't keep that envelope around, so it defaults here — pure metadata, not read by + # emit_neo4j's projection logic. Wrapping `local`'s own already-analyzed application in a + # fresh envelope reuses the identical analysis result (zero re-analysis). + from codeanalyzer.neo4j.emit import emit_neo4j + from codeanalyzer.options import AnalysisOptions + from codeanalyzer.schema.py_schema import Analysis + + neo4j_user = os.environ.get("CLDK_TEST_NEO4J_USER", "neo4j") + neo4j_password = os.environ.get("CLDK_TEST_NEO4J_PASSWORD", "neo4j") + + # app_name MUST equal proj.name (the same value PyCodeanalyzer's own AnalysisOptions + # defaulted to at analysis time — cldk never sets app_name, so codeanalyzer-python + # falls back to `self.project_dir.name`). emit_neo4j's assign_ids() is only idempotent + # when re-run with the SAME app_name: it re-stamps every module/class/callable `.id` in + # place on `local.application`, but does NOT touch the already-baked-in cfg/ddg/param_in/ + # param_out edge src/dst strings from the original analysis. A different app_name here + # would rename callable ids out from under those edges, corrupting `local`'s own identity + # consistency as a side effect of populating Neo4j (a real footgun in this two-actor + # in-place-mutation harness, hand-verified against codeanalyzer-python 1.0.2's + # `Codeanalyzer.analyze()` / `codeanalyzer.schema.assign_ids.assign_ids`). + app_name = proj.name + analysis = Analysis( + max_level=local.max_level(), + application=local.get_application_view(), + ) + emit_neo4j( + analysis, + AnalysisOptions( + input=proj, + app_name=app_name, + neo4j_uri=NEO4J_URI, + neo4j_user=neo4j_user, + neo4j_password=neo4j_password, + ), + ) + + from cldk.analysis.python.neo4j.neo4j_backend import PyNeo4jBackend + + remote = PyNeo4jBackend( + neo4j_uri=NEO4J_URI, + neo4j_username=neo4j_user, + neo4j_password=neo4j_password, + neo4j_database=None, + application_name=app_name, + ) + yield local, remote + remote.close() + + +def test_max_level_parity(both_backends): + local, remote = both_backends + assert local.max_level() == remote.max_level() == 4 + + +def test_program_graph_parity_per_callable(both_backends): + local, remote = both_backends + for cid in local._index()["callables"]: + lg, rg = local.program_graph(cid), remote.program_graph(cid) + assert set(lg.nodes) == set(rg.nodes), cid + # prov included (as a tuple, so it hashes into the set key) — #270 final review + # Finding 4(a): a backend that dropped/mismatched ddg provenance would previously pass + # this parity check silently since prov wasn't compared at all. + lset = {(u, v, d["family"], d.get("kind"), d.get("var"), tuple(d.get("prov") or [])) + for u, v, d in lg.edges(data=True)} + rset = {(u, v, d["family"], d.get("kind"), d.get("var"), tuple(d.get("prov") or [])) + for u, v, d in rg.edges(data=True)} + assert lset == rset, cid + + +def test_sdg_parity(both_backends): + local, remote = both_backends + key = lambda es: {(e.src, e.dst, e.kind) for e in es} # noqa: E731 + assert key(local.sdg_edges()) == key(remote.sdg_edges()) + + +def test_source_slice_and_callable_of_parity_per_vertex(both_backends): + # #270 final review Finding 4(b): every vertex of every callable must agree on callable_of + # (both backends), and on source_slice's location half (Neo4j's `code` is documented-lossy — + # always None there — so only the (module[:line] | None) half is compared). + local, remote = both_backends + checked = 0 + for cid in local._index()["callables"]: + for v in local.program_graph(cid).nodes(): + assert local.callable_of(v) == remote.callable_of(v), v + l_fl, _ = local.source_slice(v) + r_fl, r_code = remote.source_slice(v) + assert l_fl == r_fl, v + assert r_code is None + checked += 1 + assert checked > 0 # sanity: the fixture actually has vertices to compare + + +def test_resolve_location_parity_for_every_spanned_vertex(both_backends): + # #270 final review Finding 4(b): every (file, line) that owns at least one real vertex must + # return the SAME full hit-list on both backends, not just agree on a single hand-picked + # location (the existing verb-parity tests only ever probe one seed per verb). + local, remote = both_backends + locations = set() + for cid in local._index()["callables"]: + for v in local.program_graph(cid).nodes(): + fl, _ = local.source_slice(v) + if not fl or ":" not in fl: + continue # synthetic vertex (@entry/@exit/formal_*/actual_*) — no line to probe + file, _, line = fl.rpartition(":") + locations.add((file, int(line))) + assert locations # sanity: the fixture actually has spanned (real source line) vertices + for file, line in locations: + assert set(local.resolve_location(file, line)) == set(remote.resolve_location(file, line)), (file, line) + + +def test_verb_parity(both_backends): + from cldk.graph import Engine + + local, remote = both_backends + # A real param_in pair, discovered from the live application rather than hardcoded ids: the + # fixture's app_name is a random tmp-dir name (see both_backends), so can:// ids aren't + # stable across runs and can't be pinned as string literals here. + param_in_edge = next(e for e in local.sdg_edges() if e.kind == "param_in") + for verb, args in ( + ("slice_backward", ("pkg/mod.py:5",)), + ("slice_forward", ("pkg/mod.py:3",)), + ("def_use", ("pkg/mod.py:3",)), + ("control_deps", ("pkg/mod.py:3",)), + ("flows_to", (param_in_edge.src, param_in_edge.dst)), + ): + l = getattr(Engine(local), verb)(*args) # noqa: E741 + r = getattr(Engine(remote), verb)(*args) + assert set(l.uris()) == set(r.uris()), verb