merge PR #579: dynamic import() extraction for JS/TS

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Safi
2026-05-02 16:52:19 +01:00
co-authored by Claude Sonnet 4.6
3 changed files with 207 additions and 2 deletions
+89
View File
@@ -259,6 +259,85 @@ def _import_js(node, source: bytes, file_nid: str, stem: str, edges: list, str_p
})
def _dynamic_import_js(node, source: bytes, caller_nid: str, str_path: str, edges: list,
seen_dyn_pairs: set) -> bool:
"""Detect dynamic import() calls in JS/TS and emit imports_from edges.
Handles patterns like:
await import('./foo.js')
import('./foo.js').then(...)
const m = await import(`./foo`)
Returns True if the node was a dynamic import (caller should skip normal call handling).
"""
# Dynamic import is a call_expression whose function child is the keyword "import".
# tree-sitter-typescript parses `import('...')` as call_expression with first child
# being an "import" token (type="import").
func_node = node.child_by_field_name("function")
if func_node is None:
# Fallback: check first child directly (some TS versions)
if node.children and _read_text(node.children[0], source) == "import":
func_node = node.children[0]
else:
return False
if _read_text(func_node, source) != "import":
return False
# Extract the module path from the arguments
args = node.child_by_field_name("arguments")
if args is None:
return True # It's an import() but no args — skip
for arg in args.children:
if arg.type == "template_string":
# Skip dynamic template literals — path can't be statically resolved
if any(c.type == "template_substitution" for c in arg.children):
break
raw = _read_text(arg, source).strip("`")
elif arg.type == "string":
raw = _read_text(arg, source).strip("'\" ")
else:
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))
if resolved.suffix == ".js":
resolved = resolved.with_suffix(".ts")
elif resolved.suffix == ".jsx":
resolved = resolved.with_suffix(".tsx")
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:
tgt_nid = _make_id(str(resolved_alias))
else:
module_name = raw.split("/")[-1]
if not module_name:
break
tgt_nid = _make_id(module_name)
pair = (caller_nid, tgt_nid)
if pair not in seen_dyn_pairs:
seen_dyn_pairs.add(pair)
edges.append({
"source": caller_nid,
"target": tgt_nid,
"relation": "imports_from",
"confidence": "EXTRACTED",
"source_file": str_path,
"source_location": f"L{node.start_point[0] + 1}",
"weight": 1.0,
})
break
return True
def _import_java(node, source: bytes, file_nid: str, stem: str, edges: list, str_path: str) -> None:
def _walk_scoped(n) -> str:
parts: list[str] = []
@@ -1228,6 +1307,7 @@ def _extract_generic(path: Path, config: LanguageConfig) -> dict:
label_to_nid[normalised.lower()] = n["id"]
seen_call_pairs: set[tuple[str, str]] = set()
seen_dyn_import_pairs: set[tuple[str, str]] = set()
seen_static_ref_pairs: set[tuple[str, str, str]] = set()
seen_helper_ref_pairs: set[tuple[str, str, str]] = set()
seen_bind_pairs: set[tuple[str, str, str]] = set()
@@ -1249,6 +1329,15 @@ def _extract_generic(path: Path, config: LanguageConfig) -> dict:
return
if node.type in config.call_types:
# JS/TS dynamic imports: await import('./foo.js')
if config.ts_module in ("tree_sitter_javascript", "tree_sitter_typescript"):
if _dynamic_import_js(node, source, caller_nid, str_path,
edges, seen_dyn_import_pairs):
# Still recurse into children (import().then(...) may have calls)
for child in node.children:
walk_calls(child, caller_nid)
return
callee_name: str | None = None
is_member_call: bool = False
+32
View File
@@ -0,0 +1,32 @@
import { logger } from './logger';
async function processInbound(orgId: string, phone: string) {
const { shouldHandle, processMessage } = await import('./mayaEngine.js');
const handle = await shouldHandle(orgId, phone);
if (handle.sessionId) {
await processMessage({ orgId, phone }, handle.sessionId);
}
}
async function pollMessages(orgId: string) {
const { commsQueue } = await import('./queue.js');
await commsQueue.add('check-inbound', { orgId });
}
async function loadHandler(handlerName: string) {
// dynamic template literal — path not statically resolvable, should produce no edge
const mod = await import(`./handlers/${handlerName}`);
return mod.default;
}
async function loadStatic() {
// static template literal (no interpolation) — should resolve like a plain string
const { helper } = await import(`./staticHelper`);
return helper;
}
function syncOnly() {
logger.info('no dynamic imports here');
}
export { processInbound, pollMessages, loadHandler, loadStatic, syncOnly };
+86 -2
View File
@@ -1,15 +1,21 @@
"""Tests for language extractors: Java, C, C++, Ruby, C#, Kotlin, Scala, PHP, Swift, Go, Julia, VB.NET."""
"""Tests for language extractors: Java, C, C++, Ruby, C#, Kotlin, Scala, PHP, Swift, Go, Julia, VB.NET, JS/TS."""
from __future__ import annotations
from pathlib import Path
import pytest
from graphify.extract import (
extract_java, extract_c, extract_cpp, extract_ruby,
extract_csharp, extract_kotlin, extract_scala, extract_php,
extract_swift, extract_go, extract_julia, extract_vbnet,
extract_swift, extract_go, extract_julia, extract_vbnet, extract_js,
)
FIXTURES = Path(__file__).parent / "fixtures"
import importlib.util
vbnet_available = pytest.mark.skipif(
importlib.util.find_spec("tree_sitter_vbnet") is None,
reason="tree-sitter-vbnet not on PyPI yet"
)
def _labels(r):
return [n["label"] for n in r["nodes"]]
@@ -525,45 +531,55 @@ def test_swift_emits_calls():
# ── VB.NET ─────────────────────────────────────────────────────────────────────────
@vbnet_available
def test_vbnet_no_error():
r = extract_vbnet(FIXTURES / "sample.vb")
assert "error" not in r
@vbnet_available
def test_vbnet_finds_class():
r = extract_vbnet(FIXTURES / "sample.vb")
assert any("DataProcessor" in l for l in _labels(r))
@vbnet_available
def test_vbnet_finds_interface():
r = extract_vbnet(FIXTURES / "sample.vb")
assert any("IProcessor" in l for l in _labels(r))
@vbnet_available
def test_vbnet_finds_module():
r = extract_vbnet(FIXTURES / "sample.vb")
assert any("AppHelper" in l for l in _labels(r))
@vbnet_available
def test_vbnet_finds_structure():
r = extract_vbnet(FIXTURES / "sample.vb")
assert any("Point" in l for l in _labels(r))
@vbnet_available
def test_vbnet_finds_methods():
r = extract_vbnet(FIXTURES / "sample.vb")
labels = _labels(r)
assert any("Process" in l for l in labels)
assert any("Validate" in l for l in labels)
@vbnet_available
def test_vbnet_finds_sub():
r = extract_vbnet(FIXTURES / "sample.vb")
assert any("Run" in l for l in _labels(r))
@vbnet_available
def test_vbnet_finds_imports():
r = extract_vbnet(FIXTURES / "sample.vb")
assert "imports" in _relations(r)
@vbnet_available
def test_vbnet_inherits_edge():
r = extract_vbnet(FIXTURES / "sample.vb")
inherits = [e for e in r["edges"] if e["relation"] == "inherits"]
assert len(inherits) >= 1
@vbnet_available
def test_vbnet_inherits_baseprovessor():
r = extract_vbnet(FIXTURES / "sample.vb")
node_by_id = {n["id"]: n["label"] for n in r["nodes"]}
@@ -574,11 +590,13 @@ def test_vbnet_inherits_baseprovessor():
)
assert found, "DataProcessor should have inherits edge to BaseProcessor"
@vbnet_available
def test_vbnet_implements_edge():
r = extract_vbnet(FIXTURES / "sample.vb")
implements = [e for e in r["edges"] if e["relation"] == "implements"]
assert len(implements) >= 1
@vbnet_available
def test_vbnet_implements_iprocessor():
r = extract_vbnet(FIXTURES / "sample.vb")
node_by_id = {n["id"]: n["label"] for n in r["nodes"]}
@@ -589,6 +607,7 @@ def test_vbnet_implements_iprocessor():
)
assert found, "DataProcessor should have implements edge to IProcessor"
@vbnet_available
def test_vbnet_no_dangling_edges():
r = extract_vbnet(FIXTURES / "sample.vb")
node_ids = {n["id"] for n in r["nodes"]}
@@ -791,3 +810,68 @@ def test_julia_no_dangling_edges():
node_ids = {n["id"] for n in r["nodes"]}
for e in r["edges"]:
assert e["source"] in node_ids, f"Dangling source: {e}"
# ── TypeScript dynamic imports ───────────────────────────────────────────────
def test_ts_dynamic_import_no_error():
r = extract_js(FIXTURES / "dynamic_import.ts")
assert "error" not in r
def test_ts_dynamic_import_extracts_edges():
"""Dynamic import() calls inside functions should produce imports_from edges."""
r = extract_js(FIXTURES / "dynamic_import.ts")
dyn_edges = [e for e in r["edges"] if e["relation"] == "imports_from"]
targets = {e["target"] for e in dyn_edges}
# Should find: static ./logger, dynamic ./mayaEngine.js, dynamic ./queue.js
assert any("logger" in t for t in targets), f"Missing static import of logger: {targets}"
assert any("mayaengine" in t.lower() for t in targets), f"Missing dynamic import of mayaEngine: {targets}"
assert any("queue" in t.lower() for t in targets), f"Missing dynamic import of queue: {targets}"
def test_ts_dynamic_import_confidence():
"""Dynamic imports should have EXTRACTED confidence (they are deterministic string literals)."""
r = extract_js(FIXTURES / "dynamic_import.ts")
dyn_edges = [e for e in r["edges"]
if e["relation"] == "imports_from"
and "mayaengine" in e["target"].lower()]
assert len(dyn_edges) >= 1
assert dyn_edges[0]["confidence"] == "EXTRACTED"
def test_ts_dynamic_import_source_is_function():
"""Dynamic import edge source should be the enclosing function, not the file."""
r = extract_js(FIXTURES / "dynamic_import.ts")
node_labels = {n["id"]: n["label"] for n in r["nodes"]}
dyn_edges = [e for e in r["edges"]
if e["relation"] == "imports_from"
and "mayaengine" in e["target"].lower()]
assert len(dyn_edges) >= 1
src_label = node_labels.get(dyn_edges[0]["source"], "")
assert "processInbound" in src_label, f"Expected processInbound as source, got {src_label}"
def test_ts_no_dynamic_import_in_sync_fn():
"""Functions without dynamic imports should not get spurious imports_from edges."""
r = extract_js(FIXTURES / "dynamic_import.ts")
node_ids = {n["label"]: n["id"] for n in r["nodes"]}
sync_nid = node_ids.get("syncOnly()")
if sync_nid:
sync_imports = [e for e in r["edges"]
if e["source"] == sync_nid and e["relation"] == "imports_from"]
assert len(sync_imports) == 0
def test_ts_dynamic_template_literal_skipped():
"""Dynamic template literals (with ${}) must not produce an imports_from edge."""
r = extract_js(FIXTURES / "dynamic_import.ts")
targets = {e["target"] for e in r["edges"] if e["relation"] == "imports_from"}
# loadHandler uses `./handlers/${handlerName}` — no static path, must be absent
assert not any("handler" in t.lower() and "$" in t for t in targets), \
f"Garbage edge from dynamic template literal found: {targets}"
# More robust: no target should contain a brace character
assert not any("{" in t or "}" in t for t in targets), \
f"Target contains unresolved template expression: {targets}"
def test_ts_static_template_literal_resolved():
"""Static template literals (no ${}) should resolve the same as a plain string."""
r = extract_js(FIXTURES / "dynamic_import.ts")
targets = {e["target"] for e in r["edges"] if e["relation"] == "imports_from"}
assert any("statichelper" in t.lower() for t in targets), \
f"Static template literal import not resolved: {targets}"