diff --git a/cldk/graph/__init__.py b/cldk/graph/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/cldk/graph/capability.py b/cldk/graph/capability.py new file mode 100644 index 0000000..a95f999 --- /dev/null +++ b/cldk/graph/capability.py @@ -0,0 +1,22 @@ +from __future__ import annotations +from typing import Optional, Dict + + +class CapabilityError(Exception): + pass + + +def require(level_needed: int, provider, *, strict: bool, what: str) -> Optional[Dict]: + available = provider.max_level() + if available >= level_needed: + return None + if strict: + raise CapabilityError( + f"{what} requires analysis level {level_needed}; backend is at level {available}. " + f"Re-analyze at -a {level_needed} or drop strict=True to degrade.") + return { + "requested": level_needed, + "available": available, + "gap": f"{what} requires level {level_needed}; backend at level {available} — " + f"reduced result returned; absence of a result here is UNKNOWN, not safety.", + } diff --git a/cldk/graph/engine.py b/cldk/graph/engine.py new file mode 100644 index 0000000..673900b --- /dev/null +++ b/cldk/graph/engine.py @@ -0,0 +1,196 @@ +# cldk/graph/engine.py +from __future__ import annotations +from typing import Iterable, List, Optional, Tuple +import networkx as nx +from cldk.graph.provider import ProgramGraphProvider, resolve_vertex +from cldk.graph.capability import require +from cldk.graph.result import SliceResult, FlowResult, FlowPath + + +def _filter_edges(g: nx.MultiDiGraph, families: Iterable[str]) -> nx.MultiDiGraph: + fam = set(families) + out = nx.MultiDiGraph() + out.add_nodes_from(g.nodes(data=True)) + for u, v, k, d in g.edges(keys=True, data=True): + if d.get("family") in fam: + out.add_edge(u, v, key=k, **d) + return out + + +_TIER_RANK = {"unresolved": 0, "structural": 1, "resolved": 2} +_RANK_TIER = {v: k for k, v in _TIER_RANK.items()} + +# Provisional witness-enumeration bounds (to be tuned against real graphs in a later +# task): flows_to explores simple paths no deeper than _PATH_CUTOFF hops and stops +# collecting witnesses at _MAX_PATHS; explain()["truncated"] reports whether the +# path cap was hit (depth-cutoff drops are not separately detectable and are folded +# into the same provisional-bounds caveat). +_MAX_PATHS = 1000 +_PATH_CUTOFF = 64 + + +def _ddg_tier(prov) -> str: + # Membership, not exact-list: prov is a provenance set in list form, and + # ["ssa", "points-to"] is STRONGER evidence than ["points-to"] alone. + prov = prov or [] + if "points-to" in prov: + return "resolved" + if "ssa" in prov: + return "structural" + return "unresolved" + + +class Engine: + def __init__(self, provider: ProgramGraphProvider): + self.p = provider + + def _evidence(self, uris, seeds, roles=None, default_role="def"): + # Seeds are always "seed"; other vertices take the verb's default_role + # (slices/flows: "def", control_deps: "control", def_use: "use"). + roles = roles or {} + ev = [] + for u in uris: + fl, code = self.p.source_slice(u) + ev.append({"uri": u, "file_line": fl, "code": code, + "role": "seed" if u in seeds else roles.get(u, default_role)}) + return ev + + def _intra(self, seed, edges, backward, strict, what, interprocedural=None, + default_role="def") -> SliceResult: + note = require(3, self.p, strict=strict, what=what) + want_inter = interprocedural if interprocedural is not None else (self.p.max_level() >= 4) + if interprocedural is True: + inter_note = require(4, self.p, strict=strict, what=f"interprocedural {what}") + if inter_note: + note = inter_note + want_inter = False + # Only dataflow (param_in/param_out/summary) crosses callable boundaries, so the + # sdg overlay is additionally gated on the ddg family being requested: a cfg- or + # cdg-only slice never crosses, even on an L4 backend. want_inter is the single + # source of truth — it both gates the overlay and feeds explain()["interprocedural"]. + want_inter = want_inter and self.p.max_level() >= 4 and "ddg" in set(edges) + seeds = resolve_vertex(self.p, seed) + g = _filter_edges(self.p.program_graph(self.p.callable_of(seeds[0])), edges) + if want_inter: + for e in self.p.sdg_edges(): + g.add_edge(e.src, e.dst, family="sdg", kind=getattr(e, "kind", None), + var=getattr(e, "var", None), prov=getattr(e, "prov", [])) + walk = g.reverse(copy=False) if backward else g + reached = set(seeds) + for s in seeds: + if s in walk: + reached |= nx.descendants(walk, s) + # seed-consistency (Task 4 fix, carried here): evidence/uris must equal subgraph nodes, + # and a seed is trivially in its own slice. + sub = g.subgraph(reached & set(g.nodes())).copy() # MultiDiGraph + for s in seeds: + if s not in sub: + sub.add_node(s, kind="seed") + ev_nodes = sorted(sub.nodes()) # deterministic; evidence set == subgraph nodes + explain = {"seed": seeds, "direction": "backward" if backward else "forward", + "edges": list(edges), "level": self.p.max_level(), + "vertices": len(sub), "interprocedural": bool(want_inter)} + if note: + explain["degraded"] = note + return SliceResult(subgraph=sub, + evidence=self._evidence(ev_nodes, set(seeds), + default_role=default_role), + _explain=explain) + + def slice_backward(self, seed, *, edges=("cfg", "cdg", "ddg"), + interprocedural: Optional[bool] = None, strict: bool = False) -> SliceResult: + return self._intra(seed, edges, True, strict, "slice_backward", interprocedural) + + def slice_forward(self, seed, *, edges=("cfg", "cdg", "ddg"), + interprocedural: Optional[bool] = None, strict: bool = False) -> SliceResult: + return self._intra(seed, edges, False, strict, "slice_forward", interprocedural) + + def control_deps(self, seed, *, strict: bool = False) -> SliceResult: + # Control dependence is intraprocedural in this model — only dataflow (param/summary) + # crosses boundaries. Force interprocedural=False so the sdg overlay is never merged + # into a pure CDG slice, even on an L4 backend. + return self._intra(seed, ("cdg",), backward=True, strict=strict, + what="control_deps", interprocedural=False, + default_role="control") + + def _dataflow_graph(self, *callable_uris) -> nx.MultiDiGraph: + # Union of the given callables' intra ddg graphs, plus the summary/param_* + # (inter) sdg overlay at L4; below L4 this is intraprocedural ddg only. + g = nx.MultiDiGraph() + for c in dict.fromkeys(callable_uris): # dedupe, keep order + cg = _filter_edges(self.p.program_graph(c), ("ddg",)) + g.add_nodes_from(cg.nodes(data=True)) + for u, v, k, d in cg.edges(keys=True, data=True): + g.add_edge(u, v, key=k, **d) + if self.p.max_level() >= 4: + for e in self.p.sdg_edges(): + g.add_edge(e.src, e.dst, family="sdg", kind=getattr(e, "kind", None), + var=getattr(e, "var", None), prov=getattr(e, "prov", [])) + return g + + def flows_to(self, source_seed, sink_seed, *, strict: bool = False) -> FlowResult: + # Full flows_to semantics are interprocedural (ddg + param_in/param_out/summary), + # which is L4. Below that, non-strict degrades honestly: the note is attached and + # the intra-only ddg witnesses that CAN be computed are still returned. + note = require(4, self.p, strict=strict, what="flows_to") + src = resolve_vertex(self.p, source_seed)[0] + dst = resolve_vertex(self.p, sink_seed)[0] + # A sink in a different callable is reachable via param_in/param_out/summary, + # so the dataflow graph must span BOTH endpoint callables. (Multi-hop flows + # through a THIRD callable's interior need the whole-program graph — deferred.) + g = self._dataflow_graph(self.p.callable_of(src), self.p.callable_of(dst)) + paths: List[FlowPath] = [] + reached = set() + truncated = False + if src in g and dst in g: + # A MultiDiGraph enumerates a route once per parallel-edge combination, yielding + # byte-identical duplicate witnesses. Enumerate over a plain-DiGraph VIEW (one path + # per distinct node route) and read per-hop parallel evidence from the MultiDiGraph g. + routes = nx.DiGraph(g) + for path in nx.all_simple_paths(routes, src, dst, cutoff=_PATH_CUTOFF): + if len(paths) >= _MAX_PATHS: + truncated = True + break + hops, tiers = [], [] + for a, b in zip(path, path[1:]): + # MultiDiGraph: get_edge_data returns {key: attrdict} over parallel edges. + # Pick the strongest-confidence parallel edge as the hop's evidence (the step + # is as strong as its best evidence; the path is as weak as its weakest step). + parallels = g.get_edge_data(a, b) + best = max(parallels.values(), + key=lambda d: _TIER_RANK[_ddg_tier(d.get("prov", []))]) + t = _ddg_tier(best.get("prov", [])) + tiers.append(t) + # Intra edges report their family (cfg/cdg/ddg have no kind); sdg + # boundary edges report the concrete kind (param_in/param_out/summary). + hops.append({"from": a, "to": b, + "kind": best.get("kind") or best.get("family"), + "var": best.get("var"), "confidence": t}) + conf = _RANK_TIER[min(_TIER_RANK[t] for t in tiers)] if tiers else "unresolved" + paths.append(FlowPath(source=src, sink=dst, hops=hops, confidence=conf)) + reached.update(path) + explain = {"source": src, "sink": dst, "level": self.p.max_level(), + "paths": len(paths), "truncated": truncated} + if note: + explain["degraded"] = note + sub = g.subgraph(reached).copy() + return FlowResult(subgraph=sub, evidence=self._evidence(sorted(sub.nodes()), {src, dst}), + _explain=explain, paths=paths) + + def def_use(self, seed, *, strict: bool = False) -> FlowResult: + note = require(3, self.p, strict=strict, what="def_use") + s = resolve_vertex(self.p, seed)[0] + # NOTE: currently scoped to the seed's callable plus sdg endpoints; uses inside + # OTHER callables' interiors arrive with the whole-program dataflow graph (deferred). + g = self._dataflow_graph(self.p.callable_of(s)) + reached = {s} | (nx.descendants(g, s) if s in g else set()) + sub = g.subgraph(reached & set(g.nodes())).copy() + if s not in sub: # a seed is trivially in its own def-use result + sub.add_node(s, kind="seed") + explain = {"seed": s, "level": self.p.max_level(), "vertices": len(sub)} + if note: + explain["degraded"] = note + return FlowResult(subgraph=sub, + evidence=self._evidence(sorted(sub.nodes()), {s}, + default_role="use"), + _explain=explain, paths=[]) diff --git a/cldk/graph/provider.py b/cldk/graph/provider.py new file mode 100644 index 0000000..eb5ea3e --- /dev/null +++ b/cldk/graph/provider.py @@ -0,0 +1,52 @@ +from __future__ import annotations +import re +from abc import ABC, abstractmethod +from typing import List, Tuple, Iterable, Optional, Any +import networkx as nx + +_LOC = re.compile(r"^(?P.+?):(?P\d+)(?::(?P\d+))?$") + + +class ProgramGraphProvider(ABC): + """The per-backend data seam the shared engine consumes. Implemented by local backends + (from cpg models) and Neo4j backends (from Cypher). Traversal lives in the engine, not here. + + Below the level a verb requires, the engine gates via require(...); providers should + still answer program_graph/resolve_location/callable_of structurally — none of these + ever need L3+ data to do so.""" + + @abstractmethod + def program_graph(self, callable_uri: str) -> nx.MultiDiGraph: + """Parallel cfg/cdg/ddg edges between the same vertex pair must stay distinct + edges, each carrying its own family/var/prov/kind.""" + @abstractmethod + def sdg_edges(self) -> Iterable[Any]: ... + @abstractmethod + def resolve_location(self, file: str, line: int, col: Optional[int] = None) -> List[str]: ... + @abstractmethod + def source_slice(self, vertex_uri: str) -> Tuple[Optional[str], Optional[str]]: ... + @abstractmethod + def callable_of(self, vertex_uri: str) -> Optional[str]: ... + @abstractmethod + def max_level(self) -> int: ... + + +def resolve_vertex(provider: ProgramGraphProvider, seed: Any) -> List[str]: + """Normalize a polymorphic seed to vertex ids: a BodyNode-like object (has .id), a can:// id + string, or a 'file:line[:col]' location string.""" + if hasattr(seed, "id"): + return [seed.id] + if isinstance(seed, str): + if seed.startswith("can://"): + return [seed] + m = _LOC.match(seed) + if m: + col = int(m["col"]) if m["col"] is not None else None + found = provider.resolve_location(m["file"], int(m["line"]), col) + if not found: + # An ordinary user miss (no vertex at that line) must surface as a clean + # ValueError here — every verb indexes the result, and [] would IndexError. + raise ValueError(f"no vertex at location {seed!r}") + return found + raise ValueError(f"cannot resolve seed to a vertex: {seed!r} " + f"(expected 'file:line[:col]', a can:// id, or a body node)") diff --git a/cldk/graph/result.py b/cldk/graph/result.py new file mode 100644 index 0000000..5d61be0 --- /dev/null +++ b/cldk/graph/result.py @@ -0,0 +1,53 @@ +from __future__ import annotations +import json +from dataclasses import dataclass, field, asdict +from typing import List, Dict, Literal +import networkx as nx + +Confidence = Literal["resolved", "structural", "unresolved"] + + +@dataclass(frozen=True) +class FlowPath: + source: str + sink: str + hops: List[Dict] = field(default_factory=list) + confidence: Confidence = "unresolved" + + +@dataclass +class GraphResult: + subgraph: nx.MultiDiGraph + evidence: List[Dict] + _explain: Dict + + def uris(self) -> List[str]: + return [e["uri"] for e in self.evidence] + + def explain(self) -> Dict: + return dict(self._explain) + + def to_json(self) -> str: + return json.dumps({"evidence": self.evidence, "explain": self._explain, + "vertices": list(self.subgraph.nodes)}, sort_keys=True) + + def __len__(self) -> int: + return self.subgraph.number_of_nodes() + + def __bool__(self) -> bool: + return self.subgraph.number_of_nodes() > 0 + + +@dataclass +class SliceResult(GraphResult): + pass + + +@dataclass +class FlowResult(GraphResult): + paths: List[FlowPath] = field(default_factory=list) + + def to_json(self) -> str: + base = json.loads(super().to_json()) + base["paths"] = [asdict(p) for p in self.paths] + return json.dumps(base, sort_keys=True) diff --git a/tests/graph/__init__.py b/tests/graph/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/graph/test_capability.py b/tests/graph/test_capability.py new file mode 100644 index 0000000..1925306 --- /dev/null +++ b/tests/graph/test_capability.py @@ -0,0 +1,23 @@ +import pytest +from cldk.graph.capability import require, CapabilityError + + +class P: + def __init__(self, lvl): self._l = lvl + def max_level(self): return self._l + + +def test_satisfied_returns_none(): + assert require(3, P(4), strict=False, what="slice_backward") is None + + +def test_degrade_returns_note(): + note = require(4, P(3), strict=False, what="interprocedural flows_to") + assert note["requested"] == 4 and note["available"] == 3 + assert "UNKNOWN, not safety" in note["gap"] + + +def test_strict_raises(): + with pytest.raises(CapabilityError) as e: + require(4, P(3), strict=True, what="flows_to") + assert "level 4" in str(e.value) diff --git a/tests/graph/test_engine_flows.py b/tests/graph/test_engine_flows.py new file mode 100644 index 0000000..3b6b59a --- /dev/null +++ b/tests/graph/test_engine_flows.py @@ -0,0 +1,156 @@ +# tests/graph/test_engine_flows.py +import networkx as nx +import pytest +from cldk.graph.engine import Engine, _ddg_tier +from cldk.graph.provider import ProgramGraphProvider +from cldk.graph.capability import CapabilityError +from tests.graph.test_engine_slice import OneCallableProvider + + +def test_ddg_tier_uses_membership_not_exact_list(): + # I8: prov is a provenance SET in list form. ["ssa", "points-to"] carries strictly + # MORE evidence than ["points-to"] alone and must rank "resolved" — an exact-list + # comparison would let the stronger provenance fall through to "unresolved". + assert _ddg_tier(["ssa", "points-to"]) == "resolved" + assert _ddg_tier(["points-to"]) == "resolved" + assert _ddg_tier(["ssa"]) == "structural" + assert _ddg_tier([]) == "unresolved" + + +class ParallelEdgeProvider(ProgramGraphProvider): + # ONE node route s1 -> s2 -> s3, but the s1->s2 hop carries TWO parallel ddg edges + # (var x via ssa, var y via points-to). A MultiDiGraph enumerates the route once per + # parallel edge; the engine must collapse to one witness per distinct node route. + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + for n in ["c@1:0", "c@2:0", "c@3:0"]: + g.add_node(n, kind="statement", span=None) + g.add_edge("c@1:0", "c@2:0", key="ddg:x", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@1:0", "c@2:0", key="ddg:y", family="ddg", var="y", prov=["points-to"]) + g.add_edge("c@2:0", "c@3:0", key="ddg", family="ddg", var="z", prov=["ssa"]) + return g + + def sdg_edges(self): return [] + def resolve_location(self, file, line, col=None): return [f"c@{line}:{col or 0}"] + def source_slice(self, vertex_uri): return (f"m:{vertex_uri}", vertex_uri) + def callable_of(self, vertex_uri): return "c" + def max_level(self): return 3 + + +def test_flows_to_finds_witness_with_min_confidence(): + e = Engine(OneCallableProvider()) + r = e.flows_to("m:1", "m:3") # s1 -> s2 (ssa) -> s3 (points-to); min = structural + assert bool(r) is True + assert len(r.paths) == 1 + hops = r.paths[0].hops + assert [h["from"] for h in hops] == ["c@1:0", "c@2:0"] + assert [h["to"] for h in hops] == ["c@2:0", "c@3:0"] + assert [h["kind"] for h in hops] == ["ddg", "ddg"] + assert [h["var"] for h in hops] == ["x", "y"] + assert [h["confidence"] for h in hops] == ["structural", "resolved"] + assert r.paths[0].confidence == "structural" + + +def test_flows_to_no_path_is_falsy(): + e = Engine(OneCallableProvider()) + r = e.flows_to("m:3", "m:1") # no forward ddg path s3 -> s1 + assert not r.paths + assert bool(r) is False + + +def test_flows_to_dedups_parallel_edge_routes(): + # one node route, two parallel ddg edges on the first hop -> exactly one witness. + e = Engine(ParallelEdgeProvider()) + r = e.flows_to("m:1", "m:3") + assert len(r.paths) == 1 + p = r.paths[0] + assert [h["from"] for h in p.hops] == ["c@1:0", "c@2:0"] + assert [h["to"] for h in p.hops] == ["c@2:0", "c@3:0"] + # per-hop confidence uses the BEST (max-tier) parallel: s1->s2 resolved (points-to + # beats ssa), s2->s3 structural. Path confidence is the min over hops. + assert [h["confidence"] for h in p.hops] == ["resolved", "structural"] + assert p.confidence == "structural" + + +def test_flows_to_on_l3_degrades_but_still_returns_intra_paths(): + # C1: full flows_to semantics are interprocedural (ddg + param_in/param_out/summary), + # which is L4. On an L3 backend the non-strict call must attach a degraded note AND + # still return the intraprocedural ddg witnesses it can compute — honest degrade, + # not silent completeness and not a refusal. + e = Engine(OneCallableProvider()) # L3 backend + r = e.flows_to("m:1", "m:3") + assert "degraded" in r.explain() + assert r.explain()["degraded"]["requested"] == 4 + assert len(r.paths) == 1 # intra ddg witness still computed + + +def test_flows_to_strict_on_l3_raises(): + e = Engine(OneCallableProvider()) # L3 backend + with pytest.raises(CapabilityError): + e.flows_to("m:1", "m:3", strict=True) + + +class DiamondProvider(ProgramGraphProvider): + # TWO distinct node routes: c@s -> c@a -> c@t and c@s -> c@b -> c@t. + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + g.add_edge("c@s", "c@a", key="ddg", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@a", "c@t", key="ddg", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@s", "c@b", key="ddg", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@b", "c@t", key="ddg", family="ddg", var="x", prov=["ssa"]) + return g + def sdg_edges(self): return [] + def resolve_location(self, file, line, col=None): return [f"c@{line}"] + def source_slice(self, vertex_uri): return (vertex_uri, vertex_uri) + def callable_of(self, vertex_uri): return "c" + def max_level(self): return 4 + + +def test_flows_to_sets_truncated_when_path_cap_hit(monkeypatch): + # I5: witness enumeration is bounded; when the cap drops paths the result must SAY + # so via explain()["truncated"], instead of silently presenting a partial set as + # complete. Shrink the cap to 1 so the diamond's second route is dropped. + import cldk.graph.engine as eng + monkeypatch.setattr(eng, "_MAX_PATHS", 1) + e = Engine(DiamondProvider()) + class S: id = "c@s" + class T: id = "c@t" + r = e.flows_to(S(), T()) + assert len(r.paths) == 1 # capped at _MAX_PATHS + assert r.explain()["truncated"] is True + + +def test_flows_to_not_truncated_within_bounds(): + e = Engine(DiamondProvider()) + class S: id = "c@s" + class T: id = "c@t" + r = e.flows_to(S(), T()) + assert len(r.paths) == 2 # both diamond routes enumerated + assert r.explain()["truncated"] is False + + +def test_def_use_returns_downstream_uses(): + e = Engine(OneCallableProvider()) + r = e.def_use("m:1") # def at s1 flows to s2, s3 + assert set(r.uris()) == {"c@1:0", "c@2:0", "c@3:0"} + + +def test_def_use_evidence_role_is_use(): + # I4: downstream vertices in a def_use result are USES of the seed's definition. + e = Engine(OneCallableProvider()) + r = e.def_use("m:1") + roles = {ev["uri"]: ev["role"] for ev in r.evidence} + assert roles["c@1:0"] == "seed" + assert roles["c@2:0"] == "use" + assert roles["c@3:0"] == "use" + + +def test_def_use_seed_absent_is_consistent(): + # seed resolving to a vertex not in the dataflow graph is still in its own result; + # uris()/evidence must equal the subgraph node set (no uris/bool contradiction). + e = Engine(OneCallableProvider()) + r = e.def_use("m:99") # c@99:0 is not a node in the ddg graph + assert set(r.uris()) == set(r.subgraph.nodes()) + assert "c@99:0" in set(r.uris()) + assert bool(r) is True + assert len(r) == r.subgraph.number_of_nodes() diff --git a/tests/graph/test_engine_interproc.py b/tests/graph/test_engine_interproc.py new file mode 100644 index 0000000..3f4f1f7 --- /dev/null +++ b/tests/graph/test_engine_interproc.py @@ -0,0 +1,156 @@ +# tests/graph/test_engine_interproc.py +import networkx as nx +from cldk.graph.engine import Engine +from cldk.graph.provider import ProgramGraphProvider +from cldk.graph.capability import CapabilityError +import pytest + + +class _Edge: + def __init__(self, src, dst): self.src, self.dst, self.var, self.prov = src, dst, "a", ["points-to"] + + +class TwoCallableProvider(ProgramGraphProvider): + # caller c: c@call --param_in--> callee d; d@ret --param_out--> c@after + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() # engine's _filter_edges iterates edges(keys=True) + if callable_uri == "c": + g.add_edge("c@call", "c@after", key="ddg", family="ddg", var="a", prov=["points-to"]) + else: + g.add_node("d@ret", kind="statement", span=None) + return g + def sdg_edges(self): return [_Edge("c@call", "d@in"), _Edge("d@ret", "c@after")] + def resolve_location(self, file, line, col=None): return [f"c@{line}"] + def source_slice(self, vertex_uri): return (vertex_uri, vertex_uri) + def callable_of(self, vertex_uri): return vertex_uri.split("@")[0] + def max_level(self): return 4 + + +def test_interproc_none_crosses_at_l4(): + e = Engine(TwoCallableProvider()) + # resolve_vertex only accepts a .id-bearing object, a can:// id, or a file:line[:col] + # string (see test_provider.py::test_resolve_node_object_uses_id for the same pattern) — + # "c@call" is a raw vertex id, so it must go through the .id-object path. + class Seed: id = "c@call" + r = e.slice_forward(Seed(), edges=("ddg",), interprocedural=None) + assert "d@in" in set(r.uris()) # crossed the param_in boundary + + +def test_explicit_interproc_on_l3_strict_raises(): + class L3(TwoCallableProvider): + def max_level(self): return 3 + with pytest.raises(CapabilityError): + Engine(L3()).slice_forward("c@call", interprocedural=True, strict=True) + + +class _SDGEdge: + def __init__(self, src, dst, kind): + self.src, self.dst, self.kind = src, dst, kind + self.var, self.prov = "a", ["points-to"] + + +class CrossCallableFlowProvider(ProgramGraphProvider): + # caller c: c@src --ddg--> c@call; sdg: c@call --param_in--> d@in; + # callee d: d@in --ddg--> d@sink. The flow c@src -> d@sink exists only if the + # dataflow graph spans BOTH endpoint callables plus the sdg overlay. + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + if callable_uri == "c": + g.add_edge("c@src", "c@call", key="ddg", family="ddg", var="a", prov=["ssa"]) + else: + g.add_edge("d@in", "d@sink", key="ddg", family="ddg", var="a", prov=["ssa"]) + return g + def sdg_edges(self): return [_SDGEdge("c@call", "d@in", "param_in")] + def resolve_location(self, file, line, col=None): return [f"c@{line}"] + def source_slice(self, vertex_uri): return (vertex_uri, vertex_uri) + def callable_of(self, vertex_uri): return vertex_uri.split("@")[0] + def max_level(self): return 4 + + +def test_flows_to_crosses_callable_boundary(): + # C2: a sink in a DIFFERENT callable (reachable via param_in into the callee's + # interior) must be found. Building the dataflow graph from the source's callable + # alone loses the callee's intra ddg edges and yields a false "no flow". + e = Engine(CrossCallableFlowProvider()) + class Src: id = "c@src" + class Snk: id = "d@sink" + r = e.flows_to(Src(), Snk()) + assert len(r.paths) >= 1 # a real cross-callable flow, not empty + p = r.paths[0] + assert [h["from"] for h in p.hops] == ["c@src", "c@call", "d@in"] + assert [h["to"] for h in p.hops] == ["c@call", "d@in", "d@sink"] + + +def test_flow_boundary_hop_reports_sdg_kind(): + # I6: a hop crossing the callable boundary must report WHICH sdg edge carried the + # flow (param_in/param_out/summary), not the opaque family name "sdg". Intra hops + # keep reporting their family ("ddg"). + e = Engine(CrossCallableFlowProvider()) + class Src: id = "c@src" + class Snk: id = "d@sink" + p = e.flows_to(Src(), Snk()).paths[0] + kinds = [h["kind"] for h in p.hops] + assert kinds[0] == "ddg" # intra hop: family + assert kinds[1] in {"param_in", "param_out", "summary"} # boundary hop: sdg kind + assert kinds[1] == "param_in" + assert kinds[2] == "ddg" + + +def test_family_scoped_slice_has_no_sdg_overlay_at_l4(): + # C3: the sdg (dataflow: param_in/param_out/summary) overlay must be gated on the + # ddg family being REQUESTED, not just on level/interprocedural intent. A cfg-only + # backward slice on an L4 backend must not pull in dataflow vertices from other + # callables — only dataflow crosses boundaries, and no dataflow family was asked for. + class CFGProvider(TwoCallableProvider): + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + g.add_edge("c@1", "c@2", key="cfg", family="cfg") + g.add_edge("c@2", "c@3", key="cfg", family="cfg") + return g + def sdg_edges(self): return [_Edge("d@in", "c@2")] # foreign DATAFLOW vertex + e = Engine(CFGProvider()) + class Seed: id = "c@3" + r = e.slice_backward(Seed(), edges=("cfg",)) + assert "d@in" not in set(r.uris()) # no dataflow contamination + assert set(r.uris()) == {"c@1", "c@2", "c@3"} + assert r.explain()["interprocedural"] is False # no dataflow family => no crossing + + +def test_control_deps_stays_intraprocedural_at_l4(): + # Control dependence has NO interprocedural notion in this model — only dataflow + # (param_in/param_out/summary) crosses callable boundaries. control_deps must force + # interprocedural=False; otherwise, on an L4 backend, _intra defaults to want_inter=True and + # merges the sdg dataflow overlay into a pure CDG slice, and the backward walk pulls in + # dataflow-reachable vertices from other callables with no control-dependence relation. + # + # control_deps is a BACKWARD slice, so a leaking sdg edge must be a forward-ANCESTOR edge of + # the seed: d@in -> c@body means d@in reaches c@body, so a backward slice from c@body WOULD + # pull in d@in if the overlay were applied (verified: it leaks against the unfixed code). + class CDGProvider(TwoCallableProvider): + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() # only a control-dependence edge + g.add_edge("c@guard", "c@body", key="cdg", family="cdg") + return g + def sdg_edges(self): return [_Edge("d@in", "c@body")] # cross-callable DATAFLOW + e = Engine(CDGProvider()) + class Seed: id = "c@body" + r = e.control_deps(Seed()) + assert set(r.uris()) == {"c@guard", "c@body"} # only intra cdg reachability, no d@in + assert "d@in" not in set(r.uris()) # sdg dataflow did NOT cross the boundary + assert r.explain()["interprocedural"] is False # control_deps is always intraprocedural + + +def test_control_deps_evidence_role_is_control(): + # I4: a non-seed vertex in a control_deps result is there as a controlling guard, + # not as a definition — its evidence role must say so. + class CDGProvider(TwoCallableProvider): + def program_graph(self, callable_uri): + g = nx.MultiDiGraph() + g.add_edge("c@guard", "c@body", key="cdg", family="cdg") + return g + e = Engine(CDGProvider()) + class Seed: id = "c@body" + r = e.control_deps(Seed()) + roles = {ev["uri"]: ev["role"] for ev in r.evidence} + assert roles["c@body"] == "seed" + assert roles["c@guard"] == "control" diff --git a/tests/graph/test_engine_slice.py b/tests/graph/test_engine_slice.py new file mode 100644 index 0000000..d183995 --- /dev/null +++ b/tests/graph/test_engine_slice.py @@ -0,0 +1,79 @@ +# tests/graph/test_engine_slice.py +import networkx as nx +from cldk.graph.engine import Engine +from cldk.graph.provider import ProgramGraphProvider + + +def _callable_graph(): + # entry -> s1(x=1) -> s2(y=x) -> s3(return y); ddg x:s1->s2, y:s2->s3. + # MultiDiGraph so the cfg fallthrough (s1->s2, s2->s3) and the ddg edges on the + # same statement pairs stay as DISTINCT parallel edges, each keeping its attrs. + g = nx.MultiDiGraph() + for n in ["c@entry", "c@1:0", "c@2:0", "c@3:0"]: + g.add_node(n, kind="statement", span=None) + g.add_edge("c@entry", "c@1:0", key="cfg", family="cfg") + g.add_edge("c@1:0", "c@2:0", key="cfg", family="cfg") + g.add_edge("c@2:0", "c@3:0", key="cfg", family="cfg") + g.add_edge("c@1:0", "c@2:0", key="ddg", family="ddg", var="x", prov=["ssa"]) + g.add_edge("c@2:0", "c@3:0", key="ddg", family="ddg", var="y", prov=["points-to"]) + assert g.number_of_edges() == 5 # genuine parallel edges, not overwrites + return g + + +class OneCallableProvider(ProgramGraphProvider): + def program_graph(self, callable_uri): return _callable_graph() + def sdg_edges(self): return [] + def resolve_location(self, file, line, col=None): return [f"c@{line}:{col or 0}"] + def source_slice(self, vertex_uri): return (f"m:{vertex_uri}", vertex_uri) + def callable_of(self, vertex_uri): return "c" + def max_level(self): return 3 + + +def test_backward_slice_exact_set(): + e = Engine(OneCallableProvider()) + r = e.slice_backward("m:3", edges=("cfg", "ddg")) # seed s3 + assert set(r.uris()) == {"c@3:0", "c@2:0", "c@1:0", "c@entry"} + + +def test_forward_slice_exact_set(): + e = Engine(OneCallableProvider()) + r = e.slice_forward("m:1", edges=("ddg",)) # seed s1, ddg only + assert set(r.uris()) == {"c@1:0", "c@2:0", "c@3:0"} + + +def test_ddg_only_backward_from_s3(): + e = Engine(OneCallableProvider()) + r = e.slice_backward("m:3", edges=("ddg",)) + assert set(r.uris()) == {"c@3:0", "c@2:0", "c@1:0"} # follows ddg chain, not entry + + +def test_family_scoped_slices_differ(): + # cfg and ddg share the endpoint pairs s1->s2 and s2->s3; a MultiDiGraph keeps + # them as distinct parallel edges, so family-scoped slices must NOT collapse. + e = Engine(OneCallableProvider()) + ddg_set = set(e.slice_backward("m:3", edges=("ddg",)).uris()) + cfg_set = set(e.slice_backward("m:3", edges=("cfg",)).uris()) + assert "c@entry" not in ddg_set # entry reachable only via the cfg chain + assert "c@entry" in cfg_set # cfg fallthrough reaches entry + assert ddg_set != cfg_set # families are distinct, not merged + + +def test_slice_evidence_default_role_is_def(): + # I4 guard: slices keep the "def" role for non-seed vertices (only control_deps + # and def_use re-role their evidence). + e = Engine(OneCallableProvider()) + r = e.slice_backward("m:3", edges=("ddg",)) + roles = {ev["uri"]: ev["role"] for ev in r.evidence} + assert roles["c@3:0"] == "seed" + assert roles["c@1:0"] == "def" and roles["c@2:0"] == "def" + + +def test_seed_absent_from_graph_is_consistent(): + # a seed resolving to a vertex not in the callable graph is still in its own + # slice; uris()/evidence must equal the subgraph's node set (no contradiction). + e = Engine(OneCallableProvider()) + r = e.slice_backward("m:99", edges=("cfg", "ddg")) # c@99:0 is not a node + assert set(r.uris()) == set(r.subgraph.nodes()) + assert len(r) == r.subgraph.number_of_nodes() + assert "c@99:0" in set(r.uris()) # seed present in its own slice + assert bool(r) is True # non-empty; uris() and bool() agree diff --git a/tests/graph/test_provider.py b/tests/graph/test_provider.py new file mode 100644 index 0000000..39923aa --- /dev/null +++ b/tests/graph/test_provider.py @@ -0,0 +1,45 @@ +import pytest +from cldk.graph.provider import resolve_vertex, ProgramGraphProvider + + +class FakeProvider(ProgramGraphProvider): + def program_graph(self, callable_uri): ... + def sdg_edges(self): return [] + def resolve_location(self, file, line, col=None): + return [f"can://x/{file}/f@{line}:{col or 0}"] + def source_slice(self, vertex_uri): return ("m.py:1", "code") + def callable_of(self, vertex_uri): return "can://x/f" + def max_level(self): return 4 + + +def test_resolve_location_string(): + p = FakeProvider() + assert resolve_vertex(p, "src/m.py:42") == ["can://x/src/m.py/f@42:0"] + assert resolve_vertex(p, "src/m.py:42:5") == ["can://x/src/m.py/f@42:5"] + + +def test_resolve_can_id_passthrough(): + p = FakeProvider() + assert resolve_vertex(p, "can://x/src/m.py/f@42:5") == ["can://x/src/m.py/f@42:5"] + + +def test_resolve_node_object_uses_id(): + p = FakeProvider() + class N: id = "can://x/src/m.py/f@42:5" + assert resolve_vertex(p, N()) == ["can://x/src/m.py/f@42:5"] + + +def test_resolve_rejects_garbage(): + p = FakeProvider() + with pytest.raises(ValueError): + resolve_vertex(p, 12345) + + +def test_resolve_location_with_no_vertex_raises(): + # I1: resolve_location legitimately returns [] when no vertex sits at that line. + # Every engine verb indexes resolve_vertex(...)[0], so [] must surface as a clean + # ValueError here — not an IndexError at the call site. + class EmptyProvider(FakeProvider): + def resolve_location(self, file, line, col=None): return [] + with pytest.raises(ValueError, match="no vertex at location"): + resolve_vertex(EmptyProvider(), "m.py:99") diff --git a/tests/graph/test_result.py b/tests/graph/test_result.py new file mode 100644 index 0000000..c808016 --- /dev/null +++ b/tests/graph/test_result.py @@ -0,0 +1,45 @@ +import networkx as nx +from cldk.graph.result import GraphResult, SliceResult, FlowResult, FlowPath + + +def _graph(*nodes): + g = nx.DiGraph() + g.add_nodes_from(nodes) + return g + + +def test_graphresult_len_bool_uris(): + g = _graph("a", "b") + r = SliceResult(subgraph=g, evidence=[{"uri": "a"}, {"uri": "b"}], _explain={"level": 3}) + assert len(r) == 2 + assert bool(r) is True + assert r.uris() == ["a", "b"] + assert r.explain() == {"level": 3} + + +def test_empty_result_is_falsy(): + r = SliceResult(subgraph=_graph(), evidence=[], _explain={}) + assert not r + assert len(r) == 0 + + +def test_flowresult_carries_paths_and_serializes(): + p = FlowPath(source="a", sink="c", + hops=[{"from": "a", "to": "b", "kind": "ddg", "var": "x", "confidence": "structural"}], + confidence="structural") + r = FlowResult(subgraph=_graph("a", "b", "c"), + evidence=[{"uri": "a", "file_line": "m.py:1", "code": "x = 1", "role": "seed"}], + _explain={"level": 4}, paths=[p]) + assert r.paths[0].confidence == "structural" + assert '"file_line": "m.py:1"' in r.to_json() + + +def test_flowresult_to_json_includes_paths(): + p = FlowPath(source="a", sink="c", + hops=[{"from": "a", "to": "b", "kind": "ddg", "var": "x", "confidence": "structural"}], + confidence="structural") + r = FlowResult(subgraph=_graph("a", "b", "c"), + evidence=[{"uri": "a"}], _explain={"level": 4}, paths=[p]) + dumped = r.to_json() + assert '"confidence": "structural"' in dumped + assert '"var": "x"' in dumped