diff --git a/codegraph/analysis/dataflow.py b/codegraph/analysis/dataflow.py index d017f4d..838a6bf 100644 --- a/codegraph/analysis/dataflow.py +++ b/codegraph/analysis/dataflow.py @@ -4,23 +4,25 @@ This module exposes two complementary functions: * :func:`match_route` — given a frontend ``FETCH_CALL`` URL + method, find the - qualname of the backend handler whose ``ROUTE`` edge matches. Implemented by - the DF3 stitcher (URL pattern matching with parameter normalisation). + qualname of the backend handler whose ``ROUTE`` edge matches. * :func:`trace` — given an entry symbol (function qualname or ``url:METHOD path`` shape), walk the call graph + cross-layer edges to produce an ordered :class:`DataFlow`. Implemented by DF4. -The dataclasses are stable contract — agents fill in the function bodies but -never modify the shapes here without coordination. +The dataclasses are stable contract — never modify the shapes here without +coordination. """ from __future__ import annotations +import re from dataclasses import dataclass, field from typing import Any import networkx as nx +from codegraph.graph.schema import EdgeKind, NodeKind + @dataclass class FlowHop: @@ -81,24 +83,156 @@ def to_dict(self) -> dict[str, Any]: } +_PLACEHOLDER_RE = re.compile(r"^(\{[^}]*\}|\$\{[^}]*\}|:[A-Za-z_][A-Za-z0-9_]*|-?\d+)$") + + +def _strip_query_fragment(path: str) -> str: + """Drop ``?query`` and ``#fragment``; collapse trailing slash.""" + for sep in ("?", "#"): + if sep in path: + path = path.split(sep, 1)[0] + if len(path) > 1 and path.endswith("/"): + path = path.rstrip("/") + return path + + +def _segments(path: str) -> list[str]: + """Split ``/api/users/{id}`` into ``['api', 'users', '{id}']``.""" + return [s for s in _strip_query_fragment(path).split("/") if s] + + +def _is_placeholder(seg: str) -> bool: + """A segment is a placeholder if it's purely numeric, or wrapped in + ``{...}`` / ``${...}`` / leading ``:`` (Express style).""" + return bool(_PLACEHOLDER_RE.match(seg)) + + +def _normalise_path(path: str) -> list[str]: + """Return the list of normalised segments where every placeholder + becomes the marker ``{*}`` so two paths with different placeholder + syntaxes compare equal segment-by-segment.""" + return ["{*}" if _is_placeholder(s) else s for s in _segments(path)] + + +def _path_specificity(segs: list[str]) -> int: + """How "concrete" a path is — more literal segments means more specific. + Used to break ties when two routes match the same fetch.""" + return sum(1 for s in segs if s != "{*}") + + +def _route_candidates(graph: nx.MultiDiGraph) -> list[tuple[str, str, str]]: + """Yield ``(handler_qualname, method, path)`` for every ROUTE edge. + + ROUTE edges go from a backend handler FUNCTION/METHOD to a synthetic + target node with id ``route::::``. The handler qualname + is the source node's qualname. + """ + out: list[tuple[str, str, str]] = [] + for src, _dst, key, edata in graph.edges(keys=True, data=True): + if key != EdgeKind.ROUTE.value: + continue + meta = edata.get("metadata") or {} + if not isinstance(meta, dict): + continue + method = str(meta.get("method") or "").upper() + path = str(meta.get("path") or "") + if not method or not path: + continue + attrs = graph.nodes.get(src) or {} + qn = str(attrs.get("qualname") or src) + out.append((qn, method, path)) + return out + + +def _handler_param_names(graph: nx.MultiDiGraph, handler_qn: str) -> list[str]: + """Extract parameter names for the handler function, for body-key + overlap scoring. Looks up the node by qualname and reads + ``metadata.params`` (populated by DF0).""" + for _nid, attrs in graph.nodes(data=True): + if str(attrs.get("qualname") or "") != handler_qn: + continue + kind = str(attrs.get("kind") or "") + if kind not in (NodeKind.FUNCTION.value, NodeKind.METHOD.value): + continue + meta = attrs.get("metadata") or {} + params = meta.get("params") or [] if isinstance(meta, dict) else [] + names: list[str] = [] + for p in params: + if isinstance(p, dict): + name = str(p.get("name") or "").lstrip("*") + if name and name not in ("self", "cls"): + names.append(name) + return names + return [] + + def match_route( graph: nx.MultiDiGraph, fetch_url: str, fetch_method: str = "GET", + *, + body_keys: list[str] | None = None, ) -> tuple[str, float] | None: - """Return ``(handler_qualname, confidence)`` for the route matching this - frontend fetch, or ``None`` if no route matches. + """Return ``(handler_qualname, confidence)`` for the backend ROUTE that + matches this frontend fetch, or ``None`` if no route matches. + + Confidence rubric: + * **1.0** — exact literal-segment match, no placeholders involved + * **0.9** — placeholders in either side normalise to the same shape + * up to **+0.05** bonus if the fetch's ``body_keys`` overlap with the + handler's parameter names (clamped at 0.95 / 1.0 ceilings) + * **0.5** — only a path *prefix* matches (last-resort fuzzy) + * **None** — method mismatch or no overlap - Implemented by DF3 (agent A1). Stub returns ``None`` so DF4 can build - against the contract before A1 lands. + Trailing slashes, query strings, and fragments are stripped before + matching. Method comparison is case-insensitive. - Confidence rubric (DF3 will encode): - * 1.0 — exact path + method match - * 0.9 — path with placeholders matches (``/users/{id}`` vs ``/users/42``) - * 0.7 — method matches, path matches with body-key heuristic - * 0.5 — only path prefix matches (last-resort) + When multiple routes match at the same top confidence, the more + specific one (more literal segments) wins. """ - return None + method = (fetch_method or "GET").upper() + fetch_segs = _normalise_path(fetch_url) + raw_fetch_segs = _segments(fetch_url) + fetch_is_literal = all(not _is_placeholder(s) for s in raw_fetch_segs) + + best: tuple[str, float, int] | None = None # (qn, score, specificity) + + for handler_qn, route_method, route_path in _route_candidates(graph): + if route_method != method: + continue + route_segs = _normalise_path(route_path) + raw_route_segs = _segments(route_path) + route_is_literal = all(not _is_placeholder(s) for s in raw_route_segs) + + if fetch_segs == route_segs: + base = 1.0 if (fetch_is_literal and route_is_literal) else 0.9 + specificity = _path_specificity(route_segs) + elif ( + len(fetch_segs) >= len(route_segs) + and fetch_segs[: len(route_segs)] == route_segs + and len(route_segs) > 0 + ): + base = 0.5 + specificity = _path_specificity(route_segs) + else: + continue + + # Body-key bonus: any overlap with handler params nudges score up. + if body_keys: + param_names = _handler_param_names(graph, handler_qn) + overlap = set(body_keys) & set(param_names) + if overlap: + cap = 1.0 if base >= 1.0 else (0.95 if base >= 0.9 else 0.7) + base = min(cap, base + 0.05) + + if best is None or base > best[1] or ( + base == best[1] and specificity > best[2] + ): + best = (handler_qn, base, specificity) + + if best is None: + return None + return (best[0], best[1]) def trace( diff --git a/tests/test_dataflow_match.py b/tests/test_dataflow_match.py new file mode 100644 index 0000000..bbb4dbb --- /dev/null +++ b/tests/test_dataflow_match.py @@ -0,0 +1,256 @@ +"""DF3 — URL stitcher tests for codegraph.analysis.dataflow.match_route.""" +from __future__ import annotations + +import networkx as nx + +from codegraph.analysis.dataflow import match_route +from codegraph.graph.schema import EdgeKind, NodeKind + + +def _graph_with_route( + handler_qn: str, + method: str, + path: str, + *, + handler_params: list[dict[str, str | None]] | None = None, +) -> nx.MultiDiGraph: + """Build a minimal graph with one HANDLER + one ROUTE edge.""" + g: nx.MultiDiGraph = nx.MultiDiGraph() + handler_id = f"handler::{handler_qn}" + g.add_node( + handler_id, + qualname=handler_qn, + name=handler_qn.rsplit(".", 1)[-1], + kind=NodeKind.FUNCTION.value, + file=f"app/{handler_qn.split('.')[-1]}.py", + line_start=1, + metadata={ + "role": "HANDLER", + "params": handler_params or [], + }, + ) + route_id = f"route::{method.upper()}::{path}" + g.add_node( + route_id, + qualname=route_id, + name=f"{method.upper()} {path}", + kind=NodeKind.VARIABLE.value, + metadata={"synthetic_kind": "ROUTE"}, + ) + g.add_edge( + handler_id, + route_id, + key=EdgeKind.ROUTE.value, + kind=EdgeKind.ROUTE.value, + metadata={"method": method.upper(), "path": path, "framework": "fastapi"}, + ) + return g + + +def _add_route( + g: nx.MultiDiGraph, + handler_qn: str, + method: str, + path: str, + *, + handler_params: list[dict[str, str | None]] | None = None, +) -> None: + handler_id = f"handler::{handler_qn}" + g.add_node( + handler_id, + qualname=handler_qn, + name=handler_qn.rsplit(".", 1)[-1], + kind=NodeKind.FUNCTION.value, + file=f"app/{handler_qn.split('.')[-1]}.py", + line_start=1, + metadata={"role": "HANDLER", "params": handler_params or []}, + ) + route_id = f"route::{method.upper()}::{path}" + g.add_node( + route_id, + qualname=route_id, + kind=NodeKind.VARIABLE.value, + metadata={"synthetic_kind": "ROUTE"}, + ) + g.add_edge( + handler_id, + route_id, + key=EdgeKind.ROUTE.value, + kind=EdgeKind.ROUTE.value, + metadata={"method": method.upper(), "path": path, "framework": "fastapi"}, + ) + + +# ---- Test 1: exact literal match ---- +def test_exact_literal_match_confidence_1() -> None: + g = _graph_with_route("api.get_health", "GET", "/api/health") + out = match_route(g, "/api/health", "GET") + assert out is not None + assert out[0] == "api.get_health" + assert out[1] == 1.0 + + +# ---- Test 2: placeholder in route, literal in fetch ---- +def test_route_placeholder_vs_literal_fetch_confidence_0_9() -> None: + g = _graph_with_route("api.get_user", "GET", "/api/users/{id}") + out = match_route(g, "/api/users/42", "GET") + assert out is not None + assert out[0] == "api.get_user" + assert out[1] == 0.9 + + +# ---- Test 3: template literal placeholder ${id} ---- +def test_template_literal_placeholder_matches() -> None: + g = _graph_with_route("api.get_user", "GET", "/api/users/{id}") + out = match_route(g, "/api/users/${id}", "GET") + assert out is not None + assert out[0] == "api.get_user" + assert out[1] == 0.9 + + +# ---- Test 4: Express-style :id placeholder ---- +def test_express_style_colon_placeholder_matches() -> None: + g = _graph_with_route("api.get_user", "GET", "/api/users/{id}") + out = match_route(g, "/api/users/:id", "GET") + assert out is not None + assert out[0] == "api.get_user" + + +# ---- Test 5: method mismatch ---- +def test_method_mismatch_returns_none() -> None: + g = _graph_with_route("api.create_user", "POST", "/api/users") + out = match_route(g, "/api/users", "GET") + assert out is None + + +# ---- Test 6: same path two methods, only matching method picked ---- +def test_two_methods_same_path_picks_matching() -> None: + g = _graph_with_route("api.get_users", "GET", "/api/users") + _add_route(g, "api.create_user", "POST", "/api/users") + out = match_route(g, "/api/users", "POST") + assert out is not None + assert out[0] == "api.create_user" + + +# ---- Test 7: more specific path wins tie ---- +def test_more_specific_path_wins_tie() -> None: + g = _graph_with_route("api.get_me", "GET", "/users/me") + _add_route(g, "api.get_user_by_id", "GET", "/users/{id}") + # Fetching /users/me — both routes' normalised forms could match + # the literal "me" segment (it's not numeric/{}/$/:). So /users/me + # exact-matches the literal route; /users/{id} also normalises but + # at lower priority due to specificity. + out = match_route(g, "/users/me", "GET") + assert out is not None + assert out[0] == "api.get_me" + + +# ---- Test 8: no routes in graph ---- +def test_empty_graph_returns_none() -> None: + g: nx.MultiDiGraph = nx.MultiDiGraph() + out = match_route(g, "/api/users", "GET") + assert out is None + + +# ---- Test 9: body-key overlap bumps confidence (placeholder path) ---- +def test_body_keys_overlap_bumps_confidence() -> None: + # Use a placeholder route so the base score is 0.9 and can grow. + g = _graph_with_route( + "api.update_user", + "PUT", + "/api/users/{id}", + handler_params=[ + {"name": "id", "type": "int", "default": None}, + {"name": "email", "type": "str", "default": None}, + ], + ) + out_no_keys = match_route(g, "/api/users/42", "PUT") + assert out_no_keys is not None + base = out_no_keys[1] + + out_with_keys = match_route( + g, "/api/users/42", "PUT", body_keys=["email"] + ) + assert out_with_keys is not None + assert out_with_keys[1] > base + assert out_with_keys[1] <= 1.0 + + +# ---- Test 10: body-keys without overlap don't bump ---- +def test_body_keys_no_overlap_no_bump() -> None: + g = _graph_with_route( + "api.create_user", + "POST", + "/api/users", + handler_params=[ + {"name": "email", "type": "str", "default": None}, + ], + ) + out_no_keys = match_route(g, "/api/users", "POST") + out_with_unrelated = match_route( + g, "/api/users", "POST", body_keys=["foo", "bar"] + ) + assert out_no_keys is not None + assert out_with_unrelated is not None + assert out_no_keys[1] == out_with_unrelated[1] + + +# ---- Test 11: multiple placeholders in path ---- +def test_multiple_placeholders_match() -> None: + g = _graph_with_route( + "api.get_user_post", "GET", "/users/{uid}/posts/{pid}" + ) + out = match_route(g, "/users/42/posts/7", "GET") + assert out is not None + assert out[0] == "api.get_user_post" + assert out[1] == 0.9 + + +# ---- Test 12: trailing slash differences match ---- +def test_trailing_slash_normalised() -> None: + g = _graph_with_route("api.get_users", "GET", "/api/users") + out_with_slash = match_route(g, "/api/users/", "GET") + out_without = match_route(g, "/api/users", "GET") + assert out_with_slash is not None + assert out_without is not None + assert out_with_slash[0] == out_without[0] + + +# ---- Test 13: case-insensitive method ---- +def test_case_insensitive_method() -> None: + g = _graph_with_route("api.get_users", "GET", "/api/users") + out = match_route(g, "/api/users", "get") + assert out is not None + assert out[0] == "api.get_users" + + +# ---- Test 14: query string is stripped ---- +def test_query_string_stripped() -> None: + g = _graph_with_route("api.search", "GET", "/api/search") + out = match_route(g, "/api/search?q=hello&limit=10", "GET") + assert out is not None + assert out[0] == "api.search" + + +# ---- Test 15: fragment is stripped ---- +def test_fragment_stripped() -> None: + g = _graph_with_route("api.get_doc", "GET", "/api/docs") + out = match_route(g, "/api/docs#section-1", "GET") + assert out is not None + assert out[0] == "api.get_doc" + + +# ---- Test 16: prefix-only fuzzy match returns 0.5 ---- +def test_prefix_only_match_returns_0_5() -> None: + g = _graph_with_route("api.users_root", "GET", "/api/users") + out = match_route(g, "/api/users/42/posts", "GET") + assert out is not None + assert out[0] == "api.users_root" + assert out[1] == 0.5 + + +# ---- Test 17: completely different path returns None ---- +def test_unrelated_path_returns_none() -> None: + g = _graph_with_route("api.users", "GET", "/api/users") + out = match_route(g, "/api/orders", "GET") + assert out is None