From 7705ad1021c492286726a37a896a7a7fac43ef2a Mon Sep 17 00:00:00 2001 From: Wolfvin Date: Sun, 12 Jul 2026 02:57:18 +0000 Subject: [PATCH] fix(callgraph): collect module-scope calls + object-literal arrow fns (closes #210) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reference_count (rc) was severely undercounting cross-file calls due to two patterns silently dropped by the JS/TS backend parsers: 1. Module-scope calls — router.post(path, requirePermission('admin'), h) at the top level of a route file. The requirePermission('admin') call_expression was not inside any function declaration, so no edge was created. This caused middleware-factory functions to get rc=0 even when used in dozens of route files. 2. Arrow functions in object literals — const svc = { list: (ctx) => { ... } }. The variable_declarator value is an 'object' node (not arrow_function), so _parse_variable_declarator returned None and the arrow body calls were lost. This caused service-map modules to be invisible to the call graph. Root cause was in the PARSERS (js_backend_parser.py, ts_backend_parser.py), NOT in callgraph_engine.py as the issue body suspected. callgraph_engine.py is only used by the dataflow command; the rc field in search --mode symbol comes from parser -> edge_resolver.resolve_edges() -> registry. Fix: - Add _find_module_scope_calls pass that walks the AST and collects call_expression / new_expression nodes not inside any function body. Attributed to a synthetic node (exported=True, node_type=module) so dead-code does not flag it. - Add _parse_object_literal_pair / _collect_object_literal_pairs that register arrow functions in object-literal pairs as . nodes, so calls inside their bodies are properly attributed. - For TS parser: separate _find_object_literal_methods pass (the main walk skips export_statement children to avoid double-counting). - For JS parser: separate _collect_object_literal_pairs pass (same reason). Tests: tests/test_issue210_refcount_undercount.py — 9 tests covering both patterns for both TS and JS parsers, plus regression checks that regular function-declaration call attribution is preserved. --- scripts/parsers/js_backend_parser.py | 274 +++++++++++++ scripts/parsers/ts_backend_parser.py | 242 +++++++++++- tests/test_issue210_refcount_undercount.py | 422 +++++++++++++++++++++ 3 files changed, 936 insertions(+), 2 deletions(-) create mode 100644 tests/test_issue210_refcount_undercount.py diff --git a/scripts/parsers/js_backend_parser.py b/scripts/parsers/js_backend_parser.py index 2eb6820d..a3ff09af 100755 --- a/scripts/parsers/js_backend_parser.py +++ b/scripts/parsers/js_backend_parser.py @@ -195,6 +195,16 @@ def extract_references(self, content: str, file_path: str) -> Dict[str, List[Dic node, source, file_path, nodes, edges, ) + elif node.type == 'pair': + # Issue #210: Object-literal property whose value is an + # arrow_function or function_expression, e.g.: + # const service = { list: (ctx) => { ... } } + # Register as . so calls inside the body + # are attributed correctly instead of being lost. + self._parse_and_collect_object_pair( + node, source, file_path, nodes, edges, + ) + # Push children in reverse so they pop in source order if depth + 1 <= MAX_DEPTH: children = node.children @@ -202,11 +212,93 @@ def extract_references(self, content: str, file_path: str) -> Dict[str, List[Dic keep_alive.append(child) stack.append((child, depth + 1)) + # Issue #210: Collect module-scope calls (calls outside any + # registered function body). We track body byte ranges during + # the walk so we can attribute unattributed calls to a synthetic + # node. This catches patterns like: + # router.post(path, requirePermission('admin'), handler) + # at the top level of a route file, where requirePermission is + # a real call but has no enclosing function declaration. + # + # Also collect object-literal pair arrow functions (e.g. + # ``const svc = { list: (ctx) => {...} }``) in a separate pass + # so they are registered even when nested inside + # export_statement (which the main walk above does not descend + # into, to avoid double-counting function/class declarations). + self._collect_object_literal_pairs( + root, source, file_path, nodes, edges, keep_alive, + ) + + module_calls = self._find_module_scope_calls( + root, source, nodes, keep_alive, + ) + if module_calls: + module_node_id = f"{file_path}:0" + module_node = { + "id": module_node_id, + "fn": "", + "file": file_path, + "line": 0, + "async": False, + "exported": True, # module scope is never dead code + "node_type": "module", + } + nodes.append(module_node) + for call_info in module_calls: + edge = { + "from": module_node_id, + "to_fn": call_info["fn_name"], + "via_self": call_info.get("via_self", False) + } + if call_info.get("is_ipc_call"): + edge["is_ipc_call"] = True + edges.append(edge) + return {"nodes": nodes, "edges": edges, "skipped": []} finally: if _gc_was_enabled: _gc.enable() + def _collect_object_literal_pairs( + self, + root: Node, + source: bytes, + file_path: str, + nodes: List[Dict], + edges: List[Dict], + keep_alive: List[Node], + ) -> None: + """Walk the entire AST and register every ``pair`` node whose value + is an arrow_function or function_expression as a function node. + + Issue #210: The main walk in :meth:`extract_references` skips + children of ``export_statement`` (to avoid double-counting function + and class declarations). As a side effect, ``pair`` nodes inside + exported object literals (e.g. + ``export const svc = { list: (ctx) => {...} }``) were never + visited. This separate pass ensures they are registered. + """ + MAX_DEPTH = 200 + local_keep_alive: List[Node] = [root] + stack: List[Tuple[Node, int]] = [(root, 0)] + while stack: + node, depth = stack.pop() + if depth > MAX_DEPTH: + continue + + if node.type == 'pair': + self._parse_and_collect_object_pair( + node, source, file_path, nodes, edges, + ) + + if depth + 1 <= MAX_DEPTH: + children = node.children + for child in reversed(children): + local_keep_alive.append(child) + stack.append((child, depth + 1)) + + keep_alive.extend(local_keep_alive) + def _parse_and_collect_calls( self, decl_node: Node, @@ -261,6 +353,188 @@ def _parse_and_collect_calls( decl_info.pop("body_node", None) return decl_info + def _parse_and_collect_object_pair( + self, + pair_node: Node, + source: bytes, + file_path: str, + nodes: List[Dict], + edges: List[Dict], + ) -> None: + """Parse a ``pair`` node whose value is an arrow_function or + function_expression and collect call edges from its body. + + Issue #210: Arrow functions assigned to object-literal properties + (e.g. ``const service = { list: (ctx) => { ... } }``) were + invisible to the call graph because the enclosing + ``variable_declarator`` value is an ``object`` node, not an + ``arrow_function``, so ``_parse_variable_declarator`` returned + None and the body calls were lost. + + We register each such arrow function as a node named + ``.`` so calls inside its body are properly + attributed. The enclosing var name is found by walking up to the + nearest ``variable_declarator`` ancestor. + """ + key_node = None + value_node = None + for child in pair_node.children: + if child.type in ('property_identifier', 'identifier'): + key_node = child + elif child.type in ('arrow_function', 'function_expression'): + value_node = child + + if key_node is None or value_node is None: + return + + # Find the enclosing variable_declarator to use as the name prefix. + # Stop at function boundaries — we don't want to attribute a pair + # inside a function body to an outer-scope variable. + prefix = "" + parent = pair_node.parent + while parent is not None: + if parent.type == 'variable_declarator': + for pc in parent.children: + if pc.type == 'identifier': + prefix = self.get_text(pc, source) + break + break + if parent.type in ('function_declaration', 'generator_function_declaration', + 'arrow_function', 'function_expression', + 'method_definition'): + break + parent = parent.parent + + key_name = self.get_text(key_node, source) + fn_name = f"{prefix}.{key_name}" if prefix else key_name + + # Check async + is_async = False + for vc in value_node.children: + if vc.type == 'async': + is_async = True + break + + line = self.get_line(pair_node) + node_id = f"{file_path}:{line}" + + # Find the body + body_node = None + for child in value_node.children: + if child.type in ('statement_block', 'expression'): + body_node = child + break + + node_dict = { + "id": node_id, + "fn": fn_name, + "file": file_path, + "line": line, + "async": is_async, + "node_type": "object_method", + } + nodes.append(node_dict) + + if body_node is not None: + fn_calls = self._find_calls_in_scope(body_node, source, file_path) + for call_info in fn_calls: + edge = { + "from": node_id, + "to_fn": call_info["fn_name"], + "via_self": call_info.get("via_self", False) + } + if call_info.get("is_ipc_call"): + edge["is_ipc_call"] = True + edges.append(edge) + + def _find_module_scope_calls( + self, + root: Node, + source: bytes, + registered_nodes: List[Dict], + keep_alive: List[Node], + ) -> List[Dict]: + """Find all call_expression / new_expression nodes that are NOT + inside any registered function body. + + Issue #210: Calls at module scope (e.g. + ``router.post(path, requirePermission('admin'), handler)`` at the + top level of a route file) were silently dropped because no + enclosing function declaration existed to collect them. + + We re-walk the AST and check each call's byte range against the + registered function body ranges (derived from the nodes list). + Calls outside all ranges are returned for attribution to the + synthetic ```` node. + + ``keep_alive`` is extended with any new Node references we hold + (none in this implementation, but the parameter is kept for + symmetry with the caller's keep_alive list). + """ + # Build body byte ranges from registered nodes. We don't have the + # body_node anymore (it was popped in _parse_and_collect_calls), so + # we approximate by using the node's line to find the body. Actually, + # we can re-parse and find the body by walking. But that's expensive. + # + # Simpler approach: walk the AST, and for each call_expression, walk + # up parents to check if any ancestor is a function_declaration, + # generator_function_declaration, arrow_function, function_expression, + # or method_definition. If so, the call is inside a function body and + # already collected (or will be, if the function is anonymous and + # unregistered — but those are best-effort skipped per the existing + # "anonymous callbacks IGNORED" doc). If not, it's module-scope. + FN_BODY_TYPES = { + 'function_declaration', 'generator_function_declaration', + 'arrow_function', 'function_expression', + 'method_definition', + } + + calls: List[Dict] = [] + MAX_DEPTH = 200 + + local_keep_alive: List[Node] = [root] + stack: List[Tuple[Node, int]] = [(root, 0)] + while stack: + node, depth = stack.pop() + if depth > MAX_DEPTH: + continue + + if node.type == 'call_expression': + # Walk up parents to check if inside a function body + inside_fn = False + parent = node.parent + while parent is not None: + if parent.type in FN_BODY_TYPES: + inside_fn = True + break + parent = parent.parent + if not inside_fn: + call_info = self._parse_call(node, source) + if call_info: + calls.append(call_info) + elif node.type == 'new_expression': + inside_fn = False + parent = node.parent + while parent is not None: + if parent.type in FN_BODY_TYPES: + inside_fn = True + break + parent = parent.parent + if not inside_fn: + call_info = self._parse_new_expression(node, source) + if call_info: + calls.append(call_info) + + if depth + 1 <= MAX_DEPTH: + children = node.children + for child in reversed(children): + local_keep_alive.append(child) + stack.append((child, depth + 1)) + + # Keep alive until end of method + keep_alive.extend(local_keep_alive) + return calls + def _find_function_declarations(self, root: Node, source: bytes, file_path: str) -> List[Dict]: """Find all function declarations in the AST.""" diff --git a/scripts/parsers/ts_backend_parser.py b/scripts/parsers/ts_backend_parser.py index 34bbbd9e..c281e71c 100644 --- a/scripts/parsers/ts_backend_parser.py +++ b/scripts/parsers/ts_backend_parser.py @@ -52,6 +52,27 @@ def extract_references(self, content: str, file_path: str) -> Dict[str, List[Dic Returns: {"nodes": [...], "edges": [...]} + + Issue #210: Previously only calls inside named function bodies were + collected. Two common patterns were silently dropped: + + 1. Module-scope calls — e.g. ``router.post(path, requirePermission('admin'), handler)`` + at the top level of a route file. The ``requirePermission('admin')`` + call_expression is not inside any function declaration, so it was + never turned into an edge. This caused middleware-factory functions + like ``requirePermission`` to get ``ref_count == 0`` even when used + in dozens of route files. + + 2. Arrow functions in object literals — e.g. + ``const service = { list: (ctx) => { hasPermission(ctx.user, 'read'); ... } }``. + The ``variable_declarator`` value is an ``object`` node (not an + ``arrow_function``), so ``_parse_variable_declarator`` returned + None and the arrow function's body calls were lost. This caused + service-map-style modules to be invisible to the call graph. + + Both patterns are now handled: module-scope calls are attributed to + a synthetic ```` node per file, and object-literal arrow + functions are registered as ``.`` nodes. """ source = content.encode('utf-8') tree = self.parse(source) @@ -59,13 +80,34 @@ def extract_references(self, content: str, file_path: str) -> Dict[str, List[Dic nodes = [] edges = [] - # First pass: find all function declarations + # First pass: find all function declarations (named functions, + # classes, and variable-bound arrow/function expressions). fn_declarations = self._find_function_declarations(tree, source, file_path) + # Issue #210: Also find arrow functions assigned to object-literal + # properties (e.g. ``const svc = { list: (ctx) => {...} }``) and + # register them as . nodes. These were previously + # invisible to the call graph. + fn_declarations.extend( + self._find_object_literal_methods(tree, source, file_path) + ) + # Build a map of line -> function for scope resolution fn_scope_map = self._build_scope_map(fn_declarations) - # Second pass: find all function calls within each scope + # Track body byte ranges of every registered function so the + # module-scope pass can skip calls that are already attributed. + fn_body_ranges: List[Tuple[int, int]] = [] + for decl in fn_declarations: + body = decl.get("body_node") + if body is not None: + fn_body_ranges.append((body.start_byte, body.end_byte)) + # Also mark the enclosing decl range so calls in the + # parameter list / heritage clause are not double-counted. + # We use the body range only — calls in defaults/params are + # rare and attributing them to the function is acceptable. + + # Second pass: find all function calls within each function's body for decl in fn_declarations: nodes.append(decl["node"]) # Find calls within this function's body @@ -80,6 +122,33 @@ def extract_references(self, content: str, file_path: str) -> Dict[str, List[Dic edge["is_ipc_call"] = True edges.append(edge) + # Issue #210: Third pass — collect module-scope calls (calls that + # occur outside any registered function body). These are attributed + # to a synthetic node so that the call graph records them + # as edges, even though there is no enclosing function declaration. + module_calls = self._find_module_scope_calls(tree, source, fn_body_ranges) + if module_calls: + module_node_id = f"{file_path}:0" + module_node = { + "id": module_node_id, + "fn": "", + "file": file_path, + "line": 0, + "async": False, + "exported": True, # module scope is never dead code + "node_type": "module", # marker for downstream filters + } + nodes.append(module_node) + for call_info in module_calls: + edge = { + "from": module_node_id, + "to_fn": call_info["fn_name"], + "via_self": call_info.get("via_self", False) + } + if call_info.get("is_ipc_call"): + edge["is_ipc_call"] = True + edges.append(edge) + return {"nodes": nodes, "edges": edges} def _find_function_declarations(self, root: Node, source: bytes, @@ -139,6 +208,46 @@ def visit(node: Node, _, depth): decl = self._parse_variable_declarator(node, source, file_path) if decl: declarations.append(decl) + # Note: object-literal pair arrow functions are collected in a + # separate pass (_find_object_literal_methods) so they work + # uniformly for both exported and non-exported variables. + # Walking them here would require descending into export_statement + # children, which the export branch above explicitly avoids to + # prevent double-counting of function/class declarations. + + self.walk_tree(root, source, visit) + return declarations + + def _find_object_literal_methods(self, root: Node, source: bytes, + file_path: str) -> List[Dict]: + """Find all ``pair`` nodes whose value is an arrow_function or + function_expression, and register them as function declarations. + + Issue #210: Arrow functions assigned to object-literal properties + (e.g. ``const service = { list: (ctx) => { ... } }``) were + invisible to the call graph because: + + 1. The enclosing ``variable_declarator`` value is an ``object`` node + (not an ``arrow_function``), so ``_parse_variable_declarator`` + returned None. + 2. The main walk in ``_find_function_declarations`` does not descend + into ``export_statement`` children (to avoid double-counting), so + pairs inside exported object literals were never visited. + + This separate pass walks the entire AST and registers every pair + with an arrow/function value as a node named + ``.`` (or just ```` if the pair is not + inside a variable_declarator, e.g. an object literal passed as a + function argument). + """ + declarations: List[Dict] = [] + + def visit(node: Node, _, depth): + if node.type == 'pair': + decl = self._parse_object_literal_pair(node, source, file_path) + if decl: + declarations.append(decl) + return True self.walk_tree(root, source, visit) return declarations @@ -323,6 +432,135 @@ def _build_scope_map(self, declarations: List[Dict]) -> Dict: sorted_decls = sorted(declarations, key=lambda d: d["scope_start"]) return {i: d for i, d in enumerate(sorted_decls)} + def _parse_object_literal_pair(self, node: Node, source: bytes, + file_path: str) -> Optional[Dict]: + """Parse a ``pair`` node whose value is an arrow_function or + function_expression, e.g. ``list: (ctx) => { ... }``. + + Issue #210: Arrow functions assigned to object-literal properties + were previously invisible to the call graph because their enclosing + ``variable_declarator`` value is an ``object`` node (not an + ``arrow_function``), so ``_parse_variable_declarator`` returned None. + + We register each such arrow function as a node named + ``.`` so calls inside its body are properly + attributed. The enclosing var name is found by walking up to the + nearest ``variable_declarator`` ancestor and reading its identifier. + """ + key_node = None + value_node = None + for child in node.children: + if child.type in ('property_identifier', 'identifier'): + key_node = child + elif child.type in ('arrow_function', 'function_expression'): + value_node = child + + if key_node is None or value_node is None: + return None + + # Find the enclosing variable_declarator to use as the name prefix. + # If there is none (e.g., the pair is inside a function-argument + # object literal), fall back to just the key name. + prefix = "" + parent = node.parent + while parent is not None: + if parent.type == 'variable_declarator': + for pc in parent.children: + if pc.type == 'identifier': + prefix = self.get_text(pc, source) + break + break + # Stop at function boundaries — we don't want to attribute a + # pair inside a function body to an outer-scope variable. + if parent.type in ('function_declaration', 'generator_function_declaration', + 'arrow_function', 'function_expression', + 'method_definition'): + break + parent = parent.parent + + key_name = self.get_text(key_node, source) + if prefix: + fn_name = f"{prefix}.{key_name}" + else: + fn_name = key_name + + # Check async + is_async = False + for vc in value_node.children: + if vc.type == 'async': + is_async = True + break + + line = self.get_line(node) + node_id = f"{file_path}:{line}" + + # Find the body + body_node = None + for child in value_node.children: + if child.type in ('statement_block', 'expression'): + body_node = child + break + + return { + "node": { + "id": node_id, + "fn": fn_name, + "file": file_path, + "line": line, + "async": is_async, + "node_type": "object_method", + }, + "body_node": body_node, + "scope_start": node.start_point.row, + "scope_end": node.end_point.row, + } + + def _find_module_scope_calls(self, root: Node, source: bytes, + fn_body_ranges: List[Tuple[int, int]] + ) -> List[Dict]: + """Find all call_expression / new_expression nodes that are NOT + inside any registered function body. + + Issue #210: Calls at module scope (e.g. + ``router.post(path, requirePermission('admin'), handler)`` at the + top level of a route file) were silently dropped because no + enclosing function declaration existed to collect them. + + We walk the entire AST and check each call's byte range against + the registered function body ranges. Calls outside all ranges + are returned for attribution to the synthetic ```` node. + """ + if not fn_body_ranges: + # Fast path: no registered functions, every call is module-scope. + # Sort nothing — just walk. + pass + + calls: List[Dict] = [] + + def _is_inside_fn_body(node: Node) -> bool: + start = node.start_byte + end = node.end_byte + for body_start, body_end in fn_body_ranges: + if start >= body_start and end <= body_end: + return True + return False + + def visit(node: Node, _, depth): + if node.type == 'call_expression': + if not _is_inside_fn_body(node): + call_info = self._parse_call(node, source) + if call_info: + calls.append(call_info) + elif node.type == 'new_expression': + if not _is_inside_fn_body(node): + call_info = self._parse_new_expression(node, source) + if call_info: + calls.append(call_info) + return True + + self.walk_tree(root, source, visit) + return calls + def _find_calls_in_scope(self, body_node: Optional[Node], source: bytes, file_path: str) -> List[Dict]: """Find all function calls within a function body.""" diff --git a/tests/test_issue210_refcount_undercount.py b/tests/test_issue210_refcount_undercount.py new file mode 100644 index 00000000..351767c5 --- /dev/null +++ b/tests/test_issue210_refcount_undercount.py @@ -0,0 +1,422 @@ +""" +Regression tests for issue #210 — reference_count severely undercounts +cross-file calls. + +Two patterns that were silently dropped before the fix: + +1. **Middleware-factory-argument pattern** — a function call passed as + an argument to a top-level route registration, e.g. + ``router.post(path, requirePermission('admin'), handler)``. + The ``requirePermission('admin')`` call_expression is not inside any + function declaration, so it was never turned into a call-graph edge. + +2. **Multi-site same-file calls inside object-literal arrow functions** + — arrow functions assigned to object-literal properties, e.g. + ``const service = { list: (ctx) => { hasPermission(ctx.user, 'read'); } }``. + The ``variable_declarator`` value is an ``object`` node (not an + ``arrow_function``), so ``_parse_variable_declarator`` returned None + and the arrow function's body calls were lost. + +Both patterns are now handled. These tests verify the fix by feeding +minimal repro fixtures through the TS/JS backend parsers and the +edge_resolver, then asserting the resulting ``ref_count`` matches the +ground-truth call-site count. +""" + +import os +import sys +import pytest + +SCRIPT_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts") +sys.path.insert(0, SCRIPT_DIR) + +# ─── Parser availability guards ──────────────────────────────────── +_ts_be_parser = None +_ts_be_parser_available = False +try: + from parsers.ts_backend_parser import TSBackendParser + _ts_be_parser = TSBackendParser() + _ts_be_parser_available = True +except Exception: + pass + +_js_be_parser = None +_js_be_parser_available = False +try: + from parsers.js_backend_parser import JSBackendParser + _js_be_parser = JSBackendParser() + _js_be_parser_available = True +except Exception: + pass + +from edge_resolver import resolve_edges + + +# ─── Fixtures reproducing the two reported patterns ─────────────── + +# Case 1 — middleware-factory argument at module scope. +# requirePermission is defined in one file, called 3× at module scope +# across three route files. +TS_PERMISSION_GATE = """\ +export function requirePermission(permission: string) { + return (req: any, res: any, next: any) => { + if (!req.user) { res.sendStatus(403); return; } + next(); + }; +} +""" + +TS_ROUTE_TEMPLATE = """\ +import {{ Router }} from 'express'; +import {{ requirePermission }} from '../middleware/permission-gate'; + +const router = Router(); + +router.post('{path}', requirePermission('admin'), (req, res) => {{ + res.json({{ ok: true }}); +}}); + +export default router; +""" + +# Case 2 — object-literal arrow function pattern. +# hasPermission is defined in one file, called from arrow functions +# assigned to object-literal properties in two service files (3 sites each) +# plus 1 site in a regular function. +TS_PERMISSIONS = """\ +export function hasPermission(user: any, perm: string): boolean { + return user?.permissions?.includes(perm) ?? false; +} +""" + +TS_ASSIGNMENTS_SERVICE = """\ +import { hasPermission } from '../lib/permissions'; + +export const assignmentService = { + list: (ctx: any) => { + if (!hasPermission(ctx.user, 'read')) return null; + return []; + }, + create: (ctx: any) => { + if (!hasPermission(ctx.user, 'write')) throw new Error('no'); + return {}; + }, + remove: (ctx: any) => { + if (!hasPermission(ctx.user, 'delete')) throw new Error('no'); + return true; + }, +}; +""" + +TS_TASK_TEMPLATES_SERVICE = """\ +import { hasPermission } from '../lib/permissions'; + +export const taskTemplateService = { + list: (ctx: any) => { + if (!hasPermission(ctx.user, 'read')) return null; + return []; + }, + create: (ctx: any) => { + if (!hasPermission(ctx.user, 'write')) throw new Error('no'); + return {}; + }, + remove: (ctx: any) => { + if (!hasPermission(ctx.user, 'delete')) throw new Error('no'); + return true; + }, +}; +""" + +TS_PERMISSION_GATE_USE = """\ +import { hasPermission } from './permissions'; + +export function checkGate(ctx: any) { + return hasPermission(ctx.user, 'admin'); +} +""" + + +# JS equivalents (for the JS parser) +JS_PERMISSION_GATE = """\ +export function requirePermission(permission) { + return (req, res, next) => { + if (!req.user) { res.sendStatus(403); return; } + next(); + }; +} +""" + +JS_ROUTE_TEMPLATE = """\ +const {{ Router }} = require('express'); +const {{ requirePermission }} = require('../middleware/permission-gate'); + +const router = Router(); + +router.post('{path}', requirePermission('admin'), (req, res) => {{ + res.json({{ ok: true }}); +}}); + +module.exports = router; +""" + +JS_PERMISSIONS = """\ +export function hasPermission(user, perm) { + return user && user.permissions && user.permissions.includes(perm); +} +""" + +JS_ASSIGNMENTS_SERVICE = """\ +import { hasPermission } from './permissions'; + +export const assignmentService = { + list: (ctx) => { + if (!hasPermission(ctx.user, 'read')) return null; + return []; + }, + create: (ctx) => { + if (!hasPermission(ctx.user, 'write')) throw new Error('no'); + return {}; + }, + remove: (ctx) => { + if (!hasPermission(ctx.user, 'delete')) throw new Error('no'); + return true; + }, +}; +""" + + +def _ref_count_for(nodes: list, edges: list, target_fn: str) -> int: + """Resolve edges and return the ref_count for the first node whose + fn matches target_fn.""" + resolved_nodes, resolved_edges = resolve_edges(nodes, edges) + for n in resolved_nodes: + if n["fn"] == target_fn: + return n.get("ref_count", 0) + return -1 # not found + + +# ─── TS parser tests ────────────────────────────────────────────── + +class TestIssue210TS: + """Regression tests for issue #210 — TS backend parser.""" + + @pytest.mark.skipif(not _ts_be_parser_available, + reason="Tree-sitter TypeScript grammar not installed") + def test_module_scope_middleware_factory_call_is_collected(self): + """Case 1: requirePermission('admin') passed as argument to + router.post() at module scope must produce a call-graph edge. + + Before fix: 0 edges (call was silently dropped). + After fix: 3 edges (one per route file). + """ + files = [ + ("src/middleware/permission-gate.ts", TS_PERMISSION_GATE), + ("src/routes/foo.ts", TS_ROUTE_TEMPLATE.format(path="/foo")), + ("src/routes/bar.ts", TS_ROUTE_TEMPLATE.format(path="/bar")), + ("src/routes/baz.ts", TS_ROUTE_TEMPLATE.format(path="/baz")), + ] + all_nodes = [] + all_edges = [] + for fpath, content in files: + result = _ts_be_parser.extract_references(content, fpath) + all_nodes.extend(result.get("nodes", [])) + all_edges.extend(result.get("edges", [])) + + rc = _ref_count_for(all_nodes, all_edges, "requirePermission") + assert rc == 3, ( + f"requirePermission should have ref_count=3 (3 route files call " + f"it at module scope), got rc={rc}. Edges to requirePermission: " + f"{[e for e in all_edges if e.get('to_fn') == 'requirePermission']}" + ) + + @pytest.mark.skipif(not _ts_be_parser_available, + reason="Tree-sitter TypeScript grammar not installed") + def test_object_literal_arrow_function_calls_are_collected(self): + """Case 2: hasPermission() called from arrow functions assigned + to object-literal properties must produce call-graph edges. + + Before fix: 1 edge (only the call inside checkGate was detected). + After fix: 7 edges (3 from assignmentService.{list,create,remove} + + 3 from taskTemplateService.{list,create,remove} + + 1 from checkGate). + """ + files = [ + ("src/lib/permissions.ts", TS_PERMISSIONS), + ("src/services/assignments.ts", TS_ASSIGNMENTS_SERVICE), + ("src/services/task-templates.ts", TS_TASK_TEMPLATES_SERVICE), + ("src/middleware/permission-gate-use.ts", TS_PERMISSION_GATE_USE), + ] + all_nodes = [] + all_edges = [] + for fpath, content in files: + result = _ts_be_parser.extract_references(content, fpath) + all_nodes.extend(result.get("nodes", [])) + all_edges.extend(result.get("edges", [])) + + rc = _ref_count_for(all_nodes, all_edges, "hasPermission") + assert rc == 7, ( + f"hasPermission should have ref_count=7 (3+3+1 call sites), " + f"got rc={rc}. Edges to hasPermission: " + f"{[e for e in all_edges if e.get('to_fn') == 'hasPermission']}" + ) + + @pytest.mark.skipif(not _ts_be_parser_available, + reason="Tree-sitter TypeScript grammar not installed") + def test_object_literal_arrow_functions_registered_as_named_nodes(self): + """Arrow functions in object literals should be registered as + nodes named ``.`` so callers can trace + into them via the call graph. + """ + result = _ts_be_parser.extract_references( + TS_ASSIGNMENTS_SERVICE, "src/services/assignments.ts", + ) + fn_names = [n["fn"] for n in result["nodes"]] + assert "assignmentService.list" in fn_names + assert "assignmentService.create" in fn_names + assert "assignmentService.remove" in fn_names + + @pytest.mark.skipif(not _ts_be_parser_available, + reason="Tree-sitter TypeScript grammar not installed") + def test_module_node_only_created_when_module_scope_calls_exist(self): + """The synthetic ```` node should only be added when the + file actually has module-scope calls — files with only function + declarations should not get a node. + """ + # File with only a function declaration — no module-scope calls + ts = "export function foo() { return bar(); }" + result = _ts_be_parser.extract_references(ts, "test.ts") + fn_names = [n["fn"] for n in result["nodes"]] + assert "" not in fn_names, ( + "File with no module-scope calls should not get a node" + ) + + # File with module-scope call + ts2 = "const router = Router();" + result2 = _ts_be_parser.extract_references(ts2, "test.ts") + fn_names2 = [n["fn"] for n in result2["nodes"]] + assert "" in fn_names2, ( + "File with module-scope calls should get a node" + ) + + +# ─── JS parser tests ────────────────────────────────────────────── + +class TestIssue210JS: + """Regression tests for issue #210 — JS backend parser.""" + + @pytest.mark.skipif(not _js_be_parser_available, + reason="Tree-sitter JavaScript grammar not installed") + def test_module_scope_middleware_factory_call_is_collected(self): + """Case 1 (JS): same as TS test but using JS parser.""" + files = [ + ("src/middleware/permission-gate.js", JS_PERMISSION_GATE), + ("src/routes/foo.js", JS_ROUTE_TEMPLATE.format(path="/foo")), + ("src/routes/bar.js", JS_ROUTE_TEMPLATE.format(path="/bar")), + ("src/routes/baz.js", JS_ROUTE_TEMPLATE.format(path="/baz")), + ] + all_nodes = [] + all_edges = [] + for fpath, content in files: + result = _js_be_parser.extract_references(content, fpath) + all_nodes.extend(result.get("nodes", [])) + all_edges.extend(result.get("edges", [])) + + rc = _ref_count_for(all_nodes, all_edges, "requirePermission") + assert rc == 3, ( + f"requirePermission should have ref_count=3, got rc={rc}" + ) + + @pytest.mark.skipif(not _js_be_parser_available, + reason="Tree-sitter JavaScript grammar not installed") + def test_object_literal_arrow_function_calls_are_collected(self): + """Case 2 (JS): hasPermission() called from object-literal arrow + functions must produce edges. + """ + files = [ + ("src/lib/permissions.js", JS_PERMISSIONS), + ("src/services/assignments.js", JS_ASSIGNMENTS_SERVICE), + ] + all_nodes = [] + all_edges = [] + for fpath, content in files: + result = _js_be_parser.extract_references(content, fpath) + all_nodes.extend(result.get("nodes", [])) + all_edges.extend(result.get("edges", [])) + + rc = _ref_count_for(all_nodes, all_edges, "hasPermission") + assert rc == 3, ( + f"hasPermission should have ref_count=3 (3 object-literal " + f"arrow function call sites), got rc={rc}" + ) + + @pytest.mark.skipif(not _js_be_parser_available, + reason="Tree-sitter JavaScript grammar not installed") + def test_object_literal_arrow_functions_registered_as_named_nodes(self): + """Arrow functions in object literals should be registered as + nodes named ``.``. + """ + result = _js_be_parser.extract_references( + JS_ASSIGNMENTS_SERVICE, "src/services/assignments.js", + ) + fn_names = [n["fn"] for n in result["nodes"]] + assert "assignmentService.list" in fn_names + assert "assignmentService.create" in fn_names + assert "assignmentService.remove" in fn_names + + +# ─── Cross-cutting regression check ─────────────────────────────── + +class TestIssue210NoRegression: + """Verify the fix does not regress existing parser behavior.""" + + @pytest.mark.skipif(not _ts_be_parser_available, + reason="Tree-sitter TypeScript grammar not installed") + def test_ts_regular_function_calls_still_collected(self): + """Calls inside regular function declarations must still be + attributed to the function (not to ).""" + ts = """\ +export function getAssignments(ctx: any) { + if (!hasPermission(ctx.user, 'read')) return null; + const a = hasPermission(ctx.user, 'write'); + return { a }; +} +""" + result = _ts_be_parser.extract_references(ts, "test.ts") + nodes = result["nodes"] + # getAssignments should be a registered node + fn_names = [n["fn"] for n in nodes] + assert "getAssignments" in fn_names + # The 2 hasPermission calls should be attributed to getAssignments. + # Look up the node ID for getAssignments, then check edges. + ga_id = next(n["id"] for n in nodes if n["fn"] == "getAssignments") + hp_edges = [e for e in result["edges"] if e.get("to_fn") == "hasPermission"] + assert len(hp_edges) == 2, f"expected 2 hasPermission edges, got {len(hp_edges)}" + for e in hp_edges: + assert e["from"] == ga_id, ( + f"hasPermission call should be attributed to getAssignments " + f"(id={ga_id!r}), got from={e['from']!r}" + ) + # No node should be created (no module-scope calls) + assert "" not in fn_names + + @pytest.mark.skipif(not _js_be_parser_available, + reason="Tree-sitter JavaScript grammar not installed") + def test_js_regular_function_calls_still_collected(self): + """Same as TS test, but for JS parser.""" + js = """\ +function processOrder(order) { + return validateOrder(order); +} +function validateOrder(order) { + return order && order.id; +} +""" + result = _js_be_parser.extract_references(js, "test.js") + fn_names = [n["fn"] for n in result["nodes"]] + assert "processOrder" in fn_names + assert "validateOrder" in fn_names + # processOrder calls validateOrder — should be 1 edge + vo_edges = [e for e in result["edges"] if e.get("to_fn") == "validateOrder"] + assert len(vo_edges) == 1 + # No node + assert "" not in fn_names