From e44e6e986c44abab38d28ab865f95deea242dcf6 Mon Sep 17 00:00:00 2001 From: Danil Tarasov Date: Wed, 20 May 2026 21:50:41 +0300 Subject: [PATCH] feat: add v8 affected and import-resolution support --- graphify/__main__.py | 64 ++ graphify/affected.py | 151 +++ graphify/extract.py | 1279 +++++++++++++++++++++--- graphify/skill-windows.md | 2 +- tests/test_affected_cli.py | 60 ++ tests/test_extract.py | 153 +++ tests/test_js_import_resolution.py | 399 ++++++++ tests/test_multilang.py | 23 +- tests/test_python_import_resolution.py | 53 + 9 files changed, 2012 insertions(+), 172 deletions(-) create mode 100644 graphify/affected.py create mode 100644 tests/test_affected_cli.py create mode 100644 tests/test_js_import_resolution.py create mode 100644 tests/test_python_import_resolution.py diff --git a/graphify/__main__.py b/graphify/__main__.py index 8ca1a293c..4b08c2696 100644 --- a/graphify/__main__.py +++ b/graphify/__main__.py @@ -1270,6 +1270,10 @@ def main() -> None: print(" --context C explicit edge-context filter (repeatable)") print(" --budget N cap output at N tokens (default 2000)") print(" --graph path to graph.json (default graphify-out/graph.json)") + print(" affected \"X\" reverse traversal to find nodes impacted by X") + print(" --relation R edge relation to traverse in reverse (repeatable)") + print(" --depth N reverse traversal depth (default 2)") + print(" --graph path to graph.json (default graphify-out/graph.json)") print(" save-result save a Q&A result to graphify-out/memory/ for graph feedback loop") print(" --question Q the question asked") print(" --answer A the answer to save") @@ -1587,6 +1591,66 @@ def main() -> None: context_filters=context_filters, ) ) + elif cmd == "affected": + if len(sys.argv) < 3: + print("Usage: graphify affected \"\" [--relation R] [--depth N] [--graph path]", file=sys.stderr) + sys.exit(1) + from graphify.affected import DEFAULT_AFFECTED_RELATIONS, format_affected, load_graph + query = sys.argv[2] + graph_path = "graphify-out/graph.json" + depth = 2 + relations: list[str] = [] + args = sys.argv[3:] + i = 0 + while i < len(args): + if args[i] == "--graph" and i + 1 < len(args): + graph_path = args[i + 1] + i += 2 + elif args[i].startswith("--graph="): + graph_path = args[i].split("=", 1)[1] + i += 1 + elif args[i] == "--depth" and i + 1 < len(args): + try: + depth = int(args[i + 1]) + except ValueError: + print("error: --depth must be an integer", file=sys.stderr) + sys.exit(1) + i += 2 + elif args[i].startswith("--depth="): + try: + depth = int(args[i].split("=", 1)[1]) + except ValueError: + print("error: --depth must be an integer", file=sys.stderr) + sys.exit(1) + i += 1 + elif args[i] == "--relation" and i + 1 < len(args): + relations.append(args[i + 1]) + i += 2 + elif args[i].startswith("--relation="): + relations.append(args[i].split("=", 1)[1]) + i += 1 + else: + i += 1 + gp = Path(graph_path).resolve() + if not gp.exists(): + print(f"error: graph file not found: {gp}", file=sys.stderr) + sys.exit(1) + if not gp.suffix == ".json": + print("error: graph file must be a .json file", file=sys.stderr) + sys.exit(1) + try: + graph = load_graph(gp) + except Exception as exc: + print(f"error: could not load graph: {exc}", file=sys.stderr) + sys.exit(1) + print( + format_affected( + graph, + query, + relations=relations or DEFAULT_AFFECTED_RELATIONS, + depth=depth, + ) + ) elif cmd == "save-result": # graphify save-result --question Q --answer A --type T [--nodes N1 N2 ...] import argparse as _ap diff --git a/graphify/affected.py b/graphify/affected.py new file mode 100644 index 000000000..109eaa95e --- /dev/null +++ b/graphify/affected.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +import networkx as nx + + +DEFAULT_AFFECTED_RELATIONS = ( + "calls", + "references", + "imports", + "imports_from", + "re_exports", + "inherits", + "extends", + "implements", + "uses", + "mixes_in", + "embeds", +) + + +@dataclass(frozen=True) +class AffectedHit: + node_id: str + depth: int + via_relation: str + + +def _node_label(graph: nx.Graph, node_id: str) -> str: + data = graph.nodes[node_id] + return str(data.get("label") or node_id) + + +def _format_location(data: dict) -> str: + source_file = data.get("source_file") or "-" + source_location = data.get("source_location") + if source_location: + return f"{source_file}:{source_location}" + return str(source_file) + + +def resolve_seed(graph: nx.Graph, query: str) -> str | None: + if query in graph: + return query + query_lower = query.lower() + exact_label_matches = [ + str(node_id) + for node_id, data in graph.nodes(data=True) + if str(data.get("label", "")).lower() == query_lower + ] + if len(exact_label_matches) == 1: + return exact_label_matches[0] + exact_source_matches = [ + str(node_id) + for node_id, data in graph.nodes(data=True) + if str(data.get("source_file", "")).lower() == query_lower + ] + if len(exact_source_matches) == 1: + return exact_source_matches[0] + contains_matches = [ + str(node_id) + for node_id, data in graph.nodes(data=True) + if query_lower in str(data.get("label", "")).lower() + ] + if len(contains_matches) == 1: + return contains_matches[0] + return None + + +def affected_nodes( + graph: nx.Graph, + seed: str, + *, + relations: Iterable[str] = DEFAULT_AFFECTED_RELATIONS, + depth: int = 2, +) -> list[AffectedHit]: + relation_set = set(relations) + seen = {seed} + queue: deque[tuple[str, int]] = deque([(seed, 0)]) + hits: list[AffectedHit] = [] + + while queue: + current, current_depth = queue.popleft() + if current_depth >= depth: + continue + if hasattr(graph, "in_edges"): + incoming = graph.in_edges(current, data=True) + else: + incoming = ( + (source, target, data) + for source, target, data in graph.edges(data=True) + if target == current + ) + for source, _target, data in incoming: + relation = str(data.get("relation", "")) + if relation not in relation_set: + continue + source = str(source) + if source in seen: + continue + seen.add(source) + hit = AffectedHit(source, current_depth + 1, relation) + hits.append(hit) + queue.append((source, current_depth + 1)) + + return hits + + +def format_affected( + graph: nx.Graph, + query: str, + *, + relations: Iterable[str] = DEFAULT_AFFECTED_RELATIONS, + depth: int = 2, +) -> str: + relation_list = tuple(relations) + seed = resolve_seed(graph, query) + if seed is None: + return f"No unique node match for {query}" + + hits = affected_nodes(graph, seed, relations=relation_list, depth=depth) + lines = [ + f"Affected nodes for {_node_label(graph, seed)}", + f"Relations: {', '.join(relation_list)}", + f"Depth: {depth}", + ] + if not hits: + lines.append("No affected nodes found.") + return "\n".join(lines) + + for hit in hits: + data = graph.nodes[hit.node_id] + lines.append( + f"- {_node_label(graph, hit.node_id)} [{hit.via_relation}] {_format_location(data)}" + ) + return "\n".join(lines) + + +def load_graph(path: Path) -> nx.Graph: + import json + from networkx.readwrite import json_graph + + raw = json.loads(path.read_text(encoding="utf-8")) + try: + return json_graph.node_link_graph(raw, edges="links") + except TypeError: + return json_graph.node_link_graph(raw) diff --git a/graphify/extract.py b/graphify/extract.py index 298de16f0..5128caf3e 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -56,6 +56,43 @@ def _file_stem(path: Path) -> str: _TSCONFIG_ALIAS_CACHE: dict[str, dict[str, str]] = {} +_WORKSPACE_PACKAGE_CACHE: dict[str, dict[str, Path]] = {} +_JS_CACHE_BYPASS_SUFFIXES = {".js", ".jsx", ".mjs", ".ts", ".tsx", ".vue", ".svelte"} +_JS_RESOLVE_EXTS = (".ts", ".tsx", ".svelte", ".js", ".jsx", ".mjs") +_JS_INDEX_FILES = ("index.ts", "index.tsx", "index.svelte", "index.js", "index.jsx", "index.mjs") + + +def _resolve_js_import_path(candidate: Path) -> Path: + """Resolve a JS/TS/Svelte import target to a local file when it exists.""" + candidate = Path(os.path.normpath(candidate)) + if candidate.is_file(): + return candidate + + # TS ESM convention: imports often spell .js/.jsx while source is .ts/.tsx. + if candidate.suffix == ".js": + ts_candidate = candidate.with_suffix(".ts") + if ts_candidate.is_file(): + return ts_candidate + elif candidate.suffix == ".jsx": + tsx_candidate = candidate.with_suffix(".tsx") + if tsx_candidate.is_file(): + return tsx_candidate + + # Append extensions to the full filename, which covers extensionless imports, + # multi-dot helpers, and Svelte 5 rune files like Foo.svelte.ts. + for ext in _JS_RESOLVE_EXTS: + with_ext = candidate.parent / f"{candidate.name}{ext}" + if with_ext.is_file(): + return with_ext + + # Only fall back to directory indexes after file candidates lose. + if candidate.is_dir(): + for index_name in _JS_INDEX_FILES: + index_candidate = candidate / index_name + if index_candidate.is_file(): + return index_candidate + + return candidate def _strip_jsonc(text: str) -> str: @@ -151,6 +188,133 @@ def _load_tsconfig_aliases(start_dir: Path) -> dict[str, str]: return {} +def _find_workspace_root(start_dir: Path) -> Path | None: + current = start_dir.resolve() + for candidate in [current, *current.parents]: + if (candidate / "pnpm-workspace.yaml").exists(): + return candidate + return None + + +def _workspace_globs(workspace_file: Path) -> list[str]: + globs: list[str] = [] + in_packages = False + for raw_line in workspace_file.read_text(encoding="utf-8", errors="replace").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + if line.startswith("packages:"): + in_packages = True + continue + if in_packages and line.startswith("-"): + value = line[1:].strip().strip("'\"") + if value and not value.startswith("!"): + globs.append(value) + continue + if in_packages and not raw_line.startswith((" ", "\t")): + break + return globs + + +def _load_workspace_packages(start_dir: Path) -> dict[str, Path]: + root = _find_workspace_root(start_dir) + if root is None: + return {} + key = str(root) + if key in _WORKSPACE_PACKAGE_CACHE: + return _WORKSPACE_PACKAGE_CACHE[key] + + packages: dict[str, Path] = {} + for pattern in _workspace_globs(root / "pnpm-workspace.yaml"): + for package_dir in root.glob(pattern): + manifest = package_dir / "package.json" + if not manifest.is_file(): + continue + try: + data = json.loads(manifest.read_text(encoding="utf-8")) + except Exception: + continue + name = data.get("name") + if isinstance(name, str) and name: + packages[name] = package_dir + _WORKSPACE_PACKAGE_CACHE[key] = packages + return packages + + +def _package_entry_candidates(package_dir: Path, subpath: str) -> list[Path]: + manifest = package_dir / "package.json" + manifest_data: dict[str, Any] = {} + try: + manifest_data = json.loads(manifest.read_text(encoding="utf-8")) + except Exception: + pass + + if subpath: + return [package_dir / subpath] + + exports = manifest_data.get("exports") + if isinstance(exports, str): + return [package_dir / exports] + if isinstance(exports, dict): + dot_export = exports.get(".") + if isinstance(dot_export, str): + return [package_dir / dot_export] + if isinstance(dot_export, dict): + for key in ("types", "import", "default", "svelte"): + value = dot_export.get(key) + if isinstance(value, str): + return [package_dir / value] + + candidates: list[Path] = [] + for key in ("svelte", "module", "main", "types"): + value = manifest_data.get(key) + if isinstance(value, str): + candidates.append(package_dir / value) + candidates.append(package_dir / "src/index") + candidates.append(package_dir / "index") + return candidates + + +def _resolve_workspace_import(raw: str, start_dir: Path) -> Path | None: + packages = _load_workspace_packages(start_dir) + for package_name, package_dir in packages.items(): + if raw == package_name: + subpath = "" + elif raw.startswith(package_name + "/"): + subpath = raw[len(package_name) + 1:] + else: + continue + for candidate in _package_entry_candidates(package_dir, subpath): + resolved = _resolve_js_import_path(candidate) + if resolved.is_file(): + return resolved + return None + + +def _resolve_js_module_path(raw: str | Path, start_dir: Path | None = None) -> Path | None: + """Resolve a JS/TS module path or specifier to a local source file. + + With a Path argument this preserves the path-based helper API used by + import-extension tests. With a string plus start_dir it resolves JS/TS + module specifiers including relative paths, tsconfig aliases, and workspace + packages. + """ + if isinstance(raw, Path): + return _resolve_js_import_path(raw) + if start_dir is None: + return _resolve_js_import_path(Path(raw)) + if raw.startswith("."): + return _resolve_js_import_path(start_dir / raw) + + aliases = _load_tsconfig_aliases(start_dir) + for alias_prefix, alias_base in aliases.items(): + if raw == alias_prefix or raw.startswith(alias_prefix + "/"): + rest = raw[len(alias_prefix):].lstrip("/") + return _resolve_js_import_path(Path(os.path.normpath(Path(alias_base) / rest))) + + return _resolve_workspace_import(raw, start_dir) + + # ── LanguageConfig dataclass ───────────────────────────────────────────────── @dataclass @@ -198,75 +362,6 @@ class LanguageConfig: # ── Generic helpers ─────────────────────────────────────────────────────────── -# Vite / TypeScript resolver extensions. Used by _resolve_js_module_path() -# to map import specifiers onto real files on disk, so the resulting node -# id matches the one _extract_generic creates for the target file. -_JS_RESOLVE_EXTS = (".ts", ".tsx", ".svelte", ".js", ".jsx", ".mjs") -_JS_INDEX_FILES = ("index.ts", "index.tsx", "index.js", "index.jsx") - - -def _resolve_js_module_path(p: Path) -> Path: - """Resolve a JS/TS-style import specifier path to an actual file on disk. - - TypeScript / SvelteKit / Vite let you write imports without a file - extension and auto-resolve via a fixed extension order. The pre-existing - .js→.ts and .jsx→.tsx rewrites only covered the TS-ESM-via-.js convention; - every other shape produced a phantom node id and the edge was lost in - build_from_json. - - Order, mirroring Vite's resolver: - - 1. exact path, when it's a real file on disk - 2. directory → try index.{ts,tsx,js,jsx} - 3. .js → .ts (TS ESM convention; written as .js, file is .ts) - .jsx → .tsx - 4. append .ts/.tsx/.svelte/.js/.jsx/.mjs to the FULL filename — not - a suffix-swap. This handles, in one rule: - - bare paths: foo → foo.ts - - Svelte 5 rune files: foo.svelte → foo.svelte.ts - - multi-dot helper files: foo.shared → foo.shared.ts - - config files: foo.config → foo.config.ts - - test helper files: foo.spec → foo.spec.ts - 5. directory variant: try .//index.{ts,tsx,js,jsx} - - Falls back to the original path on no match — preserves pre-fix behaviour - for genuinely external modules (the edge gets dropped as external by - build_from_json). - """ - if p.is_file(): - return p - # TS ESM convention: import path written with .js but the real file is .ts. - # Apply BEFORE the generic append loop so we don't accidentally match - # foo.js → foo.js.ts when the real file is foo.ts. - if p.suffix == ".js": - c = p.with_suffix(".ts") - if c.is_file(): - return c - if p.suffix == ".jsx": - c = p.with_suffix(".tsx") - if c.is_file(): - return c - # Try appending extensions to the FULL filename BEFORE checking for a - # directory import. Both TypeScript and Vite resolvers prefer a file - # match over a directory match — projects routinely have a `foo.ts` - # file living alongside a `foo/` directory of sub-modules (e.g. - # `auth.ts` next to `auth/`). If we checked the directory first, those - # file imports would silently lose to a directory with no `index.*`. - for ext in _JS_RESOLVE_EXTS: - c = p.parent / (p.name + ext) - if c.is_file(): - return c - # Directory imports: try .//index.{ts,tsx,js,jsx}. Reached only - # after every file-extension candidate has been ruled out, matching the - # resolver fallback chain. - if p.is_dir(): - for idx in _JS_INDEX_FILES: - c = p / idx - if c.is_file(): - return c - return p - - def _read_text(node, source: bytes) -> str: return source[node.start_byte:node.end_byte].decode("utf-8", errors="replace") @@ -346,22 +441,15 @@ def _import_python(node, source: bytes, file_nid: str, stem: str, edges: list, s def _resolve_js_import_target(raw: str, str_path: str) -> "tuple[str, Path | None] | None": """Resolve a JS/TS import path string to (target_nid, resolved_path). - Handles relative paths, tsconfig path aliases, and bare/scoped imports. + Handles relative paths, tsconfig path aliases, workspace packages, and + bare/scoped imports. Returns None if `raw` is empty. """ if not raw: return None - if raw.startswith("."): - resolved = Path(os.path.normpath(Path(str_path).parent / raw)) - resolved = _resolve_js_module_path(resolved) - return _make_id(str(resolved)), resolved - aliases = _load_tsconfig_aliases(Path(str_path).parent) - for alias_prefix, alias_base in aliases.items(): - if raw == alias_prefix or raw.startswith(alias_prefix + "/"): - rest = raw[len(alias_prefix):].lstrip("/") - resolved_alias = Path(os.path.normpath(Path(alias_base) / rest)) - resolved_alias = _resolve_js_module_path(resolved_alias) - return _make_id(str(resolved_alias)), resolved_alias + resolved_path = _resolve_js_module_path(raw, Path(str_path).parent) + if resolved_path is not None: + return _make_id(str(resolved_path)), resolved_path module_name = raw.split("/")[-1] if not module_name: return None @@ -457,31 +545,11 @@ def _dynamic_import_js(node, source: bytes, caller_nid: str, str_path: str, edge continue if not raw: break - # Resolve path using the same logic as static imports - if raw.startswith("."): - resolved = Path(os.path.normpath(Path(str_path).parent / raw)) - # Same TS/SvelteKit resolver fixups static imports use, so - # `await import('./foo')` (bare path), `import('./bar.shared')` - # (multi-dot helper), and Svelte 5 rune-file dynamic imports - # all land on real file nodes. - resolved = _resolve_js_module_path(resolved) - tgt_nid = _make_id(str(resolved)) - else: - aliases = _load_tsconfig_aliases(Path(str_path).parent) - resolved_alias = None - for alias_prefix, alias_base in aliases.items(): - if raw == alias_prefix or raw.startswith(alias_prefix + "/"): - rest = raw[len(alias_prefix):].lstrip("/") - resolved_alias = Path(os.path.normpath(Path(alias_base) / rest)) - break - if resolved_alias is not None: - resolved_alias = _resolve_js_module_path(resolved_alias) - tgt_nid = _make_id(str(resolved_alias)) - else: - module_name = raw.split("/")[-1] - if not module_name: - break - tgt_nid = _make_id(module_name) + # Resolve path using the same logic as static imports. + resolved = _resolve_js_import_target(raw, str_path) + if resolved is None: + break + tgt_nid, _ = resolved pair = (caller_nid, tgt_nid) if pair not in seen_dyn_pairs: seen_dyn_pairs.add(pair) @@ -489,6 +557,7 @@ def _dynamic_import_js(node, source: bytes, caller_nid: str, str_path: str, edge "source": caller_nid, "target": tgt_nid, "relation": "imports_from", + "context": "import", "confidence": "EXTRACTED", "source_file": str_path, "source_location": f"L{node.start_point[0] + 1}", @@ -4208,6 +4277,878 @@ def extract_powershell(path: Path) -> dict: # ── Cross-file import resolution ────────────────────────────────────────────── +def _source_key(source_file: str, root: Path) -> str: + if not source_file: + return "" + source_path = Path(source_file) + try: + return str(source_path.resolve().relative_to(root)) + except Exception: + return str(source_path) + + +def _disambiguate_colliding_node_ids( + nodes: list[dict], + edges: list[dict], + raw_calls: list[dict], + root: Path, +) -> None: + """Rewrite only colliding node IDs, using source path as the disambiguator.""" + by_id: dict[str, list[dict]] = {} + for node in nodes: + nid = node.get("id") + if isinstance(nid, str) and nid: + by_id.setdefault(nid, []).append(node) + + remap: dict[tuple[str, str], str] = {} + ambiguous_ids: set[str] = set() + for old_id, group in by_id.items(): + source_keys = {_source_key(str(node.get("source_file", "")), root) for node in group} + if len(group) < 2 or len(source_keys) < 2: + continue + ambiguous_ids.add(old_id) + for node in group: + source_key = _source_key(str(node.get("source_file", "")), root) + if not source_key: + continue + new_id = _make_id(source_key, old_id) + remap[(old_id, source_key)] = new_id + if new_id != old_id: + node["id"] = new_id + + if not remap: + return + + unambiguous_remaps: dict[str, str] = {} + for old_id, group in by_id.items(): + if old_id in ambiguous_ids: + continue + candidates = { + node["id"] for node in group + if isinstance(node.get("id"), str) and node["id"] != old_id + } + if len(candidates) == 1: + unambiguous_remaps[old_id] = next(iter(candidates)) + + for edge in edges: + edge_source_key = _source_key(str(edge.get("source_file", "")), root) + source_key = (edge.get("source", ""), edge_source_key) + target_key = (edge.get("target", ""), edge_source_key) + if source_key in remap: + edge["source"] = remap[source_key] + elif edge.get("source") in unambiguous_remaps: + edge["source"] = unambiguous_remaps[str(edge["source"])] + if target_key in remap: + edge["target"] = remap[target_key] + elif edge.get("target") in unambiguous_remaps: + edge["target"] = unambiguous_remaps[str(edge["target"])] + + for raw_call in raw_calls: + call_source_key = _source_key(str(raw_call.get("source_file", "")), root) + caller_key = (raw_call.get("caller_nid", ""), call_source_key) + if caller_key in remap: + raw_call["caller_nid"] = remap[caller_key] + elif raw_call.get("caller_nid") in unambiguous_remaps: + raw_call["caller_nid"] = unambiguous_remaps[str(raw_call["caller_nid"])] + + +def _node_label_key(node: dict) -> str: + label = str(node.get("label", "")).strip() + return re.sub(r"[^a-zA-Z0-9]+", "", label).lower() + + +def _is_type_like_definition(node: dict) -> bool: + label = str(node.get("label", "")).strip() + if not label: + return False + if label.endswith(")") or label.startswith("."): + return False + if "." in label: + return False + return node.get("file_type") == "code" + + +def _rewire_unique_stub_nodes(nodes: list[dict], edges: list[dict]) -> None: + """Map unresolved no-source stubs to a unique real definition with the same label.""" + real_by_label: dict[str, list[dict]] = {} + stubs: list[dict] = [] + + for node in nodes: + key = _node_label_key(node) + if not key: + continue + if node.get("source_file"): + if _is_type_like_definition(node): + real_by_label.setdefault(key, []).append(node) + continue + stubs.append(node) + + remap: dict[str, str] = {} + drop_ids: set[str] = set() + for stub in stubs: + stub_id = str(stub.get("id", "")) + if not stub_id: + continue + candidates = real_by_label.get(_node_label_key(stub), []) + if len(candidates) != 1: + continue + target_id = candidates[0].get("id") + if isinstance(target_id, str) and target_id and target_id != stub_id: + remap[stub_id] = target_id + drop_ids.add(stub_id) + + if not remap: + return + + for edge in edges: + if edge.get("source") in remap: + edge["source"] = remap[str(edge["source"])] + if edge.get("target") in remap: + edge["target"] = remap[str(edge["target"])] + + nodes[:] = [node for node in nodes if node.get("id") not in drop_ids] + + +def _js_source_path(source_file: str, root: Path) -> Path | None: + if not source_file: + return None + path = Path(source_file) + if not path.is_absolute(): + path = root / path + try: + return path.resolve() + except Exception: + return path + + +@dataclass(frozen=True) +class _SymbolDeclarationFact: + file_path: Path + name: str + line: int + + +@dataclass(frozen=True) +class _SymbolImportFact: + file_path: Path + local_name: str + target_path: Path + imported_name: str + line: int + + +@dataclass(frozen=True) +class _SymbolAliasFact: + file_path: Path + alias: str + target_name: str + line: int + + +@dataclass(frozen=True) +class _SymbolExportFact: + file_path: Path + exported_name: str + line: int + local_name: str | None = None + target_path: Path | None = None + target_name: str | None = None + + +@dataclass(frozen=True) +class _StarExportFact: + file_path: Path + target_path: Path + line: int + + +@dataclass(frozen=True) +class _SymbolUseFact: + file_path: Path + source_id: str + local_name: str + relation: str + context: str + line: int + + +@dataclass +class _SymbolResolutionFacts: + declarations: list[_SymbolDeclarationFact] = field(default_factory=list) + imports: list[_SymbolImportFact] = field(default_factory=list) + aliases: list[_SymbolAliasFact] = field(default_factory=list) + exports: list[_SymbolExportFact] = field(default_factory=list) + star_exports: list[_StarExportFact] = field(default_factory=list) + uses: list[_SymbolUseFact] = field(default_factory=list) + + +def _apply_symbol_resolution_facts( + paths: list[Path], + nodes: list[dict], + edges: list[dict], + root: Path, + facts: _SymbolResolutionFacts, +) -> None: + """Apply language-provided import/export/use facts to graph edges.""" + if not ( + facts.declarations + or facts.imports + or facts.aliases + or facts.exports + or facts.star_exports + or facts.uses + ): + return + + path_by_resolved = {path.resolve(): path for path in paths} + source_file_id = {path.resolve(): _make_id(str(path)) for path in paths} + symbol_nodes: dict[tuple[Path, str], str] = {} + for node in nodes: + source_path = _js_source_path(str(node.get("source_file", "")), root) + if source_path is None: + continue + label = str(node.get("label", "")).strip().strip("()").lstrip(".") + if label and node.get("id"): + symbol_nodes[(source_path, label)] = str(node["id"]) + + def ensure_symbol_node(path: Path, name: str, line: int) -> str: + resolved_path = path.resolve() + existing = symbol_nodes.get((resolved_path, name)) + if existing is not None: + return existing + node_id = _make_id(_file_stem(path), name) + symbol_nodes[(resolved_path, name)] = node_id + nodes.append({ + "id": node_id, + "label": name, + "file_type": "code", + "source_file": str(path), + "source_location": f"L{line}", + }) + return node_id + + existing_edges = { + (str(edge.get("source")), str(edge.get("target")), str(edge.get("relation"))) + for edge in edges + } + + def add_edge(source: str, target: str, relation: str, context: str, line: int, source_path: Path) -> None: + key = (source, target, relation) + if key in existing_edges: + return + existing_edges.add(key) + edges.append({ + "source": source, + "target": target, + "relation": relation, + "context": context, + "confidence": "EXTRACTED", + "source_file": str(source_path), + "source_location": f"L{line}", + "weight": 1.0, + }) + + for declaration in facts.declarations: + ensure_symbol_node(declaration.file_path, declaration.name, declaration.line) + + local_aliases_by_file: dict[Path, dict[str, tuple[Path, str]]] = {} + for import_fact in facts.imports: + file_path = import_fact.file_path.resolve() + local_aliases_by_file.setdefault(file_path, {})[import_fact.local_name] = ( + import_fact.target_path.resolve(), + import_fact.imported_name, + ) + + pending_aliases_by_file: dict[Path, list[_SymbolAliasFact]] = {} + for alias_fact in facts.aliases: + pending_aliases_by_file.setdefault(alias_fact.file_path.resolve(), []).append(alias_fact) + + for file_path, aliases in pending_aliases_by_file.items(): + local_aliases = local_aliases_by_file.setdefault(file_path, {}) + changed = True + while changed: + changed = False + for alias_fact in aliases: + if alias_fact.alias in local_aliases: + continue + origin = local_aliases.get(alias_fact.target_name) + if origin is not None: + local_aliases[alias_fact.alias] = origin + changed = True + + named_exports_by_file: dict[Path, dict[str, tuple[Path, str]]] = {} + star_exports_by_file: dict[Path, list[Path]] = {} + + for star_fact in facts.star_exports: + source_path = star_fact.file_path.resolve() + target_path = star_fact.target_path.resolve() + star_exports_by_file.setdefault(source_path, []).append(target_path) + source_id = source_file_id.get(source_path) + if source_id is not None: + add_edge( + source_id, + _make_id(str(path_by_resolved.get(target_path, target_path))), + "re_exports", + "export", + star_fact.line, + star_fact.file_path, + ) + + for export_fact in facts.exports: + file_path = export_fact.file_path.resolve() + origin: tuple[Path, str] | None = None + if export_fact.target_path is not None and export_fact.target_name is not None: + origin = (export_fact.target_path.resolve(), export_fact.target_name) + elif export_fact.local_name is not None: + origin = local_aliases_by_file.get(file_path, {}).get(export_fact.local_name) + if origin is None and (file_path, export_fact.local_name) in symbol_nodes: + origin = (file_path, export_fact.local_name) + if origin is None: + continue + named_exports_by_file.setdefault(file_path, {})[export_fact.exported_name] = origin + if origin[0] != file_path: + source_id = source_file_id.get(file_path) + if source_id is not None: + add_edge( + source_id, + _make_id(str(path_by_resolved.get(origin[0], origin[0]))), + "re_exports", + "export", + export_fact.line, + export_fact.file_path, + ) + + def resolve_exported_origin(target_path: Path, imported_name: str, seen: set[tuple[Path, str]] | None = None) -> tuple[Path, str]: + target_path = target_path.resolve() + key = (target_path, imported_name) + if seen is None: + seen = set() + if key in seen: + return key + seen.add(key) + origin = named_exports_by_file.get(target_path, {}).get(imported_name) + if origin is not None: + return resolve_exported_origin(origin[0], origin[1], seen) + for star_target in star_exports_by_file.get(target_path, []): + star_key = (star_target, imported_name) + if star_key in symbol_nodes: + return star_key + resolved = resolve_exported_origin(star_target, imported_name, seen) + if resolved in symbol_nodes: + return resolved + return key + + for import_fact in facts.imports: + source_id = source_file_id.get(import_fact.file_path.resolve()) + if source_id is None: + continue + origin_path, origin_symbol = resolve_exported_origin( + import_fact.target_path, + import_fact.imported_name, + ) + target_id = symbol_nodes.get((origin_path, origin_symbol)) + if target_id is None: + continue + add_edge( + source_id, + target_id, + "imports", + "import", + import_fact.line, + import_fact.file_path, + ) + + for use_fact in facts.uses: + file_path = use_fact.file_path.resolve() + unresolved_origin = local_aliases_by_file.get(file_path, {}).get(use_fact.local_name) + if unresolved_origin is None: + continue + origin_path, origin_symbol = resolve_exported_origin(*unresolved_origin) + target_id = symbol_nodes.get((origin_path, origin_symbol)) + if target_id is None: + continue + add_edge( + use_fact.source_id, + target_id, + use_fact.relation, + use_fact.context, + use_fact.line, + use_fact.file_path, + ) + + +def _parse_js_tree(path: Path): + try: + from tree_sitter import Language, Parser + if path.suffix in (".ts", ".tsx"): + import tree_sitter_typescript as tstypescript + language = Language(tstypescript.language_typescript()) + else: + import tree_sitter_javascript as tsjavascript + language = Language(tsjavascript.language()) + source = path.read_bytes() + parser = Parser(language) + return source, parser.parse(source).root_node + except Exception: + return None + + +def _walk_js_tree(node): + yield node + for child in node.children: + yield from _walk_js_tree(child) + + +def _js_module_specifier(node, source: bytes) -> str | None: + source_node = node.child_by_field_name("source") + if source_node is None: + for child in node.children: + if child.type == "string": + source_node = child + break + if source_node is None: + return None + raw = _read_text(source_node, source).strip() + return raw.strip("'\"`") or None + + +def _js_named_specifiers(node, source: bytes, specifier_type: str) -> list[tuple[str, str]]: + pairs: list[tuple[str, str]] = [] + for child in _walk_js_tree(node): + if child.type != specifier_type: + continue + name_node = child.child_by_field_name("name") + if name_node is None: + continue + alias_node = child.child_by_field_name("alias") + name = _read_text(name_node, source) + exposed = _read_text(alias_node, source) if alias_node is not None else name + if name and exposed: + pairs.append((name, exposed)) + return pairs + + +def _js_export_clause(node): + for child in node.children: + if child.type == "export_clause": + return child + return None + + +def _js_export_statement_is_star(node) -> bool: + return any(child.type == "*" for child in node.children) + + +def _js_lexical_aliases(node, source: bytes) -> list[tuple[str, str]]: + aliases: list[tuple[str, str]] = [] + if node.type != "lexical_declaration": + return aliases + for child in node.children: + if child.type != "variable_declarator": + continue + name_node = child.child_by_field_name("name") + value_node = child.child_by_field_name("value") + if ( + name_node is not None + and value_node is not None + and value_node.type in ("identifier", "type_identifier") + ): + aliases.append((_read_text(name_node, source), _read_text(value_node, source))) + return aliases + + +def _js_exported_declaration_names(node, source: bytes) -> list[str]: + names: list[str] = [] + declaration = node.child_by_field_name("declaration") + if declaration is None: + return names + + if declaration.type == "lexical_declaration": + names.extend(alias for alias, _target in _js_lexical_aliases(declaration, source)) + return names + + if declaration.type in ( + "class_declaration", + "abstract_class_declaration", + "interface_declaration", + "type_alias_declaration", + "function_declaration", + ): + name_node = declaration.child_by_field_name("name") + if name_node is not None: + names.append(_read_text(name_node, source)) + return names + + +def _js_top_level_function_bodies(path: Path, root_node, source: bytes) -> list[tuple[str, object]]: + bodies: list[tuple[str, object]] = [] + stem = _file_stem(path) + for node in root_node.children: + if node.type == "function_declaration": + name_node = node.child_by_field_name("name") + body = node.child_by_field_name("body") + if name_node is not None and body is not None: + bodies.append((_make_id(stem, _read_text(name_node, source)), body)) + continue + if node.type != "lexical_declaration": + continue + for child in node.children: + if child.type != "variable_declarator": + continue + name_node = child.child_by_field_name("name") + value_node = child.child_by_field_name("value") + if ( + name_node is not None + and value_node is not None + and value_node.type == "arrow_function" + ): + bodies.append((_make_id(stem, _read_text(name_node, source)), value_node)) + return bodies + + +def _js_call_identifier(node, source: bytes) -> str | None: + if node.type != "call_expression": + return None + function_node = node.child_by_field_name("function") + if function_node is None: + for child in node.children: + if child.is_named: + function_node = child + break + if function_node is not None and function_node.type in ("identifier", "type_identifier"): + return _read_text(function_node, source) + return None + + +def _collect_js_symbol_resolution_facts(paths: list[Path], facts: _SymbolResolutionFacts) -> None: + js_paths = [ + path for path in paths + if path.suffix in _JS_CACHE_BYPASS_SUFFIXES and path.suffix != ".vue" + ] + if not js_paths: + return + + trees: dict[Path, tuple[bytes, object]] = {} + + for path in js_paths: + resolved_path = path.resolve() + parsed = _parse_js_tree(path) + if parsed is None: + continue + source, root_node = parsed + trees[resolved_path] = parsed + + for node in _walk_js_tree(root_node): + if node.type == "export_statement": + for name in _js_exported_declaration_names(node, source): + facts.declarations.append( + _SymbolDeclarationFact(path, name, node.start_point[0] + 1) + ) + + if node.type != "import_statement": + continue + raw_module = _js_module_specifier(node, source) + if raw_module is None: + continue + target_path = _resolve_js_module_path(raw_module, path.parent) + if target_path is None: + continue + target_path = target_path.resolve() + for imported_name, local_name in _js_named_specifiers(node, source, "import_specifier"): + facts.imports.append( + _SymbolImportFact( + path, + local_name, + target_path, + imported_name, + node.start_point[0] + 1, + ) + ) + + for node in _walk_js_tree(root_node): + for alias, target in _js_lexical_aliases(node, source): + facts.aliases.append( + _SymbolAliasFact(path, alias, target, node.start_point[0] + 1) + ) + + for path in js_paths: + resolved_path = path.resolve() + parsed = trees.get(resolved_path) + if parsed is None: + continue + source, root_node = parsed + + for node in _walk_js_tree(root_node): + if node.type != "export_statement": + continue + + raw_module = _js_module_specifier(node, source) + export_clause = _js_export_clause(node) + if raw_module is not None: + target_path = _resolve_js_module_path(raw_module, path.parent) + if target_path is None: + continue + target_path = target_path.resolve() + if _js_export_statement_is_star(node): + facts.star_exports.append( + _StarExportFact(path, target_path, node.start_point[0] + 1) + ) + if export_clause is not None: + for original_name, exported_name in _js_named_specifiers( + export_clause, source, "export_specifier" + ): + facts.exports.append( + _SymbolExportFact( + path, + exported_name, + node.start_point[0] + 1, + target_path=target_path, + target_name=original_name, + ) + ) + continue + + if export_clause is not None: + for local_name, exported_name in _js_named_specifiers( + export_clause, source, "export_specifier" + ): + facts.exports.append( + _SymbolExportFact( + path, + exported_name, + node.start_point[0] + 1, + local_name=local_name, + ) + ) + continue + + for exported_name in _js_exported_declaration_names(node, source): + facts.exports.append( + _SymbolExportFact( + path, + exported_name, + node.start_point[0] + 1, + local_name=exported_name, + ) + ) + + for path in js_paths: + resolved_path = path.resolve() + parsed = trees.get(resolved_path) + if parsed is None: + continue + source, root_node = parsed + for source_id, body in _js_top_level_function_bodies(path, root_node, source): + for node in _walk_js_tree(body): + imported_name = _js_call_identifier(node, source) + if imported_name is None: + continue + facts.uses.append( + _SymbolUseFact( + path, + source_id, + imported_name, + "calls", + "call", + node.start_point[0] + 1, + ) + ) + + +def _parse_python_tree(path: Path): + try: + from tree_sitter import Language, Parser + import tree_sitter_python as tspython + source = path.read_bytes() + parser = Parser(Language(tspython.language())) + return source, parser.parse(source).root_node + except Exception: + return None + + +def _walk_python_tree(node): + yield node + for child in node.children: + yield from _walk_python_tree(child) + + +def _python_import_from_module(node, source: bytes) -> tuple[int, str] | None: + level = 0 + module_name = "" + for child in node.children: + if child.type == "import": + break + if child.type == "relative_import": + raw = _read_text(child, source) + level = len(raw) - len(raw.lstrip(".")) + remainder = raw.lstrip(".") + if remainder: + module_name = remainder + for sub in child.children: + if sub.type == "dotted_name": + module_name = _read_text(sub, source) + elif child.type == "dotted_name": + module_name = _read_text(child, source) + if level == 0 and not module_name: + return None + return level, module_name + + +def _python_imported_names(node, source: bytes) -> list[tuple[str, str]]: + names: list[tuple[str, str]] = [] + past_import = False + for child in node.children: + if child.type == "import": + past_import = True + continue + if not past_import: + continue + if child.type == "dotted_name": + name = _read_text(child, source) + names.append((name, name.split(".")[-1])) + elif child.type == "aliased_import": + name_node = child.child_by_field_name("name") + alias_node = child.child_by_field_name("alias") + if name_node is None: + continue + name = _read_text(name_node, source) + local = _read_text(alias_node, source) if alias_node is not None else name.split(".")[-1] + names.append((name, local)) + return names + + +def _resolve_python_module_path(module_name: str, current_path: Path, root: Path, level: int) -> Path | None: + if level > 0: + base = current_path.parent + for _ in range(level - 1): + base = base.parent + candidate = base / module_name.replace(".", "/") if module_name else base + else: + candidate = root / module_name.replace(".", "/") + + if candidate.is_dir(): + init_path = candidate / "__init__.py" + if init_path.is_file(): + return init_path + if candidate.is_file(): + return candidate + py_candidate = candidate.with_suffix(".py") + if py_candidate.is_file(): + return py_candidate + return None + + +def _python_top_level_function_bodies(path: Path, root_node, source: bytes) -> list[tuple[str, object]]: + bodies: list[tuple[str, object]] = [] + stem = _file_stem(path) + for node in root_node.children: + if node.type != "function_definition": + continue + name_node = node.child_by_field_name("name") + body = node.child_by_field_name("body") + if name_node is not None and body is not None: + bodies.append((_make_id(stem, _read_text(name_node, source)), body)) + return bodies + + +def _python_call_identifier(node, source: bytes) -> str | None: + if node.type != "call": + return None + function_node = node.child_by_field_name("function") + if function_node is not None and function_node.type == "identifier": + return _read_text(function_node, source) + return None + + +def _collect_python_symbol_resolution_facts( + paths: list[Path], + root: Path, + facts: _SymbolResolutionFacts, +) -> None: + py_paths = [path for path in paths if path.suffix == ".py"] + if not py_paths: + return + + trees: dict[Path, tuple[bytes, object]] = {} + for path in py_paths: + parsed = _parse_python_tree(path) + if parsed is None: + continue + source, root_node = parsed + trees[path.resolve()] = parsed + + for node in _walk_python_tree(root_node): + if node.type != "import_from_statement": + continue + module = _python_import_from_module(node, source) + if module is None: + continue + level, module_name = module + target_path = _resolve_python_module_path(module_name, path, root, level) + if target_path is None: + continue + for imported_name, local_name in _python_imported_names(node, source): + line = node.start_point[0] + 1 + facts.imports.append( + _SymbolImportFact(path, local_name, target_path, imported_name, line) + ) + if path.name == "__init__.py": + facts.exports.append( + _SymbolExportFact( + path, + local_name, + line, + target_path=target_path, + target_name=imported_name, + ) + ) + + for path in py_paths: + parsed = trees.get(path.resolve()) + if parsed is None: + continue + source, root_node = parsed + for source_id, body in _python_top_level_function_bodies(path, root_node, source): + for node in _walk_python_tree(body): + imported_name = _python_call_identifier(node, source) + if imported_name is None: + continue + facts.uses.append( + _SymbolUseFact( + path, + source_id, + imported_name, + "calls", + "call", + node.start_point[0] + 1, + ) + ) + + +def _augment_symbol_resolution_edges( + paths: list[Path], + nodes: list[dict], + edges: list[dict], + root: Path, +) -> None: + facts = _SymbolResolutionFacts() + _collect_js_symbol_resolution_facts(paths, facts) + _collect_python_symbol_resolution_facts(paths, root, facts) + _apply_symbol_resolution_facts(paths, nodes, edges, root, facts) + + +def _augment_js_reexport_edges( + paths: list[Path], + nodes: list[dict], + edges: list[dict], + root: Path, +) -> None: + """Compatibility wrapper for the JS/TS symbol-resolution post-pass.""" + facts = _SymbolResolutionFacts() + _collect_js_symbol_resolution_facts(paths, facts) + _apply_symbol_resolution_facts(paths, nodes, edges, root, facts) + + def _resolve_cross_file_imports( per_file: list[dict], paths: list[Path], @@ -6384,18 +7325,20 @@ def _extract_single_file(args: tuple) -> tuple[int, dict]: path = Path(path_str) cache_root = Path(cache_root_str) _raise_recursion_limit() + bypass_cache = path.suffix in _JS_CACHE_BYPASS_SUFFIXES # Check cache first (avoid re-extraction) - cached = load_cached(path, cache_root) - if cached is not None: - return idx, cached + if not bypass_cache: + cached = load_cached(path, cache_root) + if cached is not None: + return idx, cached extractor = _get_extractor(path) if extractor is None: return idx, {"nodes": [], "edges": []} result = _safe_extract(extractor, path) - if "error" not in result: + if not bypass_cache and "error" not in result: save_cached(path, result, cache_root) return idx, result @@ -6501,8 +7444,9 @@ def _extract_sequential( if extractor is None: per_file[idx] = {"nodes": [], "edges": []} continue + bypass_cache = path.suffix in _JS_CACHE_BYPASS_SUFFIXES result = _safe_extract(extractor, path) - if "error" not in result: + if not bypass_cache and "error" not in result: save_cached(path, result, effective_root) per_file[idx] = result if total_files >= _PROGRESS_INTERVAL: @@ -6538,6 +7482,8 @@ def extract( """ _check_tree_sitter_version() _raise_recursion_limit() + # Workspace package manifests/globs can change during watch or repeated extraction. + _WORKSPACE_PACKAGE_CACHE.clear() # Infer a common root for cache keys (use first diverging segment, not sum of all matches) try: @@ -6556,6 +7502,8 @@ def extract( root = Path(*paths[0].parts[:common_len]) if common_len else Path(".") except Exception: root = Path(".") + if cache_root is not None: + root = cache_root root = root.resolve() effective_root = cache_root or root @@ -6569,10 +7517,12 @@ def extract( if _get_extractor(path) is None: per_file[i] = {"nodes": [], "edges": []} continue - cached = load_cached(path, effective_root) - if cached is not None: - per_file[i] = cached - continue + bypass_cache = path.suffix in _JS_CACHE_BYPASS_SUFFIXES + if not bypass_cache: + cached = load_cached(path, effective_root) + if cached is not None: + per_file[i] = cached + continue uncached_work.append((i, path)) # Phase 2: extract uncached files (parallel or sequential) @@ -6592,9 +7542,13 @@ def extract( all_nodes: list[dict] = [] all_edges: list[dict] = [] + all_raw_calls: list[dict] = [] for result in per_file: all_nodes.extend(result.get("nodes", [])) all_edges.extend(result.get("edges", [])) + all_raw_calls.extend(result.get("raw_calls", [])) + + _augment_symbol_resolution_edges(paths, all_nodes, all_edges, root) # Remap file node IDs from absolute-path-derived to project-relative so # graph.json edge endpoints are stable across machines (#502) @@ -6618,6 +7572,8 @@ def extract( e["target"] = id_remap[e["target"]] _merge_swift_extensions(per_file, all_nodes, all_edges) + _disambiguate_colliding_node_ids(all_nodes, all_edges, all_raw_calls, root) + _rewire_unique_stub_nodes(all_nodes, all_edges) # Add cross-file class-level edges (Python only - uses Python parser internally) py_paths = [p for p in paths if p.suffix == ".py"] @@ -6689,53 +7645,52 @@ def extract( nid_to_file_nid[n["id"]] = _make_id(str(sf_rel)) existing_pairs = {(e["source"], e["target"]) for e in all_edges} - for result in per_file: - for rc in result.get("raw_calls", []): - callee = rc.get("callee", "") - if not callee: - continue - # Skip member-call callees: obj.log() → "log" has no import evidence - # and collides with any top-level function named "log" in the corpus. - if rc.get("is_member_call"): - continue - candidates = global_label_to_nids.get(callee.lower(), []) - # Skip ambiguous names that resolve to multiple nodes — these are - # common short names (log, execute, find) with no import evidence - # to pick the right target; emitting all edges inflates god_nodes. - if len(candidates) != 1: - continue - tgt = candidates[0] - caller = rc["caller_nid"] - if tgt != caller and (caller, tgt) not in existing_pairs: - existing_pairs.add((caller, tgt)) - # Promote to EXTRACTED when there's a direct import edge from the - # caller's file pointing at either the callee symbol itself or the - # file the callee lives in. - caller_file_nid = nid_to_file_nid.get(caller) - callee_file_nid = nid_to_file_nid.get(tgt) - imported_symbols = file_to_symbol_imports.get(caller_file_nid, set()) - imported_modules = file_to_module_imports.get(caller_file_nid, set()) - has_import_evidence = ( - tgt in imported_symbols - or (callee_file_nid is not None and callee_file_nid in imported_modules) - ) - if has_import_evidence: - confidence = "EXTRACTED" - confidence_score = 1.0 - else: - confidence = "INFERRED" - confidence_score = 0.8 - all_edges.append({ - "source": caller, - "target": tgt, - "relation": "calls", - "context": "call", - "confidence": confidence, - "confidence_score": confidence_score, - "source_file": rc.get("source_file", ""), - "source_location": rc.get("source_location"), - "weight": 1.0, - }) + for rc in all_raw_calls: + callee = rc.get("callee", "") + if not callee: + continue + # Skip member-call callees: obj.log() → "log" has no import evidence + # and collides with any top-level function named "log" in the corpus. + if rc.get("is_member_call"): + continue + candidates = global_label_to_nids.get(callee.lower(), []) + # Skip ambiguous names that resolve to multiple nodes — these are + # common short names (log, execute, find) with no import evidence + # to pick the right target; emitting all edges inflates god_nodes. + if len(candidates) != 1: + continue + tgt = candidates[0] + caller = rc["caller_nid"] + if tgt != caller and (caller, tgt) not in existing_pairs: + existing_pairs.add((caller, tgt)) + # Promote to EXTRACTED when there's a direct import edge from the + # caller's file pointing at either the callee symbol itself or the + # file the callee lives in. + caller_file_nid = nid_to_file_nid.get(caller) + callee_file_nid = nid_to_file_nid.get(tgt) + imported_symbols = file_to_symbol_imports.get(caller_file_nid, set()) + imported_modules = file_to_module_imports.get(caller_file_nid, set()) + has_import_evidence = ( + tgt in imported_symbols + or (callee_file_nid is not None and callee_file_nid in imported_modules) + ) + if has_import_evidence: + confidence = "EXTRACTED" + confidence_score = 1.0 + else: + confidence = "INFERRED" + confidence_score = 0.8 + all_edges.append({ + "source": caller, + "target": tgt, + "relation": "calls", + "context": "call", + "confidence": confidence, + "confidence_score": confidence_score, + "source_file": rc.get("source_file", ""), + "source_location": rc.get("source_location"), + "weight": 1.0, + }) # Relativize source_file fields so paths are portable across machines (#555) for item in all_nodes + all_edges: diff --git a/graphify/skill-windows.md b/graphify/skill-windows.md index 2c9a034f8..24d6800a0 100644 --- a/graphify/skill-windows.md +++ b/graphify/skill-windows.md @@ -957,7 +957,7 @@ from graphify.detect import save_manifest save_manifest(incremental['files']) print('[graphify update] Manifest saved.') '@ | Out-File -FilePath graphify-out\.graphify_step_for_update_incremental_re_extracti_21.py -Encoding utf8 -& (Get-Content graphify-out\.graphify_python) graphify-out\.graphify_step_for_update_incremental_re_extracti_21.py +& (Get-Content graphify-out\.graphify_python) graphify-out\.graphify_step_for_update_incremental_re_extracti_21.py Remove-Item -ErrorAction SilentlyContinue graphify-out\.graphify_step_for_update_incremental_re_extracti_21.py ``` diff --git a/tests/test_affected_cli.py b/tests/test_affected_cli.py new file mode 100644 index 000000000..65a3be8ca --- /dev/null +++ b/tests/test_affected_cli.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import json + +import networkx as nx +from networkx.readwrite import json_graph + +import graphify.__main__ as mainmod + + +def _write_graph(tmp_path): + graph = nx.DiGraph() + graph.add_node("target", label="Foo", source_file="pkg/foo.py", source_location="L1") + graph.add_node("caller", label="X()", source_file="app.py", source_location="L4") + graph.add_node("barrel", label="__init__.py", source_file="pkg/__init__.py", source_location=None) + graph.add_node("consumer", label="app.py", source_file="app.py", source_location=None) + graph.add_edge("caller", "target", relation="calls", context="call", confidence="EXTRACTED") + graph.add_edge("barrel", "target", relation="re_exports", context="export", confidence="EXTRACTED") + graph.add_edge("consumer", "target", relation="imports", context="import", confidence="EXTRACTED") + graph_path = tmp_path / "graph.json" + graph_path.write_text(json.dumps(json_graph.node_link_data(graph, edges="links")), encoding="utf-8") + return graph_path + + +def test_affected_cli_reverse_traverses_impact_edges(monkeypatch, tmp_path, capsys): + graph_path = _write_graph(tmp_path) + monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None) + monkeypatch.setattr( + mainmod.sys, + "argv", + ["graphify", "affected", "Foo", "--graph", str(graph_path)], + ) + + mainmod.main() + + out = capsys.readouterr().out + assert "Affected nodes for Foo" in out + assert "X()" in out + assert "calls" in out + assert "__init__.py" in out + assert "re_exports" in out + assert "app.py" in out + assert "imports" in out + + +def test_affected_cli_relation_filter_limits_reverse_traversal(monkeypatch, tmp_path, capsys): + graph_path = _write_graph(tmp_path) + monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None) + monkeypatch.setattr( + mainmod.sys, + "argv", + ["graphify", "affected", "Foo", "--relation", "calls", "--graph", str(graph_path)], + ) + + mainmod.main() + + out = capsys.readouterr().out + assert "Relations: calls" in out + assert "X()" in out + assert "__init__.py" not in out diff --git a/tests/test_extract.py b/tests/test_extract.py index 8f2a996c8..ea2bbc7e8 100644 --- a/tests/test_extract.py +++ b/tests/test_extract.py @@ -56,6 +56,159 @@ def test_extract_merges_multiple_files(): assert result["input_tokens"] == 0 +def test_extract_disambiguates_duplicate_symbol_ids_by_source_path(tmp_path): + first = tmp_path / "apps/api/Program.cs" + second = tmp_path / "tools/api/Program.cs" + first.parent.mkdir(parents=True) + second.parent.mkdir(parents=True) + first.write_text("class Program { void Run() {} }\n", encoding="utf-8") + second.write_text("class Program { void Run() {} }\n", encoding="utf-8") + + result = extract([first, second], cache_root=tmp_path) + program_nodes = [ + node for node in result["nodes"] + if node["label"] == "Program" and node.get("source_file", "").endswith("Program.cs") + ] + + assert len(program_nodes) == 2 + assert len({node["id"] for node in program_nodes}) == 2 + + node_ids = {node["id"] for node in result["nodes"]} + program_by_source = {node["source_file"]: node["id"] for node in program_nodes} + file_nodes_by_source = { + node["source_file"]: node["id"] + for node in result["nodes"] + if node["label"] == "Program.cs" + } + + assert set(program_by_source) == set(file_nodes_by_source) + contains_edges = [ + edge for edge in result["edges"] + if edge["relation"] == "contains" and edge["source_file"] in program_by_source + ] + assert len(contains_edges) == 2 + for edge in contains_edges: + assert edge["source"] == file_nodes_by_source[edge["source_file"]] + assert edge["target"] == program_by_source[edge["source_file"]] + + for edge in result["edges"]: + if edge["relation"] in {"contains", "method"}: + assert edge["source"] in node_ids, f"Dangling structural source: {edge}" + assert edge["target"] in node_ids, f"Dangling structural target: {edge}" + + +def test_extract_updates_raw_call_callers_after_duplicate_id_disambiguation(tmp_path): + first = tmp_path / "apps/api/Program.cs" + second = tmp_path / "tools/api/Program.cs" + target = tmp_path / "shared/Helper.cs" + first.parent.mkdir(parents=True) + second.parent.mkdir(parents=True) + target.parent.mkdir(parents=True) + first.write_text("class Program { void Run() { SharedHelper(); } }\n", encoding="utf-8") + second.write_text("class Program { void Run() {} }\n", encoding="utf-8") + target.write_text("class Helper { void SharedHelper() {} }\n", encoding="utf-8") + + result = extract([first, second, target], cache_root=tmp_path) + node_ids = {node["id"] for node in result["nodes"]} + + for edge in result["edges"]: + if edge["relation"] == "calls": + assert edge["source"] in node_ids + assert edge["target"] in node_ids + + +def test_extract_rewires_unique_inheritance_stub_to_real_definition(tmp_path): + definition = tmp_path / "interfaces.py" + implementation = tmp_path / "services/BookStore.cs" + definition.write_text("class BookStore:\n pass\n", encoding="utf-8") + implementation.parent.mkdir(parents=True) + implementation.write_text("class SqliteBookStore : BookStore { }\n", encoding="utf-8") + + result = extract([definition, implementation], cache_root=tmp_path) + node_by_id = {node["id"]: node for node in result["nodes"]} + inherits_edges = [edge for edge in result["edges"] if edge["relation"] == "inherits"] + + matching = [ + edge for edge in inherits_edges + if node_by_id[edge["source"]]["label"] == "SqliteBookStore" + and node_by_id[edge["target"]]["label"] == "BookStore" + ] + + assert matching + assert matching[0]["target"] == next( + node["id"] for node in result["nodes"] + if node["label"] == "BookStore" and node.get("source_file") == "interfaces.py" + ) + assert all( + not (node["label"] == "BookStore" and not node.get("source_file")) + for node in result["nodes"] + ) + + +def test_extract_keeps_stub_when_multiple_real_definitions_match(tmp_path): + first = tmp_path / "a/interfaces.py" + second = tmp_path / "b/interfaces.py" + implementation = tmp_path / "services/BookStore.cs" + first.parent.mkdir(parents=True) + second.parent.mkdir(parents=True) + implementation.parent.mkdir(parents=True) + first.write_text("class BookStore:\n pass\n", encoding="utf-8") + second.write_text("class BookStore:\n pass\n", encoding="utf-8") + implementation.write_text("class SqliteBookStore : BookStore { }\n", encoding="utf-8") + + result = extract([first, second, implementation], cache_root=tmp_path) + stubs = [ + node for node in result["nodes"] + if node["label"] == "BookStore" and not node.get("source_file") + ] + + assert stubs + + +def test_extract_does_not_rewire_inheritance_stub_to_same_named_function(tmp_path): + definition = tmp_path / "factory.py" + implementation = tmp_path / "services/BookStore.cs" + definition.write_text("def BookStore():\n return object()\n", encoding="utf-8") + implementation.parent.mkdir(parents=True) + implementation.write_text("class SqliteBookStore : BookStore { }\n", encoding="utf-8") + + result = extract([definition, implementation], cache_root=tmp_path) + node_by_id = {node["id"]: node for node in result["nodes"]} + inherits_edges = [edge for edge in result["edges"] if edge["relation"] == "inherits"] + + assert any( + node["label"] == "BookStore" and not node.get("source_file") + for node in result["nodes"] + ) + assert not any( + node_by_id[edge["source"]]["label"] == "SqliteBookStore" + and node_by_id[edge["target"]]["label"] == "BookStore()" + for edge in inherits_edges + ) + + +def test_extract_does_not_rewire_constructor_method_to_same_named_class(tmp_path): + source = tmp_path / "Sample.java" + source.write_text( + "class DataProcessor {\n" + " public DataProcessor() {}\n" + "}\n", + encoding="utf-8", + ) + + result = extract([source], cache_root=tmp_path) + + constructor_nodes = [ + node for node in result["nodes"] + if node["label"] == ".DataProcessor()" + ] + assert constructor_nodes + assert not any( + edge["source"] == edge["target"] + for edge in result["edges"] + ) + + def test_collect_files_from_dir(): from graphify.extract import _DISPATCH files = collect_files(FIXTURES) diff --git a/tests/test_js_import_resolution.py b/tests/test_js_import_resolution.py new file mode 100644 index 000000000..bb85f756d --- /dev/null +++ b/tests/test_js_import_resolution.py @@ -0,0 +1,399 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from graphify.extract import _file_stem, _make_id, extract + + +def _write(path: Path, text: str) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + +def _extract_for(paths: list[Path], root: Path): + return extract(paths, cache_root=root) + + +def _has_edge(result: dict, source: str, target: str, relation: str = "imports_from") -> bool: + expected = (_make_id(source), _make_id(target), relation) + actual = { + (edge["source"], edge["target"], edge["relation"]) + for edge in result["edges"] + } + return expected in actual + + +def _has_symbol_edge( + result: dict, + source: str, + target_file: str, + symbol: str, + relation: str = "imports", +) -> bool: + expected = (_make_id(source), _make_id(_file_stem(Path(target_file)), symbol), relation) + actual = { + (edge["source"], edge["target"], edge["relation"]) + for edge in result["edges"] + } + return expected in actual + + +def _has_symbol_to_symbol_edge( + result: dict, + source_file: str, + source_symbol: str, + target_file: str, + target_symbol: str, + relation: str, +) -> bool: + expected = ( + _make_id(_file_stem(Path(source_file)), source_symbol), + _make_id(_file_stem(Path(target_file)), target_symbol), + relation, + ) + actual = { + (edge["source"], edge["target"], edge["relation"]) + for edge in result["edges"] + } + return expected in actual + + +def _has_no_symbol_to_symbol_edge( + result: dict, + source_file: str, + source_symbol: str, + target_file: str, + target_symbol: str, + relation: str, +) -> bool: + return not _has_symbol_to_symbol_edge( + result, + source_file, + source_symbol, + target_file, + target_symbol, + relation, + ) + + +def test_ts_bare_relative_import_resolves_existing_ts_file(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export const foo = 1\n") + importer = _write( + tmp_path / "src/lib/page.ts", + "import { foo } from './foo'\nconsole.log(foo)\n", + ) + + result = _extract_for([target, importer], tmp_path) + + assert _has_edge(result, "src/lib/page.ts", "src/lib/foo.ts") + + +def test_ts_directory_import_resolves_index_ts(tmp_path: Path): + target = _write(tmp_path / "src/lib/server/queue/index.ts", "export const queue = 1\n") + importer = _write( + tmp_path / "src/lib/page.ts", + "import { queue } from './server/queue'\nconsole.log(queue)\n", + ) + + result = _extract_for([target, importer], tmp_path) + + assert _has_edge(result, "src/lib/page.ts", "src/lib/server/queue/index.ts") + + +def test_ts_named_reexport_alias_from_index_resolves_imported_symbol_to_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export class InternalFoo { id = '' }\n") + barrel = _write( + tmp_path / "src/lib/index.ts", + "export { InternalFoo as Foo } from './foo'\n", + ) + consumer = _write( + tmp_path / "src/routes/page.ts", + "import type { Foo } from '../lib/index'\nexport type X = Foo\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge( + result, + "src/routes/page.ts", + "src/lib/foo.ts", + "InternalFoo", + ) + + +def test_ts_export_star_from_index_resolves_imported_symbol_to_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export class Foo { id = '' }\n") + barrel = _write(tmp_path / "src/lib/index.ts", "export * from './foo'\n") + consumer = _write( + tmp_path / "src/routes/page.ts", + "import type { Foo } from '../lib/index'\nexport type X = Foo\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge(result, "src/routes/page.ts", "src/lib/foo.ts", "Foo") + + +def test_ts_import_alias_then_reexport_alias_resolves_imported_symbol_to_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export class Foo { id = '' }\n") + barrel = _write( + tmp_path / "src/lib/index.ts", + "import type { Foo as LocalFoo } from './foo'\nexport type { LocalFoo as PublicFoo }\n", + ) + consumer = _write( + tmp_path / "src/routes/page.ts", + "import type { PublicFoo } from '../lib/index'\nexport type X = PublicFoo\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge(result, "src/routes/page.ts", "src/lib/foo.ts", "Foo") + + +def test_ts_import_from_index_then_exported_type_alias_resolves_to_origin_symbol(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export class Foo { id = '' }\n") + barrel = _write(tmp_path / "src/lib/index.ts", "export { Foo } from './foo'\n") + consumer = _write( + tmp_path / "src/routes/page.ts", + "import type { Foo } from '../lib/index'\nexport type X = Foo\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge(result, "src/routes/page.ts", "src/lib/foo.ts", "Foo") + + +def test_ts_reexported_interface_resolves_imported_symbol_to_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export interface Foo { id: string }\n") + barrel = _write(tmp_path / "src/lib/index.ts", "export type { Foo } from './foo'\n") + consumer = _write( + tmp_path / "src/routes/page.ts", + "import type { Foo } from '../lib/index'\nexport type X = Foo\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge(result, "src/routes/page.ts", "src/lib/foo.ts", "Foo") + + +def test_ts_reexported_type_alias_resolves_imported_symbol_to_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export type Foo = { id: string }\n") + barrel = _write(tmp_path / "src/lib/index.ts", "export type { Foo } from './foo'\n") + consumer = _write( + tmp_path / "src/routes/page.ts", + "import type { Foo } from '../lib/index'\nexport type X = Foo\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge(result, "src/routes/page.ts", "src/lib/foo.ts", "Foo") + + +def test_ts_reexported_abstract_class_resolves_imported_symbol_to_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export abstract class Foo { abstract run(): void }\n") + barrel = _write(tmp_path / "src/lib/index.ts", "export { Foo } from './foo'\n") + consumer = _write( + tmp_path / "src/routes/page.ts", + "import { Foo } from '../lib/index'\nclass Impl extends Foo { run() {} }\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge(result, "src/routes/page.ts", "src/lib/foo.ts", "Foo") + + +def test_ts_const_alias_reexport_resolves_imported_symbol_to_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export class Foo { id = '' }\n") + barrel = _write( + tmp_path / "src/lib/index.ts", + "import { Foo } from './foo'\nexport const PublicFoo = Foo\n", + ) + consumer = _write( + tmp_path / "src/routes/page.ts", + "import { PublicFoo } from '../lib/index'\nnew PublicFoo()\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge(result, "src/routes/page.ts", "src/lib/foo.ts", "Foo") + + +def test_ts_local_const_alias_then_named_reexport_resolves_imported_symbol_to_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export function makeFoo() { return {} }\n") + barrel = _write( + tmp_path / "src/lib/index.ts", + "import { makeFoo } from './foo'\nconst PublicFactory = makeFoo\nexport { PublicFactory }\n", + ) + consumer = _write( + tmp_path / "src/routes/page.ts", + "import { PublicFactory } from '../lib/index'\nPublicFactory()\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_edge(result, "src/lib/index.ts", "src/lib/foo.ts", "re_exports") + assert _has_symbol_edge(result, "src/routes/page.ts", "src/lib/foo.ts", "makeFoo") + + +def test_ts_arrow_function_call_through_barrel_targets_origin_symbol(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export function Foo() { return 1 }\n") + unrelated = _write(tmp_path / "src/other/foo.ts", "export function Foo() { return 2 }\n") + barrel = _write(tmp_path / "src/lib/index.ts", "export { Foo } from './foo'\n") + consumer = _write( + tmp_path / "src/routes/page.ts", + "import { Foo } from '../lib/index'\nconst X = () => Foo()\n", + ) + + result = _extract_for([target, unrelated, barrel, consumer], tmp_path) + + assert _has_symbol_to_symbol_edge( + result, + "src/routes/page.ts", + "X", + "src/lib/foo.ts", + "Foo", + "calls", + ) + + +def test_ts_import_alias_does_not_affect_same_named_local_symbol_when_unused(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export function Foo() { return 1 }\n") + barrel = _write(tmp_path / "src/lib/index.ts", "export { Foo } from './foo'\n") + consumer = _write( + tmp_path / "src/routes/page.ts", + "import { Foo as Bar } from '../lib/index'\nconst Foo = () => {}\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_no_symbol_to_symbol_edge( + result, + "src/routes/page.ts", + "Foo", + "src/lib/foo.ts", + "Foo", + "calls", + ) + + +def test_ts_import_alias_call_from_same_named_local_symbol_targets_origin(tmp_path: Path): + target = _write(tmp_path / "src/lib/foo.ts", "export function Foo() { return 1 }\n") + barrel = _write(tmp_path / "src/lib/index.ts", "export { Foo } from './foo'\n") + consumer = _write( + tmp_path / "src/routes/page.ts", + "import { Foo as Bar } from '../lib/index'\nconst Foo = () => Bar()\n", + ) + + result = _extract_for([target, barrel, consumer], tmp_path) + + assert _has_symbol_to_symbol_edge( + result, + "src/routes/page.ts", + "Foo", + "src/lib/foo.ts", + "Foo", + "calls", + ) + + +def test_svelte_rune_import_resolves_svelte_ts_file(tmp_path: Path): + target = _write(tmp_path / "src/lib/hooks/is-mobile.svelte.ts", "export const isMobile = true\n") + importer = _write( + tmp_path / "src/routes/page.ts", + "import { isMobile } from '../lib/hooks/is-mobile.svelte'\nconsole.log(isMobile)\n", + ) + + result = _extract_for([target, importer], tmp_path) + + assert _has_edge(result, "src/routes/page.ts", "src/lib/hooks/is-mobile.svelte.ts") + + +def test_tsconfig_alias_import_resolves_existing_ts_file(tmp_path: Path): + _write( + tmp_path / "tsconfig.json", + json.dumps({"compilerOptions": {"baseUrl": ".", "paths": {"$lib/*": ["src/lib/*"]}}}), + ) + target = _write(tmp_path / "src/lib/types/type-helpers.ts", "export type Helper = string\n") + importer = _write( + tmp_path / "src/routes/page.ts", + "import type { Helper } from '$lib/types/type-helpers'\nconst value: Helper = 'x'\n", + ) + + result = _extract_for([target, importer], tmp_path) + + assert _has_edge(result, "src/routes/page.ts", "src/lib/types/type-helpers.ts") + + +def test_pnpm_workspace_package_import_resolves_package_entry(tmp_path: Path): + _write( + tmp_path / "pnpm-workspace.yaml", + "packages:\n - 'apps/*'\n - 'packages/*'\n", + ) + _write( + tmp_path / "packages/types/package.json", + json.dumps({"name": "@workspace/types", "exports": "./src/index.ts"}), + ) + target = _write( + tmp_path / "packages/types/src/index.ts", + "export interface SomeDto { id: string }\n", + ) + importer = _write( + tmp_path / "apps/web/src/page.ts", + "import type { SomeDto } from '@workspace/types'\nconst dto: SomeDto = { id: '1' }\n", + ) + + result = _extract_for([target, importer], tmp_path) + + assert _has_edge(result, "apps/web/src/page.ts", "packages/types/src/index.ts") + + +def test_js_import_resolution_ignores_stale_importer_cache_when_target_appears(tmp_path: Path): + importer = _write( + tmp_path / "src/lib/page.ts", + "import { foo } from './foo'\nconsole.log(foo)\n", + ) + + first = _extract_for([importer], tmp_path) + assert not _has_edge(first, "src/lib/page.ts", "src/lib/foo.ts") + + target = _write(tmp_path / "src/lib/foo.ts", "export const foo = 1\n") + second = _extract_for([target, importer], tmp_path) + + assert _has_edge(second, "src/lib/page.ts", "src/lib/foo.ts") + + +def test_workspace_package_cache_refreshes_between_extract_calls(tmp_path: Path): + _write( + tmp_path / "pnpm-workspace.yaml", + "packages:\n - 'apps/*'\n - 'packages/*'\n", + ) + importer = _write( + tmp_path / "apps/web/src/page.ts", + "import type { SomeDto } from '@workspace/types'\nconst dto: SomeDto = { id: '1' }\n", + ) + + first = _extract_for([importer], tmp_path) + assert not _has_edge(first, "apps/web/src/page.ts", "packages/types/src/index.ts") + + _write( + tmp_path / "packages/types/package.json", + json.dumps({"name": "@workspace/types", "exports": "./src/index.ts"}), + ) + target = _write( + tmp_path / "packages/types/src/index.ts", + "export interface SomeDto { id: string }\n", + ) + + second = _extract_for([target, importer], tmp_path) + + assert _has_edge(second, "apps/web/src/page.ts", "packages/types/src/index.ts") diff --git a/tests/test_multilang.py b/tests/test_multilang.py index 022d71736..a0e39c2d4 100644 --- a/tests/test_multilang.py +++ b/tests/test_multilang.py @@ -242,41 +242,46 @@ def test_cache_miss_after_file_change(tmp_path): # ── SQL ─────────────────────────────────────────────────────────────────────── +def _extract_sql_or_skip(fixture: str = "sample.sql"): + pytest.importorskip("tree_sitter_sql") + return extract_sql(FIXTURES / fixture) + + def test_sql_finds_tables(): - r = extract_sql(FIXTURES / "sample.sql") + r = _extract_sql_or_skip() labels = [n["label"] for n in r["nodes"]] assert any("users" in l for l in labels) assert any("organizations" in l for l in labels) def test_sql_finds_view(): - r = extract_sql(FIXTURES / "sample.sql") + r = _extract_sql_or_skip() labels = [n["label"] for n in r["nodes"]] assert any("active_users" in l for l in labels) def test_sql_finds_function(): - r = extract_sql(FIXTURES / "sample.sql") + r = _extract_sql_or_skip() labels = [n["label"] for n in r["nodes"]] assert any("get_user" in l for l in labels) def test_sql_emits_foreign_key_edge(): - r = extract_sql(FIXTURES / "sample.sql") + r = _extract_sql_or_skip() relations = {e["relation"] for e in r["edges"]} assert "references" in relations def test_sql_emits_reads_from_edge(): - r = extract_sql(FIXTURES / "sample.sql") + r = _extract_sql_or_skip() relations = {e["relation"] for e in r["edges"]} assert "reads_from" in relations def test_sql_no_dangling_edges(): - r = extract_sql(FIXTURES / "sample.sql") + r = _extract_sql_or_skip() node_ids = {n["id"] for n in r["nodes"]} for e in r["edges"]: assert e["source"] in node_ids, f"dangling source: {e['source']}" def test_sql_alter_table_fk_edge(): """ALTER TABLE ... FOREIGN KEY ... REFERENCES produces a references edge.""" - r = extract_sql(FIXTURES / "sample_alter_fk.sql") + r = _extract_sql_or_skip("sample_alter_fk.sql") fk_edges = [e for e in r["edges"] if e["relation"] == "references"] assert len(fk_edges) >= 1 node_ids = {n["id"] for n in r["nodes"]} @@ -286,14 +291,14 @@ def test_sql_alter_table_fk_edge(): def test_sql_schema_qualified_names(): """Schema-qualified table names (Schema.Table) are preserved.""" - r = extract_sql(FIXTURES / "sample_schema_qualified.sql") + r = _extract_sql_or_skip("sample_schema_qualified.sql") labels = [n["label"] for n in r["nodes"]] assert any("Sales.Customer" in l for l in labels) assert any("Sales.SalesOrder" in l for l in labels) def test_sql_schema_qualified_alter_fk(): """ALTER TABLE with schema-qualified names produces correct edges.""" - r = extract_sql(FIXTURES / "sample_schema_qualified.sql") + r = _extract_sql_or_skip("sample_schema_qualified.sql") fk_edges = [e for e in r["edges"] if e["relation"] == "references"] assert len(fk_edges) >= 1 node_ids = {n["id"] for n in r["nodes"]} diff --git a/tests/test_python_import_resolution.py b/tests/test_python_import_resolution.py new file mode 100644 index 000000000..fccdda307 --- /dev/null +++ b/tests/test_python_import_resolution.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from pathlib import Path + +from graphify.extract import extract + + +def _write(path: Path, text: str) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + +def _node_id(result: dict, label: str, source_file: str) -> str: + matches = [ + node["id"] + for node in result["nodes"] + if node.get("label") == label and node.get("source_file") == source_file + ] + assert len(matches) == 1 + return matches[0] + + +def _has_edge(result: dict, source: str, target: str, relation: str) -> bool: + return any( + edge["source"] == source + and edge["target"] == target + and edge["relation"] == relation + for edge in result["edges"] + ) + + +def test_python_package_reexport_resolves_import_and_call_to_origin_symbol(tmp_path: Path): + origin = _write(tmp_path / "pkg/foo.py", "def Foo():\n return 1\n") + barrel = _write(tmp_path / "pkg/__init__.py", "from .foo import Foo as PublicFoo\n") + consumer = _write( + tmp_path / "app.py", + "from pkg import PublicFoo\n\n" + "def X():\n" + " return PublicFoo()\n", + ) + + result = extract([origin, barrel, consumer], cache_root=tmp_path) + + origin_file = _node_id(result, "foo.py", "pkg/foo.py") + barrel_file = _node_id(result, "__init__.py", "pkg/__init__.py") + consumer_file = _node_id(result, "app.py", "app.py") + origin_symbol = _node_id(result, "Foo()", "pkg/foo.py") + consumer_symbol = _node_id(result, "X()", "app.py") + + assert _has_edge(result, barrel_file, origin_file, "re_exports") + assert _has_edge(result, consumer_file, origin_symbol, "imports") + assert _has_edge(result, consumer_symbol, origin_symbol, "calls")