mirror of
https://github.com/safishamsi/graphify.git
synced 2026-10-23 12:21:30 +00:00
samie a.
918483e6a3
#3782 left a reference placed above `struct Handle` on a sourceless stub
whenever another file-level item folded to the same id (`fn handle`,
`const HANDLE`, an `impl` for `HANDLE`), because walk() gave that id to
whichever item came first. Those items now get ids of their own, so the
struct owns its id and the reference can resolve to it, as a reference
below the declaration already did. Without this, the stub only folds back
when the type's name is unique across the corpus.
A declared type is now pre-registered unless a non-type item has its
exact name (`fn Foo()` beside `struct Foo {}` still share one node), and a
`type HANDLE` alias still keeps its stub. The #3782 test that pinned the
stub for the case-only shapes now asserts the resolution; the shapes that
still share a node keep the stub assertion.
local_type_names becomes a frozenset, which is what
_rust_collect_type_refs is annotated to take.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit f710c8a16e)
1874 lines
82 KiB
Python
1874 lines
82 KiB
Python
"""Tests for multi-language AST extraction: JS/TS, Go, Rust, SQL."""
|
|
from __future__ import annotations
|
|
import shutil
|
|
from pathlib import Path
|
|
import pytest
|
|
from graphify.extract import extract_js, extract_go, extract_rust, extract, extract_sql
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
def _labels(result):
|
|
return [n["label"] for n in result["nodes"]]
|
|
|
|
def _call_pairs(result):
|
|
node_by_id = {n["id"]: n["label"] for n in result["nodes"]}
|
|
return {
|
|
(node_by_id.get(e["source"], e["source"]), node_by_id.get(e["target"], e["target"]))
|
|
for e in result["edges"] if e["relation"] == "calls"
|
|
}
|
|
|
|
def _confidences(result):
|
|
return {e["confidence"] for e in result["edges"]}
|
|
|
|
|
|
def _edges_with_relation(result, *relations):
|
|
return [e for e in result["edges"] if e["relation"] in relations]
|
|
|
|
|
|
def _normalize_symbol_label(label: str) -> str:
|
|
return label.strip("()").lstrip(".")
|
|
|
|
|
|
def _edge_labels(result, relation, context=None):
|
|
labels = {n["id"]: _normalize_symbol_label(n["label"]) for n in result["nodes"]}
|
|
pairs = set()
|
|
for e in result["edges"]:
|
|
if e.get("relation") != relation:
|
|
continue
|
|
if context is not None and e.get("context") != context:
|
|
continue
|
|
pairs.add((labels.get(e["source"], e["source"]), labels.get(e["target"], e["target"])))
|
|
return pairs
|
|
|
|
|
|
# ── TypeScript ────────────────────────────────────────────────────────────────
|
|
|
|
def test_ts_finds_class():
|
|
r = extract_js(FIXTURES / "sample.ts")
|
|
assert "error" not in r
|
|
assert "HttpClient" in _labels(r)
|
|
|
|
def test_ts_finds_methods():
|
|
r = extract_js(FIXTURES / "sample.ts")
|
|
labels = _labels(r)
|
|
assert any("get" in l for l in labels)
|
|
assert any("post" in l for l in labels)
|
|
|
|
def test_ts_finds_function():
|
|
r = extract_js(FIXTURES / "sample.ts")
|
|
assert any("buildHeaders" in l for l in _labels(r))
|
|
|
|
def test_ts_emits_calls():
|
|
r = extract_js(FIXTURES / "sample.ts")
|
|
calls = _call_pairs(r)
|
|
# .post() calls .get()
|
|
assert any("post" in src and "get" in tgt for src, tgt in calls)
|
|
|
|
def test_ts_calls_are_extracted():
|
|
r = extract_js(FIXTURES / "sample.ts")
|
|
for e in r["edges"]:
|
|
if e["relation"] == "calls":
|
|
assert e["confidence"] == "EXTRACTED"
|
|
|
|
|
|
def test_ts_import_edges_have_import_context():
|
|
r = extract_js(FIXTURES / "sample.ts")
|
|
import_edges = _edges_with_relation(r, "imports", "imports_from")
|
|
assert import_edges
|
|
assert all(e.get("context") == "import" for e in import_edges)
|
|
|
|
|
|
def test_ts_call_edges_have_call_context():
|
|
r = extract_js(FIXTURES / "sample.ts")
|
|
call_edges = _edges_with_relation(r, "calls")
|
|
assert call_edges
|
|
assert all(e.get("context") == "call" for e in call_edges)
|
|
|
|
def test_ts_no_dangling_edges():
|
|
r = extract_js(FIXTURES / "sample.ts")
|
|
node_ids = {n["id"] for n in r["nodes"]}
|
|
for e in r["edges"]:
|
|
if e["relation"] in ("contains", "method", "calls"):
|
|
assert e["source"] in node_ids
|
|
|
|
|
|
# ── Go ────────────────────────────────────────────────────────────────────────
|
|
|
|
def test_go_finds_struct():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
assert "error" not in r
|
|
assert "Server" in _labels(r)
|
|
|
|
def test_go_finds_methods():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
labels = _labels(r)
|
|
assert any("Start" in l for l in labels)
|
|
assert any("Stop" in l for l in labels)
|
|
|
|
def test_go_finds_constructor():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
assert any("NewServer" in l for l in _labels(r))
|
|
|
|
def test_go_emits_calls():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
# main() calls NewServer and Start
|
|
assert len(_call_pairs(r)) > 0
|
|
|
|
def test_go_has_extracted_calls():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
assert "EXTRACTED" in _confidences(r)
|
|
|
|
|
|
def test_go_import_edges_have_import_context():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
import_edges = _edges_with_relation(r, "imports", "imports_from")
|
|
assert import_edges
|
|
assert all(e.get("context") == "import" for e in import_edges)
|
|
|
|
|
|
def test_go_call_edges_have_call_context():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
call_edges = _edges_with_relation(r, "calls")
|
|
assert call_edges
|
|
assert all(e.get("context") == "call" for e in call_edges)
|
|
|
|
def test_go_no_dangling_edges():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
node_ids = {n["id"] for n in r["nodes"]}
|
|
for e in r["edges"]:
|
|
if e["relation"] in ("contains", "method", "calls"):
|
|
assert e["source"] in node_ids
|
|
|
|
|
|
def test_go_embeds_struct_field():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
assert ("DataProcessor", "BaseProcessor") in _edge_labels(r, "embeds")
|
|
|
|
|
|
def test_go_interface_embedding_emits_embeds():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
assert ("ReaderLogger", "Logger") in _edge_labels(r, "embeds")
|
|
|
|
|
|
def test_go_struct_named_field_emits_field_context():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
assert ("DataProcessor", "Result") in _edge_labels(r, "references", "field")
|
|
|
|
|
|
def test_go_method_parameter_return_contexts():
|
|
r = extract_go(FIXTURES / "sample.go")
|
|
assert ("Build", "DataProcessor") in _edge_labels(r, "references", "parameter_type")
|
|
assert ("Build", "Result") in _edge_labels(r, "references", "return_type")
|
|
|
|
|
|
def test_go_method_declaration_emits_refs_only_when_name_present():
|
|
"""Regression: review feedback flagged a hypothetical UnboundLocalError in
|
|
extract_go's method_declaration branch if `name_node` were None. Statically
|
|
verify that every use of `method_nid` (and the `emit_go_method_refs` call
|
|
that consumes it) is guarded by a `name_node` truthiness check — either
|
|
nested inside `if name_node:` or following an early `if not name_node: return`.
|
|
Same for function_declaration and `func_nid`.
|
|
"""
|
|
import ast
|
|
import inspect
|
|
from graphify.extract import extract_go
|
|
|
|
tree = ast.parse(inspect.getsource(extract_go))
|
|
|
|
def _find_branch(root: ast.AST, type_literal: str) -> ast.If | None:
|
|
"""Return the `if t == '<type_literal>':` branch inside the walk function."""
|
|
for child in ast.walk(root):
|
|
if (isinstance(child, ast.If)
|
|
and isinstance(child.test, ast.Compare)
|
|
and isinstance(child.test.left, ast.Name)
|
|
and child.test.left.id == "t"
|
|
and len(child.test.comparators) == 1
|
|
and isinstance(child.test.comparators[0], ast.Constant)
|
|
and child.test.comparators[0].value == type_literal):
|
|
return child
|
|
return None
|
|
|
|
method_branch = _find_branch(tree, "method_declaration")
|
|
function_branch = _find_branch(tree, "function_declaration")
|
|
assert method_branch is not None, "method_declaration branch not found in extract_go"
|
|
assert function_branch is not None, "function_declaration branch not found in extract_go"
|
|
|
|
def _is_early_return_on_falsy_name_node(stmt: ast.AST) -> bool:
|
|
"""True iff `stmt` is `if not name_node: return` (or raise/continue/break)."""
|
|
if not isinstance(stmt, ast.If):
|
|
return False
|
|
test = stmt.test
|
|
is_falsy_check = (
|
|
isinstance(test, ast.UnaryOp)
|
|
and isinstance(test.op, ast.Not)
|
|
and isinstance(test.operand, ast.Name)
|
|
and test.operand.id == "name_node"
|
|
)
|
|
if not is_falsy_check:
|
|
return False
|
|
terminators = (ast.Return, ast.Raise, ast.Continue, ast.Break)
|
|
return any(isinstance(s, terminators) for s in stmt.body)
|
|
|
|
def _guarded_by_name_node(branch: ast.If, var_name: str) -> bool:
|
|
"""True iff every read of `var_name` in `branch` is guarded by a
|
|
`name_node` truthiness check — either lexically nested under
|
|
`if name_node:` or after a preceding `if not name_node: return`."""
|
|
parents: dict[int, ast.AST] = {}
|
|
for parent in ast.walk(branch):
|
|
for child in ast.iter_child_nodes(parent):
|
|
parents[id(child)] = parent
|
|
|
|
def _stmt_chain(start: ast.AST) -> list[tuple[ast.stmt, list[ast.stmt]]]:
|
|
"""Walk up to each enclosing statement-list, returning (stmt, siblings)."""
|
|
chain: list[tuple[ast.stmt, list[ast.stmt]]] = []
|
|
cur: ast.AST | None = start
|
|
while cur is not None:
|
|
parent = parents.get(id(cur))
|
|
if parent is None:
|
|
break
|
|
if isinstance(cur, ast.stmt):
|
|
for attr in ("body", "orelse", "finalbody"):
|
|
siblings = getattr(parent, attr, None)
|
|
if isinstance(siblings, list) and cur in siblings:
|
|
chain.append((cur, siblings))
|
|
break
|
|
cur = parent
|
|
return chain
|
|
|
|
def _is_guarded(use: ast.AST) -> bool:
|
|
for stmt, siblings in _stmt_chain(use):
|
|
parent = parents.get(id(stmt))
|
|
# Case 1: lexically nested under `if name_node:` body
|
|
if (isinstance(parent, ast.If)
|
|
and isinstance(parent.test, ast.Name)
|
|
and parent.test.id == "name_node"
|
|
and stmt in parent.body):
|
|
return True
|
|
# Case 2: a preceding sibling is `if not name_node: return`
|
|
idx = siblings.index(stmt)
|
|
if any(_is_early_return_on_falsy_name_node(s) for s in siblings[:idx]):
|
|
return True
|
|
return False
|
|
|
|
for node in ast.walk(branch):
|
|
if isinstance(node, ast.Name) and node.id == var_name:
|
|
if not _is_guarded(node):
|
|
return False
|
|
return True
|
|
|
|
assert _guarded_by_name_node(method_branch, "method_nid"), (
|
|
"method_nid use is not guarded by a name_node check in method_declaration branch"
|
|
)
|
|
assert _guarded_by_name_node(function_branch, "func_nid"), (
|
|
"func_nid use is not guarded by a name_node check in function_declaration branch"
|
|
)
|
|
|
|
# Negative control: confirm the checker would actually reject the buggy
|
|
# layout the reviewer described. A `method_nid` reference dangling without
|
|
# any name_node guard must be caught.
|
|
bad_source = (
|
|
"def walk(node):\n"
|
|
" if t == 'method_declaration':\n"
|
|
" name_node = node.child_by_field_name('name')\n"
|
|
" if name_node:\n"
|
|
" method_nid = make_id('x')\n"
|
|
" emit_go_method_refs(node, method_nid, 1)\n"
|
|
" return\n"
|
|
)
|
|
bad_tree = ast.parse(bad_source)
|
|
bad_branch = _find_branch(bad_tree, "method_declaration")
|
|
assert bad_branch is not None
|
|
assert not _guarded_by_name_node(bad_branch, "method_nid"), (
|
|
"checker should reject method_nid used without a name_node guard"
|
|
)
|
|
|
|
|
|
# ── Rust ──────────────────────────────────────────────────────────────────────
|
|
|
|
def test_rust_finds_struct():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
assert "error" not in r
|
|
assert "Graph" in _labels(r)
|
|
|
|
def test_rust_finds_impl_methods():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
labels = _labels(r)
|
|
assert any("add_node" in l for l in labels)
|
|
assert any("add_edge" in l for l in labels)
|
|
|
|
def test_rust_finds_function():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
assert any("build_graph" in l for l in _labels(r))
|
|
|
|
def test_rust_emits_calls():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
calls = _call_pairs(r)
|
|
assert any("build_graph" in src for src, _ in calls)
|
|
|
|
def test_rust_calls_are_extracted():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
for e in r["edges"]:
|
|
if e["relation"] == "calls":
|
|
assert e["confidence"] == "EXTRACTED"
|
|
|
|
|
|
def test_rust_finds_static_and_const_items():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
by_id = {n["id"]: n["label"] for n in r["nodes"]}
|
|
labels = set(by_id.values())
|
|
for name in ("RETRY_LIMIT", "DEFAULT_MODE", "NODE_LIMIT"):
|
|
assert name in labels, name
|
|
assert any(
|
|
e["relation"] == "contains" and by_id.get(e["target"]) == name
|
|
for e in r["edges"]
|
|
), f"{name} has no file-level contains edge"
|
|
# An associated const is attributed to its impl, like a method.
|
|
assert ".CAPACITY" in labels
|
|
assert any(
|
|
e["relation"] == "contains"
|
|
and by_id.get(e["source"]) == "Graph"
|
|
and by_id.get(e["target"]) == ".CAPACITY"
|
|
for e in r["edges"]
|
|
)
|
|
# The declared type is referenced like a struct field type.
|
|
assert any(
|
|
e["relation"] == "references"
|
|
and by_id.get(e["source"]) == "NODE_LIMIT"
|
|
and by_id.get(e["target"]) == "Limit"
|
|
for e in r["edges"]
|
|
)
|
|
|
|
def test_rust_import_edges_have_import_context():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
import_edges = _edges_with_relation(r, "imports", "imports_from")
|
|
assert import_edges
|
|
assert all(e.get("context") == "import" for e in import_edges)
|
|
|
|
|
|
def test_rust_call_edges_have_call_context():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
call_edges = _edges_with_relation(r, "calls")
|
|
assert call_edges
|
|
assert all(e.get("context") == "call" for e in call_edges)
|
|
|
|
def test_rust_no_dangling_edges():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
node_ids = {n["id"] for n in r["nodes"]}
|
|
for e in r["edges"]:
|
|
if e["relation"] in ("contains", "method", "calls"):
|
|
assert e["source"] in node_ids
|
|
|
|
|
|
def test_rust_trait_impl_emits_implements():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
assert ("DataProcessor", "Processor") in _edge_labels(r, "implements")
|
|
|
|
|
|
def test_rust_trait_method_declarations_are_captured():
|
|
# #3366: signature-only trait method declarations (and default-bodied ones)
|
|
# were dropped because trait bodies were never walked. Both `Processor.run`
|
|
# and `Logger.log` are declaration-only trait methods in the fixture.
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
method_pairs = _edge_labels(r, "method")
|
|
assert ("Processor", "run") in method_pairs
|
|
assert ("Logger", "log") in method_pairs
|
|
|
|
|
|
def test_rust_trait_decl_method_is_distinct_from_impl_method():
|
|
# #3366: the trait-declared `Processor.run` must be its own node, not collide
|
|
# onto the impl `DataProcessor.run` — the crux of the fix.
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
by_id = {n["id"]: n["label"] for n in r["nodes"]}
|
|
run_targets = {
|
|
e["target"]
|
|
for e in r["edges"]
|
|
if e["relation"] == "method"
|
|
and _normalize_symbol_label(by_id.get(e["target"], "")) == "run"
|
|
}
|
|
# one `run()` declared in `trait Processor`, one defined in the impl — two ids
|
|
assert len(run_targets) >= 2
|
|
|
|
|
|
def test_rust_supertrait_emits_inherits():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
assert ("Logger", "Processor") in _edge_labels(r, "inherits")
|
|
|
|
|
|
def test_rust_enum_variant_references():
|
|
"""Enum variant payload types must emit `references` edges.
|
|
|
|
Tuple variants (`Click(T)`) and struct variants (`Resize { x: T }`) nest
|
|
their field types under enum_variant_list -> enum_variant; that path was
|
|
never traversed, so every enum-variant type reference was dropped.
|
|
"""
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
refs = _edge_labels(r, "references")
|
|
assert ("GraphEvent", "Graph") in refs, "tuple-variant reference missing"
|
|
assert ("GraphEvent", "DataProcessor") in refs, "struct-variant reference missing"
|
|
|
|
|
|
def test_rust_enum_variants_emit_case_of_nodes():
|
|
"""Each enum variant must become a node with a `case_of` edge to its enum.
|
|
|
|
The enum handler only walked variants to collect their payload type
|
|
references; the variants themselves (`NodeAdded`, `Processed`) never became
|
|
nodes, so the enum was left a memberless leaf. Every other language with
|
|
enums (Java #1719, Kotlin #1738, Swift, Scala) emits a node per member with
|
|
a `case_of` edge; this brings Rust to parity.
|
|
"""
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
labels = {n["label"] for n in r["nodes"]}
|
|
assert "NodeAdded" in labels
|
|
assert "Processed" in labels
|
|
case_of = _edge_labels(r, "case_of")
|
|
assert ("GraphEvent", "NodeAdded") in case_of
|
|
assert ("GraphEvent", "Processed") in case_of
|
|
|
|
|
|
def test_rust_struct_field_emits_field_context():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
assert ("DataProcessor", "Result") in _edge_labels(r, "references", "field")
|
|
assert ("DataProcessor", "DataProcessor") not in _edge_labels(r, "references", "field")
|
|
|
|
|
|
def test_rust_tuple_struct_field_references():
|
|
"""Tuple struct fields (`struct Wrapper(A, B);`) nest their positional types
|
|
under ordered_field_declaration_list with no field_declaration wrapper -- the
|
|
same shape as tuple enum variants. That path was not traversed for structs, so
|
|
tuple-struct field type references were silently dropped.
|
|
"""
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
field_refs = _edge_labels(r, "references", "field")
|
|
assert ("GraphPair", "Graph") in field_refs, "tuple-struct field reference missing"
|
|
assert ("GraphPair", "Result") in field_refs, "tuple-struct generic field reference missing"
|
|
assert ("GraphPair", "DataProcessor") in _edge_labels(r, "references", "generic_arg")
|
|
|
|
|
|
def test_rust_method_parameter_return_and_generic_contexts():
|
|
r = extract_rust(FIXTURES / "sample.rs")
|
|
assert ("build", "DataProcessor") in _edge_labels(r, "references", "parameter_type")
|
|
assert ("build", "Result") in _edge_labels(r, "references", "return_type")
|
|
assert ("build", "DataProcessor") in _edge_labels(r, "references", "generic_arg")
|
|
|
|
|
|
def test_rust_no_cross_crate_spurious_edges():
|
|
"""Scoped calls (Type::method) and blocklisted names must not produce
|
|
INFERRED cross-crate calls edges (#908)."""
|
|
from graphify.extract import extract
|
|
crate_a = FIXTURES / "crate_a" / "src" / "lib.rs"
|
|
crate_b = FIXTURES / "crate_b" / "src" / "lib.rs"
|
|
r = extract([crate_a, crate_b])
|
|
node_ids_a = {n["id"] for n in r["nodes"] if "crate_a" in (n.get("source_file") or "")}
|
|
node_ids_b = {n["id"] for n in r["nodes"] if "crate_b" in (n.get("source_file") or "")}
|
|
# No calls edge should cross from crate_b into crate_a
|
|
cross_crate_calls = [
|
|
e for e in r["edges"]
|
|
if e["relation"] == "calls"
|
|
and e["source"] in node_ids_b
|
|
and e["target"] in node_ids_a
|
|
]
|
|
assert cross_crate_calls == [], (
|
|
f"Spurious cross-crate edges: {cross_crate_calls}"
|
|
)
|
|
|
|
|
|
def test_rust_forward_type_reference_resolves_to_local_declaration(tmp_path):
|
|
"""#3782: a struct, enum or trait used above its declaration in the same file
|
|
must resolve to that declaration, not to a sourceless stub."""
|
|
p = tmp_path / "engine.rs"
|
|
p.write_text(
|
|
"fn before(s: &Sink) -> Mode { Mode::A }\n"
|
|
"\n"
|
|
"struct Sink {\n"
|
|
" n: u32,\n"
|
|
"}\n"
|
|
"\n"
|
|
"impl Handler for Sink {}\n"
|
|
"\n"
|
|
"enum Mode {\n"
|
|
" A,\n"
|
|
"}\n"
|
|
"\n"
|
|
"trait Handler {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_rust(p)
|
|
assert [n["label"] for n in r["nodes"] if not n["source_file"]] == []
|
|
assert ("before", "Sink") in _edge_labels(r, "references", "parameter_type")
|
|
assert ("before", "Mode") in _edge_labels(r, "references", "return_type")
|
|
assert ("Sink", "Handler") in _edge_labels(r, "implements")
|
|
|
|
|
|
def test_rust_forward_reference_prescan_skips_items_that_are_not_nodes(tmp_path):
|
|
"""#3782: only struct/enum/trait declarations become nodes, so only they may
|
|
resolve a forward reference. A `type` alias, a `union`, a struct declared in a
|
|
function body, a name declared nowhere in the file and a generic type that only
|
|
an `impl` names produce no node of that name here; references to them must keep
|
|
their sourceless stub instead of being dropped."""
|
|
p = tmp_path / "shapes.rs"
|
|
p.write_text(
|
|
"fn user(a: &Alias, u: &Bits, l: &Local, e: &Elsewhere, w: &Wrapper<u8>) {}\n"
|
|
"\n"
|
|
"type Alias = u32;\n"
|
|
"\n"
|
|
"union Bits { a: u32, b: f32 }\n"
|
|
"\n"
|
|
"fn host() {\n"
|
|
" struct Local;\n"
|
|
"}\n"
|
|
"\n"
|
|
"trait Tr {}\n"
|
|
"\n"
|
|
"impl<T> Tr for Wrapper<T> {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_rust(p)
|
|
by_id = {n["id"]: n for n in r["nodes"]}
|
|
targets = {
|
|
by_id[e["target"]]["label"]: by_id[e["target"]]
|
|
for e in r["edges"]
|
|
if e["relation"] == "references" and e["target"] in by_id
|
|
}
|
|
for name in ("Alias", "Bits", "Local", "Elsewhere", "Wrapper"):
|
|
assert name in targets, f"reference to {name} was dropped"
|
|
assert targets[name]["source_file"] == "", name
|
|
|
|
|
|
def test_rust_forward_reference_resolves_into_an_inline_module(tmp_path):
|
|
"""#3782: the pre-scan reaches the same items walk() does, so a type declared
|
|
in an inline `mod` below its use resolves like a use placed after it."""
|
|
p = tmp_path / "nested.rs"
|
|
p.write_text(
|
|
"use inner::Sink;\n"
|
|
"\n"
|
|
"fn before(s: &Sink) {}\n"
|
|
"\n"
|
|
"mod inner {\n"
|
|
" pub struct Sink;\n"
|
|
"}\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_rust(p)
|
|
sink = next(n for n in r["nodes"] if n["label"] == "Sink")
|
|
before = next(n for n in r["nodes"] if n["label"] == "before()")
|
|
assert sink["source_file"] == str(p)
|
|
assert any(
|
|
e["relation"] == "references" and e["source"] == before["id"] and e["target"] == sink["id"]
|
|
for e in r["edges"]
|
|
)
|
|
|
|
|
|
def test_rust_forward_reference_skips_items_in_const_and_static_initializers(tmp_path):
|
|
"""#3782: walk() never enters a const or static initializer, so a struct
|
|
declared there is no node; a reference to it keeps its sourceless stub."""
|
|
p = tmp_path / "initializers.rs"
|
|
p.write_text(
|
|
"fn before(a: &Hidden, b: &Guarded) {}\n"
|
|
"\n"
|
|
"const _: () = {\n"
|
|
" struct Hidden;\n"
|
|
"};\n"
|
|
"\n"
|
|
"static GUARD: () = {\n"
|
|
" struct Guarded;\n"
|
|
"};\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_rust(p)
|
|
by_id = {n["id"]: n for n in r["nodes"]}
|
|
targets = {
|
|
by_id[e["target"]]["label"]: by_id[e["target"]]["source_file"]
|
|
for e in r["edges"]
|
|
if e["relation"] == "references" and e["target"] in by_id
|
|
}
|
|
assert targets == {"Hidden": "", "Guarded": ""}
|
|
|
|
|
|
def test_rust_forward_reference_to_a_casefold_colliding_type_resolves_to_it(tmp_path):
|
|
"""#3782: ids are casefolded, so `fn handle` (also in an `extern` block),
|
|
`const HANDLE`, `struct HANDLE` or an `impl` for `HANDLE` fold to the id of
|
|
`struct Handle`. Those items now get ids of their own (#3302's rule), so the
|
|
struct owns its id and a reference above the declarations resolves to it, as
|
|
a reference below already did, instead of to a sourceless stub."""
|
|
extern_fn = 'extern "C" {\n fn handle();\n}'
|
|
impl_for_alias = "type HANDLE = usize;\n\ntrait Tr {}\n\nimpl Tr for HANDLE {}"
|
|
for name, other in (
|
|
("fn_collision.rs", "fn handle() {}"),
|
|
("extern_fn_collision.rs", extern_fn),
|
|
("const_collision.rs", "const HANDLE: u32 = 0;"),
|
|
("static_collision.rs", "static HANDLE: u32 = 0;"),
|
|
("type_collision.rs", "struct HANDLE;"),
|
|
("impl_collision.rs", impl_for_alias),
|
|
):
|
|
p = tmp_path / name
|
|
p.write_text(
|
|
f"fn user(h: &Handle) {{}}\n\n{other}\n\nstruct Handle {{\n n: u32,\n}}\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_rust(p)
|
|
by_id = {n["id"]: n for n in r["nodes"]}
|
|
handle = next(n for n in r["nodes"] if n["label"] == "Handle")
|
|
targets = [
|
|
e["target"] for e in r["edges"]
|
|
if e["relation"] == "references" and e["target"] in by_id
|
|
]
|
|
assert targets == [handle["id"]], name
|
|
assert handle["source_file"] == str(p), name
|
|
|
|
|
|
def test_rust_forward_reference_keeps_its_stub_when_a_non_type_shares_the_name(tmp_path):
|
|
"""#3782: a `type HANDLE` alias is not a node, and `fn Handle()` beside
|
|
`struct Handle {}` has the type's exact name, so the two still share one node
|
|
that walk() gives to whichever comes first. A reference above the
|
|
declarations keeps the sourceless stub v8's extractor gives it."""
|
|
for name, user, other, label in (
|
|
("alias_collision.rs", "fn user(h: HANDLE) {}", "type HANDLE = usize;", "HANDLE"),
|
|
("same_name_fn.rs", "fn user(h: &Handle) {}", "fn Handle() {}", "Handle"),
|
|
):
|
|
p = tmp_path / name
|
|
p.write_text(
|
|
f"{user}\n\n{other}\n\nstruct Handle {{\n n: u32,\n}}\n", encoding="utf-8"
|
|
)
|
|
r = extract_rust(p)
|
|
by_id = {n["id"]: n for n in r["nodes"]}
|
|
targets = [
|
|
(by_id[e["target"]]["label"], by_id[e["target"]]["source_file"])
|
|
for e in r["edges"]
|
|
if e["relation"] == "references" and e["target"] in by_id
|
|
]
|
|
assert targets == [(label, "")], name
|
|
|
|
|
|
def test_rust_forward_reference_resolves_when_no_other_item_shares_the_id(tmp_path):
|
|
"""#3782: the collision rule only withholds a type whose id another item
|
|
shares. A function with another name, `#[cfg]` twins of the type itself (one
|
|
node in walk()), an `impl` for the type itself or a method named like the type
|
|
(its id is qualified by the impl) still let the reference above resolve to it."""
|
|
impl_for_type = (
|
|
"struct Handle<'a> {\n s: &'a str,\n}\n\ntrait Tr {}\n\n"
|
|
"impl Tr for Handle<'_> {}\n\nimpl Tr for &Handle<'_> {}\n"
|
|
)
|
|
for name, source in (
|
|
("no_collision.rs", "struct Handle {\n n: u32,\n}\n\nfn other() {}\n"),
|
|
("cfg_twins.rs", "#[cfg(unix)]\nstruct Handle;\n\n#[cfg(not(unix))]\nstruct Handle;\n"),
|
|
("impl_for_type.rs", impl_for_type),
|
|
("method_like_type.rs", "struct Handle;\n\nimpl Handle {\n fn handle(&self) {}\n}\n"),
|
|
):
|
|
p = tmp_path / name
|
|
p.write_text("fn user(h: &Handle) {}\n\n" + source, encoding="utf-8")
|
|
r = extract_rust(p)
|
|
handle = next(n for n in r["nodes"] if n["label"] == "Handle")
|
|
user = next(n for n in r["nodes"] if n["label"] == "user()")
|
|
assert handle["source_file"] == str(p), name
|
|
assert any(
|
|
e["relation"] == "references"
|
|
and e["source"] == user["id"]
|
|
and e["target"] == handle["id"]
|
|
for e in r["edges"]
|
|
), name
|
|
|
|
|
|
def test_rust_forward_reference_to_ambiguous_type_name_stays_local(tmp_path):
|
|
"""#3782: when two files declare the same struct, the corpus-level rewire
|
|
declines the ambiguous stub, so the reference above the declaration must
|
|
already point at the local struct. Checked cold, then from the warm cache."""
|
|
(tmp_path / "engine_a.rs").write_text(
|
|
"fn before(s: &Sink) {}\n\nstruct Sink {\n n: u32,\n}\n\nfn after(s: &Sink) {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "engine_b.rs").write_text(
|
|
"struct Sink {\n n: u32,\n}\n", encoding="utf-8",
|
|
)
|
|
files = sorted(tmp_path.glob("*.rs"))
|
|
for _ in range(2):
|
|
r = extract(files, cache_root=tmp_path, parallel=False)
|
|
sinks = [n for n in r["nodes"] if n["label"] == "Sink"]
|
|
assert all(n["source_file"] for n in sinks), "sourceless Sink stub left behind"
|
|
local = next(n["id"] for n in sinks if Path(n["source_file"]).name == "engine_a.rs")
|
|
before = next(n["id"] for n in r["nodes"] if n["label"] == "before()")
|
|
after = next(n["id"] for n in r["nodes"] if n["label"] == "after()")
|
|
refs = {(e["source"], e["target"]) for e in r["edges"] if e["relation"] == "references"}
|
|
assert (before, local) in refs
|
|
assert (after, local) in refs
|
|
|
|
|
|
def test_rust_cross_file_reference_skips_impl_blocks_in_other_files(tmp_path):
|
|
"""#4283: an `impl Thing` block in another file adds a second `Thing` node,
|
|
but only the struct is a declaration. A cross-file reference must bind to
|
|
that declaration instead of being left on a sourceless stub. Checked cold,
|
|
then from the warm cache."""
|
|
(tmp_path / "a.rs").write_text("pub struct Thing {\n pub n: u32,\n}\n", encoding="utf-8")
|
|
(tmp_path / "b.rs").write_text(
|
|
"use crate::a::Thing;\n\nimpl Thing {\n pub fn new() -> Self {\n"
|
|
" Thing { n: 0 }\n }\n}\n",
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "c.rs").write_text(
|
|
"use crate::a::Thing;\n\npub fn make(t: Thing) -> u32 {\n t.n\n}\n",
|
|
encoding="utf-8",
|
|
)
|
|
files = sorted(tmp_path.glob("*.rs"))
|
|
for _ in range(2):
|
|
r = extract(files, cache_root=tmp_path, parallel=False)
|
|
things = [n for n in r["nodes"] if n["label"] == "Thing"]
|
|
assert all(n["source_file"] for n in things), "sourceless Thing stub left behind"
|
|
declared = next(n["id"] for n in things if Path(n["source_file"]).name == "a.rs")
|
|
make = next(n["id"] for n in r["nodes"] if n["label"] == "make()")
|
|
refs = {(e["source"], e["target"]) for e in r["edges"] if e["relation"] == "references"}
|
|
assert (make, declared) in refs
|
|
|
|
|
|
def test_rust_cross_file_reference_to_twice_declared_type_keeps_its_stub(tmp_path):
|
|
"""#4283: impl blocks are only skipped in favour of a SINGLE declaration. With
|
|
two real `struct Thing` declarations a cross-file reference has no one
|
|
correct target, so it must stay on its sourceless stub (fail closed)."""
|
|
(tmp_path / "a.rs").write_text("pub struct Thing;\n", encoding="utf-8")
|
|
(tmp_path / "b.rs").write_text(
|
|
"pub struct Thing;\n\nimpl Thing {\n pub fn new() -> Self {\n Thing\n }\n}\n",
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "c.rs").write_text("pub fn make(t: Thing) {}\n", encoding="utf-8")
|
|
r = extract(sorted(tmp_path.glob("*.rs")), cache_root=tmp_path, parallel=False)
|
|
by_id = {n["id"]: n for n in r["nodes"]}
|
|
make = next(n["id"] for n in r["nodes"] if n["label"] == "make()")
|
|
targets = [
|
|
by_id[e["target"]] for e in r["edges"]
|
|
if e["relation"] == "references" and e["source"] == make
|
|
]
|
|
assert [t["source_file"] for t in targets if t["label"] == "Thing"] == [""]
|
|
|
|
|
|
def _thing_targets_of_make(tmp_path, files):
|
|
for name, text in files.items():
|
|
(tmp_path / name).write_text(text, encoding="utf-8")
|
|
r = extract(sorted(tmp_path.iterdir()), cache_root=tmp_path, parallel=False)
|
|
by_id = {n["id"]: n for n in r["nodes"]}
|
|
make = next(n["id"] for n in r["nodes"] if n["label"] == "make()")
|
|
return [
|
|
by_id[e["target"]]["source_file"] for e in r["edges"]
|
|
if e["relation"] == "references" and e["source"] == make
|
|
and by_id[e["target"]]["label"] == "Thing"
|
|
]
|
|
|
|
|
|
def test_rust_declaration_tiebreak_keeps_stub_beside_another_language(tmp_path):
|
|
"""#4283: `_rust_declaration_count` only tells a Rust declaration from a Rust
|
|
impl block. A same-named TypeScript class is a declaration too, so a Rust
|
|
declaration plus a TS class is still ambiguous and the stub stays."""
|
|
targets = _thing_targets_of_make(tmp_path, {
|
|
"a.rs": "pub struct Thing;\n",
|
|
"b.ts": "export class Thing {}\n",
|
|
"c.rs": "pub fn make(t: Thing) {}\n",
|
|
})
|
|
assert targets == [""]
|
|
|
|
|
|
def test_rust_declaration_tiebreak_keeps_stub_for_declarations_folded_in_one_node(tmp_path):
|
|
"""#4283: two inline modules declaring `Thing` in one file fold into a single
|
|
node with `_rust_declaration_count == 2`. That is two declarations, so an
|
|
impl block elsewhere must not let the reference bind to the folded node."""
|
|
targets = _thing_targets_of_make(tmp_path, {
|
|
"a.rs": "mod x {\n pub struct Thing;\n}\nmod y {\n pub struct Thing;\n}\n",
|
|
"b.rs": "impl Thing {\n pub fn new() {}\n}\n",
|
|
"c.rs": "pub fn make(t: Thing) {}\n",
|
|
})
|
|
assert targets == [""]
|
|
|
|
|
|
# ── extract() dispatch ────────────────────────────────────────────────────────
|
|
|
|
def test_extract_dispatches_all_languages():
|
|
files = [
|
|
FIXTURES / "sample.py",
|
|
FIXTURES / "sample.ts",
|
|
FIXTURES / "sample.go",
|
|
FIXTURES / "sample.rs",
|
|
]
|
|
r = extract(files)
|
|
source_files = {n["source_file"] for n in r["nodes"] if n["source_file"]}
|
|
# All four files should contribute nodes
|
|
assert any("sample.py" in f for f in source_files)
|
|
assert any("sample.ts" in f for f in source_files)
|
|
assert any("sample.go" in f for f in source_files)
|
|
assert any("sample.rs" in f for f in source_files)
|
|
|
|
|
|
# ── Cache ─────────────────────────────────────────────────────────────────────
|
|
|
|
def test_cache_hit_returns_same_result(tmp_path):
|
|
src = FIXTURES / "sample.py"
|
|
dst = tmp_path / "sample.py"
|
|
dst.write_bytes(src.read_bytes())
|
|
|
|
r1 = extract([dst])
|
|
r2 = extract([dst])
|
|
assert len(r1["nodes"]) == len(r2["nodes"])
|
|
assert len(r1["edges"]) == len(r2["edges"])
|
|
|
|
def test_cache_miss_after_file_change(tmp_path):
|
|
dst = tmp_path / "a.py"
|
|
dst.write_text("def foo(): pass\n")
|
|
r1 = extract([dst])
|
|
|
|
dst.write_text("def foo(): pass\ndef bar(): pass\n")
|
|
r2 = extract([dst])
|
|
# bar() should appear in the second result
|
|
labels2 = [n["label"] for n in r2["nodes"]]
|
|
assert any("bar" in l for l in labels2)
|
|
|
|
|
|
# ── 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_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_create_table_inside_transaction_block():
|
|
"""#2953: DDL wrapped in BEGIN; ... COMMIT; must emit table nodes."""
|
|
r = _extract_sql_or_skip("sample_transaction.sql")
|
|
labels = [n["label"] for n in r["nodes"]]
|
|
assert any("alfa" in l for l in labels)
|
|
assert any("gamma" in l for l in labels)
|
|
assert any("delta" in l for l in labels)
|
|
refs = [
|
|
(e["source"], e["target"])
|
|
for e in r["edges"]
|
|
if e["relation"] == "references"
|
|
]
|
|
node_by_id = {n["id"]: n["label"] for n in r["nodes"]}
|
|
assert any(
|
|
"delta" in node_by_id.get(s, "") and "alfa" in node_by_id.get(t, "")
|
|
for s, t in refs
|
|
)
|
|
|
|
|
|
def test_sql_create_table_before_do_block_in_transaction(tmp_path):
|
|
"""#3886: with a bare `BEGIN;`, a later `DO $$ ... END $$;` makes the parser
|
|
read the transaction as a block ending at the DO body's END. Tables inside
|
|
that block were dropped because the top-level loop skipped block nodes.
|
|
"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"BEGIN;\n"
|
|
"\n"
|
|
"CREATE TABLE users (\n"
|
|
" id BIGSERIAL PRIMARY KEY\n"
|
|
");\n"
|
|
"\n"
|
|
"DO $$\n"
|
|
"BEGIN\n"
|
|
" IF NOT EXISTS (SELECT 1 FROM pg_trigger WHERE tgname = 't') THEN\n"
|
|
" CREATE TRIGGER t BEFORE UPDATE ON users\n"
|
|
" FOR EACH ROW EXECUTE FUNCTION touch();\n"
|
|
" END IF;\n"
|
|
"END $$;\n"
|
|
"\n"
|
|
"COMMIT;\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_sql(p)
|
|
labels = [n["label"] for n in r["nodes"]]
|
|
assert "users" in labels
|
|
|
|
|
|
def test_sql_finds_view():
|
|
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_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_or_skip()
|
|
relations = {e["relation"] for e in r["edges"]}
|
|
assert "references" in relations
|
|
|
|
def test_sql_emits_reads_from_edge():
|
|
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_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_create_index_emits_index_node_linked_to_its_table(tmp_path):
|
|
"""#3467: CREATE [UNIQUE] INDEX parsed fine but the walk never dispatched
|
|
on create_index, so every index was silently dropped."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"CREATE TABLE public.profiles (id uuid PRIMARY KEY, owner uuid NOT NULL);\n"
|
|
"CREATE INDEX profiles_owner_idx ON public.profiles (owner);\n"
|
|
"CREATE UNIQUE INDEX IF NOT EXISTS profiles_id_uniq ON public.profiles (id);\n"
|
|
"CREATE INDEX CONCURRENTLY orders_customer_idx ON public.orders (customer_id);\n"
|
|
'CREATE INDEX "quoted idx" ON public.profiles (owner);\n'
|
|
"CREATE INDEX ON public.profiles (owner);\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_sql(p)
|
|
by_label = {n["label"]: n for n in r["nodes"]}
|
|
for name in ("profiles_owner_idx", "profiles_id_uniq", "orders_customer_idx", "quoted idx"):
|
|
assert name in by_label, name
|
|
assert by_label[name]["source_file"] == str(p)
|
|
edges = {(e["source"], e["relation"], e["target"]) for e in r["edges"]}
|
|
profiles = by_label["public.profiles"]["id"]
|
|
assert (by_label["profiles_owner_idx"]["id"], "indexes", profiles) in edges
|
|
assert (by_label["profiles_id_uniq"]["id"], "indexes", profiles) in edges
|
|
assert (by_label["quoted idx"]["id"], "indexes", profiles) in edges
|
|
# An index on a table defined in another file links to a sourceless stub,
|
|
# the same way a trigger does (#2324).
|
|
orders = by_label["public.orders"]
|
|
assert orders["source_file"] == ""
|
|
assert (by_label["orders_customer_idx"]["id"], "indexes", orders["id"]) in edges
|
|
# The unnamed index is skipped and nothing dangles.
|
|
node_ids = {n["id"] for n in r["nodes"]}
|
|
assert all(e["source"] in node_ids and e["target"] in node_ids for e in r["edges"])
|
|
assert sum(1 for e in r["edges"] if e["relation"] == "indexes") == 4
|
|
|
|
|
|
def test_sql_clean_trigger_links_to_its_on_table(tmp_path):
|
|
"""A trigger's subject table follows ON, not FOR.
|
|
|
|
The create_trigger branch read the table off keyword_for, but `FOR EACH ROW`
|
|
carries no table — so a cleanly-parsed trigger got a node with no link to the
|
|
table it fires on.
|
|
"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"CREATE TABLE users (id INT);\n"
|
|
"CREATE TRIGGER audit_ins AFTER INSERT ON users EXECUTE FUNCTION log_it();\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_sql(p)
|
|
by_label = {n["label"]: n for n in r["nodes"]}
|
|
assert "audit_ins" in by_label
|
|
edges = {(e["source"], e["relation"], e["target"]) for e in r["edges"]}
|
|
assert (by_label["audit_ins"]["id"], "triggers", by_label["users"]["id"]) in edges
|
|
|
|
|
|
def test_sql_trigger_links_to_the_function_it_executes(tmp_path):
|
|
"""A trigger exists to run a function, yet the EXECUTE FUNCTION/PROCEDURE
|
|
target was never linked. Only the ON table got an edge, so the trigger's
|
|
whole reason to exist — the routine it fires — was dropped from the graph.
|
|
"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"CREATE TABLE users (id INT);\n"
|
|
"CREATE FUNCTION log_it() RETURNS TRIGGER AS 'SELECT 1';\n"
|
|
"CREATE TRIGGER audit_ins AFTER INSERT ON users\n"
|
|
" FOR EACH ROW EXECUTE FUNCTION log_it();\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_sql(p)
|
|
by_label = {n["label"]: n for n in r["nodes"]}
|
|
assert "audit_ins" in by_label and "log_it()" in by_label
|
|
edges = {(e["source"], e["relation"], e["target"]) for e in r["edges"]}
|
|
# The trigger fires on its ON table ...
|
|
assert (by_label["audit_ins"]["id"], "triggers", by_label["users"]["id"]) in edges
|
|
# ... and executes the function it names.
|
|
assert (by_label["audit_ins"]["id"], "executes", by_label["log_it()"]["id"]) in edges
|
|
# Nothing dangles.
|
|
node_ids = {n["id"] for n in r["nodes"]}
|
|
assert all(e["source"] in node_ids and e["target"] in node_ids for e in r["edges"])
|
|
|
|
|
|
def test_sql_trigger_executing_an_undefined_function_fails_closed(tmp_path):
|
|
"""When the executed routine is not defined in this file, link nothing
|
|
rather than mint a bare-name stub that could never rewire onto the real
|
|
`name()` function node. The ON-table edge is unaffected.
|
|
"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"CREATE TABLE users (id INT);\n"
|
|
"CREATE TRIGGER audit_ins AFTER INSERT ON users EXECUTE FUNCTION elsewhere();\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_sql(p)
|
|
by_label = {n["label"]: n for n in r["nodes"]}
|
|
assert (by_label["audit_ins"]["id"], "triggers", by_label["users"]["id"]) in {
|
|
(e["source"], e["relation"], e["target"]) for e in r["edges"]
|
|
}
|
|
# No executes edge, and the unresolved routine was not minted as a node.
|
|
assert not any(e["relation"] == "executes" for e in r["edges"])
|
|
assert "elsewhere()" not in by_label
|
|
node_ids = {n["id"] for n in r["nodes"]}
|
|
assert all(e["source"] in node_ids and e["target"] in node_ids for e in r["edges"])
|
|
|
|
|
|
def test_sql_procedural_body_trigger_is_recovered(tmp_path):
|
|
"""A trigger with a `FOR EACH ROW BEGIN ... END` body has no grammar parse,
|
|
so the statement lands in ERROR recovery. TRIGGER was excluded from the
|
|
routine-recovery pattern, so the whole trigger — and its table — was dropped.
|
|
"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"CREATE TABLE users (id INT);\n"
|
|
"CREATE TABLE stats (cnt INT);\n"
|
|
"CREATE TRIGGER trg_after AFTER INSERT ON users\n"
|
|
"FOR EACH ROW\n"
|
|
"BEGIN\n"
|
|
" UPDATE stats SET cnt = cnt + 1;\n"
|
|
"END;\n",
|
|
encoding="utf-8",
|
|
)
|
|
r = extract_sql(p)
|
|
by_label = {n["label"]: n for n in r["nodes"]}
|
|
assert "trg_after" in by_label, "procedural-body trigger dropped"
|
|
# a trigger is not callable — its label carries no ()
|
|
assert by_label["trg_after"]["label"] == "trg_after"
|
|
edges = {(e["source"], e["relation"], e["target"]) for e in r["edges"]}
|
|
assert (by_label["trg_after"]["id"], "triggers", by_label["users"]["id"]) in edges
|
|
# exactly one triggers edge, and nothing dangles
|
|
assert sum(1 for e in r["edges"] if e["relation"] == "triggers") == 1
|
|
node_ids = {n["id"] for n in r["nodes"]}
|
|
assert all(e["source"] in node_ids and e["target"] in node_ids for e in r["edges"])
|
|
|
|
|
|
def test_sql_tsql_bracketed_procedure_is_recovered(tmp_path):
|
|
"""T-SQL CREATE PROCEDURE [Schema].[Name] ... AS BEGIN...END.
|
|
|
|
The grammar has no create_procedure parse for T-SQL's AS BEGIN...END body
|
|
idiom, so the statement lands in ERROR recovery — where a name pattern
|
|
without a bracket-delimited alternative recovered nothing. In a T-SQL
|
|
codebase that brackets every identifier (a common house standard), that
|
|
made every stored procedure invisible.
|
|
"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "proc.sql"
|
|
p.write_text(
|
|
"CREATE PROCEDURE [dbo].[usp_LoadDebtors]\n"
|
|
" @BatchId INT\n"
|
|
"AS\n"
|
|
"BEGIN\n"
|
|
" SET NOCOUNT ON;\n"
|
|
" INSERT INTO dbo.Debtors (Id) SELECT Id FROM staging.Debtors;\n"
|
|
"END;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
routine = [n["label"] for n in r["nodes"] if n["label"] != "proc.sql"]
|
|
assert routine == ["[dbo].[usp_LoadDebtors]()"], routine
|
|
|
|
|
|
def test_sql_tsql_create_or_alter_procedure_is_recovered(tmp_path):
|
|
"""T-SQL spells idempotent re-creation CREATE OR ALTER (no OR REPLACE)."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "proc.sql"
|
|
p.write_text(
|
|
"CREATE OR ALTER PROCEDURE [Utils].[ValidateSourceView]\n"
|
|
"AS\nBEGIN\n SELECT 1;\nEND;\n"
|
|
"CREATE OR ALTER PROCEDURE usp_Bare AS\nBEGIN\n SELECT 1;\nEND;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
labels = sorted(n["label"] for n in r["nodes"] if n["label"] != "proc.sql")
|
|
assert labels == ["[Utils].[ValidateSourceView]()", "usp_Bare()"], labels
|
|
|
|
|
|
def test_sql_escaped_closing_bracket_in_routine_name_is_consumed(tmp_path):
|
|
"""T-SQL escapes a literal ] inside a bracketed identifier by doubling it:
|
|
[a]]b] names the identifier a]b. A pattern that stops at the first ]
|
|
truncated the name to [dbo].[a] — a phantom that could collide with a
|
|
genuinely-named [dbo].[a]."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "proc.sql"
|
|
p.write_text("CREATE PROCEDURE [dbo].[a]]b]\nAS\nBEGIN\n SELECT 1;\nEND;\n")
|
|
r = extract_sql(p)
|
|
routine = [n["label"] for n in r["nodes"] if n["label"] != "proc.sql"]
|
|
assert routine == ["[dbo].[a]]b]()"], routine
|
|
|
|
|
|
def test_sql_recovery_sites_agree_on_the_captured_name(tmp_path):
|
|
"""A mixed-delimiter name (dbo.[usp_Mixed]) must yield exactly ONE node.
|
|
|
|
The walk-time ERROR scan and the whole-file has_error fallback recover the
|
|
same statement; when their patterns drifted, they captured different names
|
|
(`dbo.[usp_Mixed]` vs `dbo`), minted different ids, and _add_node's
|
|
id-dedupe never fired — one procedure became two nodes, one of them a
|
|
phantom named after the schema. Sharing _ROUTINE_RECOVERY_RX makes the
|
|
drift impossible; this pins the observable behaviour.
|
|
"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "proc.sql"
|
|
p.write_text("CREATE PROCEDURE dbo.[usp_Mixed] AS\nBEGIN\n SELECT 1;\nEND;\n")
|
|
r = extract_sql(p)
|
|
routine = [n["label"] for n in r["nodes"] if n["label"] != "proc.sql"]
|
|
assert routine == ["dbo.[usp_Mixed]()"], routine
|
|
|
|
|
|
def test_sql_tsql_proc_shorthand_is_recovered(tmp_path):
|
|
"""T-SQL's official PROC shorthand, including CREATE OR ALTER PROC with a
|
|
qualified bracket-delimited name, must recover like the long form."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "proc.sql"
|
|
p.write_text(
|
|
"CREATE OR ALTER PROC [dbo].[usp_Short]\n"
|
|
"AS\nBEGIN\n SELECT 1;\nEND;\n"
|
|
"CREATE PROC usp_BareShort AS\nBEGIN\n SELECT 1;\nEND;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
labels = sorted(n["label"] for n in r["nodes"] if n["label"] != "proc.sql")
|
|
assert labels == ["[dbo].[usp_Short]()", "usp_BareShort()"], labels
|
|
|
|
|
|
def test_sql_commented_ddl_is_not_fabricated_by_error_recovery(tmp_path):
|
|
"""An unrelated syntax error arms the whole-file recovery scan; DDL that
|
|
exists only inside comments must not fabricate routine nodes from it."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "broken.sql"
|
|
p.write_text(
|
|
"-- CREATE PROCEDURE [dbo].[usp_LineComment] AS BEGIN SELECT 1; END;\n"
|
|
"/*\nCREATE OR ALTER PROC dbo.usp_BlockComment AS\nBEGIN SELECT 1; END;\n*/\n"
|
|
"CREATE PROCEDURE [dbo].[usp_Real]\nAS\nBEGIN\n SELECT 1;\nEND;\n"
|
|
"THIS IS NOT SQL AT ALL %%%;\n"
|
|
# Sandwiched between broken segments so one ERROR blob's byte span
|
|
# covers the comment — the shape that made a per-ERROR-node scan
|
|
# unsound (a fragment can lack the comment opener entirely) and is
|
|
# why recovery scans only the whole masked file:
|
|
"%%% BROKEN JUNK\n"
|
|
"-- CREATE PROC [dbo].[usp_ErrComment] AS BEGIN SELECT 1; END;\n"
|
|
"%%% MORE BROKEN JUNK\n"
|
|
)
|
|
r = extract_sql(p)
|
|
labels = [n["label"] for n in r["nodes"] if n["label"] != "broken.sql"]
|
|
assert "[dbo].[usp_Real]()" in labels, labels
|
|
assert not any("Comment" in l for l in labels), (
|
|
f"commented-out DDL fabricated a node: {labels}"
|
|
)
|
|
|
|
|
|
def test_sql_comment_openers_inside_string_literals_do_not_hide_ddl(tmp_path):
|
|
"""A `--` or `/*` inside a string literal is data, not a comment opener.
|
|
|
|
Both recovery scans mask comments before matching, and a mask that reads
|
|
`'-- note'` as a line comment blanks to end-of-line — hiding a routine
|
|
declared after the literal — while a `/*` inside a string blanks
|
|
everything through the next real `*/`, swallowing whole statements
|
|
between. Both must recover normally. This is behavior coverage: the
|
|
grammar's own fb_proc_or_trigger recovery can rescue these shapes even
|
|
under a literal-blind mask, so the mutation-killing pin for the mask
|
|
itself is test_mask_sql_comments_preserves_literals_and_blanks_comments.
|
|
"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "strings.sql"
|
|
# The lines carrying the literals start with junk so they cannot parse as
|
|
# proper statements: literal and routine land in the SAME ERROR blob, and
|
|
# only the mask decides whether the routine survives. (A literal in a
|
|
# statement that parses never reaches the masked scans at all.)
|
|
p.write_text(
|
|
"%%% INSERT INTO log VALUES ('-- note'); "
|
|
"CREATE PROCEDURE [dbo].[usp_AfterLineString] AS BEGIN SELECT 1; END;\n"
|
|
"%%% SELECT 'open /* here' AS x; "
|
|
"CREATE PROCEDURE [dbo].[usp_AfterBlockString] AS BEGIN SELECT 1; END;\n"
|
|
"/* a real comment, giving the false opener a closer to reach */\n"
|
|
)
|
|
r = extract_sql(p)
|
|
labels = [n["label"] for n in r["nodes"] if n["label"] != "strings.sql"]
|
|
assert "[dbo].[usp_AfterLineString]()" in labels, labels
|
|
assert "[dbo].[usp_AfterBlockString]()" in labels, labels
|
|
|
|
|
|
def test_mask_sql_comments_literal_and_comment_handling():
|
|
"""Unit pin for _mask_sql_comments: comment openers inside literals are
|
|
data, not comment starts; single-quoted strings are blanked (dynamic SQL
|
|
must not be recoverable); double-quoted and bracket identifiers are
|
|
preserved verbatim (they carry recoverable names); comments blank to
|
|
spaces with newlines and offsets preserved, an unclosed block comment
|
|
running to end-of-file."""
|
|
from graphify.extractors.sql import _mask_sql_comments as mask
|
|
|
|
# a comment opener inside a literal never blanks past the literal
|
|
assert mask("select '-- x' from t") == "select from t"
|
|
assert mask("select 'a /* b' as x, 1") == "select as x, 1"
|
|
# '' escape is consumed as one literal (no blanking past the string)
|
|
src = "select 'it''s -- ok', 1"
|
|
expected = "select " + " " * len("'it''s -- ok'") + ", 1"
|
|
assert mask(src) == expected and len(mask(src)) == len(src)
|
|
# identifier literals are preserved verbatim, escapes included
|
|
assert mask('CREATE FUNCTION "public"."a""b"()') == 'CREATE FUNCTION "public"."a""b"()'
|
|
assert mask("CREATE PROCEDURE [dbo].[a]]b] AS x") == "CREATE PROCEDURE [dbo].[a]]b] AS x"
|
|
# ...but a span that is unterminated or would swallow a comment opener is
|
|
# NOT trusted as an identifier. A distrusted span is irreducibly
|
|
# ambiguous, and any single reading exposed text another reading blanks
|
|
# (re-emitting the delimiter and rescanning even re-paired later single
|
|
# quotes, uncovering dynamic SQL), so the mask blanks the UNION of every
|
|
# reading: the rest of the line, plus carried comment state for a raw /*
|
|
# left open on it. The trade: a genuine [a--b] identifier loses its line,
|
|
# never fabricating anything.
|
|
ghost = "SELECT [Col FROM t -- CREATE PROC [dbo].[usp_Ghost] AS BEGIN SELECT 1; END;"
|
|
assert "CREATE" not in mask(ghost) and len(mask(ghost)) == len(ghost)
|
|
ghost2 = 'SELECT "Col FROM t /* CREATE PROC dbo.usp_Ghost AS BEGIN SELECT 1; END */'
|
|
assert "CREATE" not in mask(ghost2) and len(mask(ghost2)) == len(ghost2)
|
|
out = mask("select [a--b] from t")
|
|
assert out == "select " + " " * len("[a--b] from t") and len(out) == len("select [a--b] from t")
|
|
# a distrusted span cannot re-pair later single quotes: the dynamic SQL
|
|
# after it stays masked (the re-emit-and-rescan defect, found by fuzzing)
|
|
repair = "SELECT [Col's -- x] ; EXEC(N'CREATE PROC [dbo].[usp_Fake] AS BEGIN SELECT 1; END');"
|
|
assert "CREATE" not in mask(repair) and len(mask(repair)) == len(repair)
|
|
# a raw /* on a distrusted line with no */ carries across the newline —
|
|
# some reading of the broken line left it open
|
|
carry = "SELECT [a--b] /*\nCREATE PROC dbo.Fake AS BEGIN SELECT 1; END\n"
|
|
assert "CREATE" not in mask(carry) and len(mask(carry)) == len(carry)
|
|
# where a carry closes MID-line, the remainder of that line is blanked
|
|
# too (under the no-carry reading the whole line may be comment or
|
|
# string — emitting the post-*/ tail verbatim was an exposure, found by
|
|
# differential fuzzing); recovery resumes on the NEXT line
|
|
carry_closed = "SELECT [a--b] /*\nx */ CREATE PROC dbo.Lost AS x\nCREATE PROC dbo.Kept AS x\n"
|
|
out = mask(carry_closed)
|
|
assert "dbo.Lost" not in out and len(out) == len(carry_closed)
|
|
assert "CREATE PROC dbo.Kept" in out
|
|
# and a fresh /* after the close carries again
|
|
recarry = "SELECT [a--b] /*\nx */ y /*\nCREATE PROC dbo.Ghost3 AS x\n"
|
|
assert "Ghost3" not in mask(recarry)
|
|
# dynamic SQL contents cannot survive the mask
|
|
dyn = "EXEC(N'CREATE PROC [dbo].[Fake] AS BEGIN SELECT 1; END');"
|
|
assert "CREATE" not in mask(dyn) and len(mask(dyn)) == len(dyn)
|
|
# comments blank to spaces, newlines kept, offsets stable
|
|
assert mask("a -- b") == "a "
|
|
assert mask("x /* y */ z") == "x z"
|
|
masked = mask("a /* m\nl */ b")
|
|
assert masked == "a \n b" and len(masked) == len("a /* m\nl */ b")
|
|
# an unclosed block comment masks to end-of-file, newlines kept
|
|
unclosed = "/* CREATE PROC [dbo].[Ghost] AS\nBEGIN SELECT 1; END"
|
|
assert "CREATE" not in mask(unclosed) and len(mask(unclosed)) == len(unclosed)
|
|
# a /* inside a MULTI-LINE string is consumed as (blanked, line-scoped)
|
|
# string content — it must NOT open a comment that swallows the file
|
|
multi = (
|
|
"SET @sql = N'SELECT a /* c1\n , b FROM t';\nGO\n"
|
|
"CREATE PROCEDURE dbo.usp_Real AS BEGIN SELECT 1; END;\n"
|
|
)
|
|
out = mask(multi)
|
|
assert "CREATE PROCEDURE dbo.usp_Real" in out and len(out) == len(multi)
|
|
# block comments NEST (SQL Server and PostgreSQL both nest /* */): DDL
|
|
# commented out inside a nested comment must not survive the mask
|
|
nested = "/* outer\n /* inner */\n CREATE PROC dbo.usp_Fake AS BEGIN SELECT 1; END;\n*/\n"
|
|
out = mask(nested)
|
|
assert "CREATE" not in out and len(out) == len(nested)
|
|
|
|
|
|
def test_mask_sql_comments_invariants_fuzz():
|
|
"""Deterministic fuzz over the mask's structural invariants.
|
|
|
|
Both rounds of masking defects were shapes nobody thought to write down,
|
|
so pin properties instead: (1) one output character per input character,
|
|
(2) newlines exactly preserved, (3) blanked spans are spaces, (4) the
|
|
mask is idempotent — re-masking its own output changes nothing (the
|
|
quote re-pairing defect broke this: exposed text re-masked differently).
|
|
"""
|
|
import random
|
|
|
|
from graphify.extractors.sql import _mask_sql_comments as mask
|
|
|
|
rng = random.Random(0xC0FFEE)
|
|
alphabet = "ab[]\"'-*/ \n;.$"
|
|
for _ in range(20_000):
|
|
s = "".join(rng.choice(alphabet) for _ in range(rng.randint(1, 26)))
|
|
out = mask(s)
|
|
assert len(out) == len(s), (s, out)
|
|
for a, b in zip(s, out):
|
|
assert (a == "\n") == (b == "\n"), (s, out)
|
|
assert b == a or b == " ", (s, out)
|
|
assert mask(out) == out, (s, out)
|
|
|
|
|
|
# Frozen copy of _mask_sql_comments as of the 2026-08-24 hardening series,
|
|
# for the differential monotonicity fuzz below. Deliberately NOT imported
|
|
# from graphify: the point is that future edits to the live mask are
|
|
# compared against this fixed baseline. Update it only when a deliberate,
|
|
# reviewed decision changes what the mask must blank.
|
|
def _frozen_mask_2026_08_24(text: str) -> str:
|
|
"""Blank comment and string-literal spans, preserving every offset.
|
|
|
|
One output character per input character: non-newline characters inside
|
|
a blanked span become spaces and newlines are kept, so positions and
|
|
line numbers computed against the masked text are valid against the
|
|
original. Double-quoted and bracket-delimited identifiers are preserved
|
|
verbatim (they carry recoverable routine names); single-quoted strings,
|
|
line comments, and (nesting-aware) block comments are blanked. Used by
|
|
both routine-recovery scans so CREATE PROCEDURE/FUNCTION DDL reachable
|
|
only through a comment or a single-quoted string cannot fabricate a
|
|
routine node when an unrelated parse error arms recovery.
|
|
"""
|
|
out: list[str] = []
|
|
i, n = 0, len(text)
|
|
|
|
def _blank(upto: int) -> int:
|
|
"""Blank [i, upto), keeping newlines; return upto."""
|
|
for c in text[i:upto]:
|
|
out.append("\n" if c == "\n" else " ")
|
|
return upto
|
|
|
|
def _blank_tail_and_carry(start: int) -> int:
|
|
"""Blank from start to end-of-line, carrying comment state forward.
|
|
|
|
The union-of-readings blank for an ambiguous stretch: the rest of the
|
|
line is blanked outright; if the raw text of that stretch leaves a /*
|
|
unclosed on its own line (the maximum comment depth any reading could
|
|
be left holding), blanking continues, nesting-aware, to the closing
|
|
*/ or EOF. Where a carry closes MID-line the same rule applies to the
|
|
remainder of that line — under the reading where the carry never
|
|
opened, that whole line may be a comment or a string, so emitting the
|
|
post-*/ text verbatim exposed it (found by differential fuzzing).
|
|
Repeats until a line ends with no carry pending. Appends one output
|
|
character per input character; returns the resume index.
|
|
"""
|
|
k = start
|
|
while True:
|
|
eol = text.find("\n", k)
|
|
eol = n if eol == -1 else eol
|
|
depth = 0
|
|
m2 = k
|
|
while m2 < eol:
|
|
if text.startswith("/*", m2):
|
|
depth += 1
|
|
m2 += 2
|
|
elif text.startswith("*/", m2):
|
|
if depth:
|
|
depth -= 1
|
|
m2 += 2
|
|
else:
|
|
m2 += 1
|
|
for ch in text[k:eol]:
|
|
out.append("\n" if ch == "\n" else " ")
|
|
k = eol
|
|
if not depth:
|
|
return k
|
|
j2 = k
|
|
while j2 < n and depth:
|
|
if text.startswith("/*", j2):
|
|
depth += 1
|
|
j2 += 2
|
|
elif text.startswith("*/", j2):
|
|
depth -= 1
|
|
j2 += 2
|
|
else:
|
|
j2 += 1
|
|
for ch in text[k:j2]:
|
|
out.append("\n" if ch == "\n" else " ")
|
|
k = j2
|
|
if k >= n:
|
|
return k
|
|
# the carry closed mid-line: the remainder of THIS line is the
|
|
# same ambiguous stretch — loop and blank it too
|
|
|
|
while i < n:
|
|
c = text[i]
|
|
if c == "'":
|
|
# Single-quoted string: blank it. '' is an escaped quote; a
|
|
# newline abandons the literal (see the comment above).
|
|
j = i + 1
|
|
while j < n and text[j] != "\n":
|
|
if text[j] == "'":
|
|
if j + 1 < n and text[j + 1] == "'":
|
|
j += 2
|
|
continue
|
|
j += 1
|
|
break
|
|
j += 1
|
|
i = _blank(j)
|
|
elif c == '"' or c == "[":
|
|
# Delimited identifier: preserve verbatim. Doubled closers are
|
|
# escapes. A span is DISTRUSTED when it is unterminated (no
|
|
# closer before the newline) or would swallow a comment opener
|
|
# on its way to the closer ('SELECT [Col FROM t -- CREATE PROC
|
|
# [dbo]' closes on [dbo]'s bracket) — a stray delimiter is
|
|
# ordinary in exactly the broken files this mask runs on.
|
|
#
|
|
# A distrusted span is irreducibly ambiguous (identifier data vs
|
|
# stray delimiter before real comments/strings), and any attempt
|
|
# to pick one reading exposed text the other reading blanks —
|
|
# re-emitting the delimiter and rescanning even re-paired later
|
|
# single quotes and uncovered dynamic SQL. So blank the UNION of
|
|
# every reading: the rest of the line is blanked outright, and
|
|
# any raw /* on it with no later */ on the same line carries
|
|
# forward as (nesting-aware) comment state, since some reading
|
|
# may have left it open — and where that carry closes mid-line,
|
|
# the remainder of THAT line gets the same treatment, repeated
|
|
# until a line ends carry-free (under the no-carry reading the
|
|
# close line may itself be all comment or string, so emitting its
|
|
# post-*/ tail verbatim was an exposure). Over-blanking loses at
|
|
# most routines on lines already entangled with the broken one (a
|
|
# conservative false negative; a routine named like [a--b] is
|
|
# inside that loss); under-blanking is what fabricates, and every
|
|
# reading's blank set stays a subset of this one.
|
|
closer = '"' if c == '"' else "]"
|
|
j = i + 1
|
|
closed = False
|
|
while j < n and text[j] != "\n":
|
|
if text[j] == closer:
|
|
if j + 1 < n and text[j + 1] == closer:
|
|
j += 2
|
|
continue
|
|
j += 1
|
|
closed = True
|
|
break
|
|
j += 1
|
|
span = text[i:j]
|
|
if closed and "--" not in span and "/*" not in span:
|
|
out.append(span)
|
|
i = j
|
|
else:
|
|
i = _blank_tail_and_carry(i)
|
|
elif c == "-" and i + 1 < n and text[i + 1] == "-":
|
|
j = i
|
|
while j < n and text[j] != "\n":
|
|
j += 1
|
|
i = _blank(j)
|
|
elif c == "/" and i + 1 < n and text[i + 1] == "*":
|
|
depth, j = 1, i + 2
|
|
while j < n and depth:
|
|
if text[j] == "/" and j + 1 < n and text[j + 1] == "*":
|
|
depth += 1
|
|
j += 2
|
|
elif text[j] == "*" and j + 1 < n and text[j + 1] == "/":
|
|
depth -= 1
|
|
j += 2
|
|
else:
|
|
j += 1
|
|
i = _blank(j) # unclosed comment: j == n, blanks to end-of-file
|
|
else:
|
|
out.append(c)
|
|
i += 1
|
|
return "".join(out)
|
|
|
|
|
|
def test_mask_sql_comments_monotone_against_frozen_baseline():
|
|
"""Differential fuzz: the live mask must never EXPOSE a character the
|
|
frozen baseline blanks. Blanking more is allowed (conservative); blanking
|
|
less is the fabrication channel — every masking defect in the 2026-08-24
|
|
series (quote re-pairing, carry-close exposure) was a monotonicity
|
|
violation of exactly this kind, and none was caught by hand-written
|
|
shapes first.
|
|
|
|
If an INTENTIONAL over-blanking change lands (the live mask blanks more
|
|
than the baseline), this still passes; if an intentional EXPOSURE change
|
|
lands (the mask deliberately blanks less), refresh the frozen copy above
|
|
to the new revision as part of that reviewed change — do not delete the
|
|
test or loosen the property.
|
|
"""
|
|
import random
|
|
|
|
from graphify.extractors.sql import _mask_sql_comments as mask
|
|
|
|
rng = random.Random(0xBA5E11E)
|
|
alphabet = "ab[]\"'-*/ \n;.$"
|
|
for _ in range(20_000):
|
|
s = "".join(rng.choice(alphabet) for _ in range(rng.randint(1, 40)))
|
|
live = mask(s)
|
|
base = _frozen_mask_2026_08_24(s)
|
|
assert len(live) == len(base) == len(s)
|
|
for idx, (b, lv, orig) in enumerate(zip(base, live, s)):
|
|
if b == " " and orig != " ":
|
|
assert lv == " ", (
|
|
f"live mask exposes char {idx} ({orig!r}) that the frozen "
|
|
f"baseline blanks:\n input={s!r}\n base={base!r}\n live={live!r}"
|
|
)
|
|
|
|
|
|
def test_sql_dynamic_sql_and_unclosed_comment_do_not_fabricate_routines(tmp_path):
|
|
"""DDL text reachable only through a single-quoted string (dynamic SQL) or
|
|
an unterminated block comment must not mint routine nodes when an
|
|
unrelated parse error arms the recovery scans."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "dynamic.sql"
|
|
p.write_text(
|
|
"THIS IS NOT SQL AT ALL %%%;\n"
|
|
"EXEC(N'CREATE OR ALTER PROC [dbo].[usp_Dynamic] AS BEGIN SELECT 1; END');\n"
|
|
"SET @sql = N'SELECT 1 /* opener inside a multi-line string\n"
|
|
" must not swallow what follows';\n"
|
|
"/* nested /* comments */ hide this:\n"
|
|
"CREATE PROC [dbo].[usp_Nested] AS BEGIN SELECT 1; END;\n*/\n"
|
|
"SELECT [Col FROM t -- CREATE PROC [dbo].[usp_Bracketed] AS BEGIN SELECT 1; END;\n"
|
|
"CREATE PROCEDURE [dbo].[usp_Real]\nAS\nBEGIN\n SELECT 1;\nEND;\n"
|
|
"/* an unterminated comment swallows the rest of the file\n"
|
|
"CREATE PROC [dbo].[usp_Unterminated] AS BEGIN SELECT 1; END;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
labels = [n["label"] for n in r["nodes"] if n["label"] != "dynamic.sql"]
|
|
assert "[dbo].[usp_Real]()" in labels, labels
|
|
fabricated = [
|
|
label for label in labels
|
|
if any(k in label for k in ("Dynamic", "Nested", "Bracketed", "Unterminated"))
|
|
]
|
|
assert not fabricated, (
|
|
f"string-embedded or comment-swallowed DDL fabricated a node: {labels}"
|
|
)
|
|
|
|
|
|
def test_sql_ddl_keywords_inside_delimited_identifiers_do_not_fabricate(tmp_path):
|
|
"""A preserved delimited identifier keeps its text verbatim in the masked
|
|
scan (it may BE a recoverable name), so DDL keywords occurring inside one
|
|
('SELECT 1 AS [CREATE PROCEDURE dbo.usp_Phantom pending]') were matched
|
|
by the recovery regex and minted a phantom routine. Recovery now skips a
|
|
match whose CREATE starts inside a preserved identifier span — a genuine
|
|
statement's NAME may be a delimited identifier; its CREATE never is."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "alias.sql"
|
|
p.write_text(
|
|
"THIS IS NOT SQL AT ALL %%%;\n"
|
|
"SELECT 1 AS [CREATE PROCEDURE dbo.usp_Phantom pending];\n"
|
|
'SELECT 2 AS "CREATE PROC dbo.usp_Phantom2 queued";\n'
|
|
"CREATE PROCEDURE [dbo].[usp_Real]\nAS\nBEGIN\n SELECT 1;\nEND;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
labels = [n["label"] for n in r["nodes"] if n["label"] != "alias.sql"]
|
|
assert labels == ["[dbo].[usp_Real]()"], labels
|
|
|
|
|
|
def test_sql_create_inside_a_bare_word_does_not_fabricate(tmp_path):
|
|
"""CREATE must match as a word: without a leading boundary the recovery
|
|
regex matched CREATE inside a bare identifier, so 'SELECT AUTOCREATE
|
|
PROCEDURE x FROM t;' in an error-bearing file minted a phantom x().
|
|
(Delimited identifiers are span-skipped at the scan site; a bare word
|
|
has no span, so the regex itself must refuse.)"""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "bare.sql"
|
|
p.write_text(
|
|
"THIS IS NOT SQL AT ALL %%%;\n"
|
|
"SELECT AUTOCREATE PROCEDURE x FROM t;\n"
|
|
"EXEC dbo.PRECREATE FUNCTION y;\n"
|
|
"CREATE PROCEDURE [dbo].[usp_Real]\nAS\nBEGIN\n SELECT 1;\nEND;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
labels = [n["label"] for n in r["nodes"] if n["label"] != "bare.sql"]
|
|
assert labels == ["[dbo].[usp_Real]()"], labels
|
|
|
|
|
|
def test_sql_escaped_double_quote_in_routine_name_is_consumed(tmp_path):
|
|
"""ANSI SQL escapes a literal double quote inside a delimited identifier
|
|
by doubling it: "a""b" names the identifier a"b. A pattern that stops at
|
|
the first closing quote truncated the recovered name to "dbo"."a" — the
|
|
"" twin of the bracket ]] escape handled above.
|
|
|
|
The AS BEGIN body idiom keeps the whole statement in ERROR recovery: for a
|
|
statement the grammar CAN parse, tree-sitter-sql itself truncates the
|
|
object_reference at the "" escape (emitting `"b"` as a stray ERROR child),
|
|
which is an upstream grammar defect this extractor cannot repair."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "fn.sql"
|
|
p.write_text('CREATE PROCEDURE "dbo"."a""b"\nAS\nBEGIN\n SELECT 1;\nEND;\n')
|
|
r = extract_sql(p)
|
|
routine = [n["label"] for n in r["nodes"] if n["label"] != "fn.sql"]
|
|
assert routine == ['"dbo"."a""b"()'], routine
|
|
|
|
|
|
def test_sql_cte_is_not_read_as_a_table():
|
|
"""#2577: a name bound by WITH ... AS (...) is scoped to its statement, not a table.
|
|
|
|
Emitting it as a reads_from target minted a bare, sourceless stub carrying no
|
|
schema, file, or language namespace, so a CTE named `levels` or `slug` collided
|
|
with a same-named node from another language. The real table in the same
|
|
FROM/JOIN must still resolve.
|
|
"""
|
|
r = _extract_sql_or_skip("sample_cte.sql")
|
|
labels = [n["label"] for n in r["nodes"]]
|
|
assert "levels" not in labels, "CTE name leaked into the graph as a table node"
|
|
|
|
reads = [e for e in r["edges"] if e["relation"] == "reads_from"]
|
|
assert reads, "the real table reference should still emit a reads_from edge"
|
|
assert not any(e["target"] == "levels" for e in reads), "CTE emitted as a reads_from target"
|
|
|
|
# the real v_roles -> users edge is kept
|
|
nid = {n["label"]: n["id"] for n in r["nodes"]}
|
|
assert (nid["v_roles"], nid["users"]) in {(e["source"], e["target"]) for e in reads}
|
|
|
|
def test_sql_column_list_cte_is_not_read_as_a_table(tmp_path):
|
|
"""#2577: `WITH levels(a, b) AS (...)` — the name precedes a column list."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"CREATE TABLE users (id INT, role TEXT);\n"
|
|
"CREATE VIEW v AS\n"
|
|
" WITH levels(role, rank) AS (SELECT 'admin', 1)\n"
|
|
" SELECT * FROM users JOIN levels ON levels.role = users.role;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
labels = [n["label"] for n in r["nodes"]]
|
|
assert "levels" not in labels
|
|
reads = [e for e in r["edges"] if e["relation"] == "reads_from"]
|
|
assert not any(e["target"] == "levels" for e in reads)
|
|
|
|
def test_sql_cte_shadows_same_named_table_within_its_statement(tmp_path):
|
|
"""#2577: inside the declaring statement the CTE shadows a real same-named
|
|
table (SQL scoping), so v1's FROM binds to the CTE and emits nothing; v2 has
|
|
no CTE in scope and reads the real table. Exactly one deterministic edge."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"CREATE TABLE levels (role TEXT);\n"
|
|
"CREATE VIEW v1 AS WITH levels AS (SELECT 'admin' AS role)"
|
|
" SELECT * FROM levels;\n"
|
|
"CREATE VIEW v2 AS SELECT * FROM levels;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
reads = [e for e in r["edges"] if e["relation"] == "reads_from"]
|
|
assert len(reads) == 1, f"expected exactly one reads_from, got {reads}"
|
|
nid = {n["label"]: n["id"] for n in r["nodes"]}
|
|
assert reads[0]["source"] == nid["v2"]
|
|
assert reads[0]["target"] == nid["levels"] # the real, sourced table node
|
|
|
|
def test_sql_subquery_cte_does_not_suppress_outer_real_table(tmp_path):
|
|
"""#2577 refinement: a WITH inside a subquery is scoped to that subquery
|
|
only. A statement-wide pre-collect would also swallow the OUTER reference
|
|
to the real `t2`, dropping a true edge — per-subtree scoping keeps it."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
p = tmp_path / "schema.sql"
|
|
p.write_text(
|
|
"CREATE TABLE t2 (id INT);\n"
|
|
"CREATE VIEW v6 AS SELECT * FROM t2 JOIN"
|
|
" (WITH t2 AS (SELECT 1 AS id) SELECT * FROM t2) sub"
|
|
" ON sub.id = t2.id;\n"
|
|
)
|
|
r = extract_sql(p)
|
|
nid = {n["label"]: n["id"] for n in r["nodes"]}
|
|
reads = {(e["source"], e["target"]) for e in r["edges"]
|
|
if e["relation"] == "reads_from"}
|
|
assert (nid["v6"], nid["t2"]) in reads, (
|
|
"outer reference to the real t2 table was wrongly suppressed"
|
|
)
|
|
|
|
def test_php_builtin_base_class_never_inherits_from_cross_language_class(tmp_path):
|
|
"""#2812: `class FooApiException extends \\Exception` names PHP's global
|
|
built-in. The sourceless stub it mints was unique corpus-wide, so the rewire
|
|
bound it to an unrelated TypeScript `Exception` class and the PHP class
|
|
inherited across languages. No supertype edge may target a TypeScript node."""
|
|
ts = tmp_path / "app" / "exception.ts"
|
|
ts.parent.mkdir(parents=True)
|
|
ts.write_text(
|
|
"export class Exception extends Error {\n"
|
|
" constructor(message: string) { super(message); }\n"
|
|
"}\n"
|
|
)
|
|
php = tmp_path / "packages" / "FooApiException.php"
|
|
php.parent.mkdir(parents=True)
|
|
php.write_text(
|
|
"<?php\n"
|
|
"\n"
|
|
"namespace Foo\\Exceptions;\n"
|
|
"\n"
|
|
"class FooApiException extends \\Exception\n"
|
|
"{\n"
|
|
"}\n"
|
|
)
|
|
|
|
r = extract([php, ts], root=tmp_path)
|
|
ts_nodes = {n["id"] for n in r["nodes"]
|
|
if str(n.get("source_file", "")).endswith(".ts")}
|
|
supertypes = [e for e in r["edges"]
|
|
if e["relation"] in ("inherits", "implements", "extends")
|
|
and str(e.get("source_file", "")).endswith(".php")]
|
|
assert supertypes, "PHP inheritance was not extracted at all"
|
|
for e in supertypes:
|
|
assert e["target"] not in ts_nodes, (
|
|
f"PHP supertype leaked cross-language: {e}"
|
|
)
|
|
# The base stays on its sourceless external stub rather than vanishing.
|
|
sourceless = {n["id"] for n in r["nodes"] if not n.get("source_file")}
|
|
assert any(e["target"] in sourceless for e in supertypes), (
|
|
f"PHP base class lost its external stub: {supertypes}"
|
|
)
|
|
|
|
|
|
def test_sql_cte_never_binds_to_cross_language_symbol(tmp_path):
|
|
"""#2577: the reported leak — the CTE's sourceless stub was unique corpus-wide,
|
|
so _rewire_unique_stub_nodes bound it to a same-named symbol from ANOTHER
|
|
language (schema_v_roles -> ui_levels). With the CTE excluded, no reads_from
|
|
edge may target a TypeScript node."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
sql = tmp_path / "schema.sql"
|
|
sql.write_text(
|
|
"CREATE TABLE users (id INT, role TEXT);\n"
|
|
"CREATE VIEW v_roles AS\n"
|
|
" WITH levels AS (SELECT 'admin' AS role)\n"
|
|
" SELECT * FROM users JOIN levels ON levels.role = users.role;\n"
|
|
)
|
|
ts = tmp_path / "ui.ts"
|
|
ts.write_text("export function levels() { return ['admin']; }\n")
|
|
|
|
r = extract([sql, ts], root=tmp_path)
|
|
ts_nodes = {n["id"] for n in r["nodes"]
|
|
if str(n.get("source_file", "")).endswith(".ts")}
|
|
for e in r["edges"]:
|
|
if e["relation"] == "reads_from":
|
|
assert e["target"] not in ts_nodes, (
|
|
f"SQL reads_from leaked cross-language: {e}"
|
|
)
|
|
|
|
def test_sql_cross_file_fk_resolves_and_never_leaks_scan_path(tmp_path):
|
|
"""#2324: a REFERENCES target defined in ANOTHER file must collapse onto the
|
|
real table node (via the sourceless-stub rewire), and no node id or edge
|
|
endpoint may embed the absolute scan path. Before the fix, the fallback
|
|
minted a node-less id under the referencing file's own stem, which with
|
|
absolute inputs leaked the machine path AND could never match the m1
|
|
definition, so prisma-style cross-migration FKs dangled."""
|
|
pytest.importorskip("tree_sitter_sql")
|
|
from graphify.ids import make_id
|
|
|
|
m1 = tmp_path / "prisma" / "migrations" / "m1"
|
|
m2 = tmp_path / "prisma" / "migrations" / "m2"
|
|
m1.mkdir(parents=True)
|
|
m2.mkdir(parents=True)
|
|
(m1 / "migration.sql").write_text(
|
|
'CREATE TABLE "Tenant" (\n'
|
|
' "id" TEXT NOT NULL,\n'
|
|
' CONSTRAINT "Tenant_pkey" PRIMARY KEY ("id")\n'
|
|
');\n'
|
|
)
|
|
(m2 / "migration.sql").write_text(
|
|
'CREATE TABLE "StockGapEvent" (\n'
|
|
' "id" TEXT NOT NULL,\n'
|
|
' "tenantId" TEXT NOT NULL,\n'
|
|
' CONSTRAINT "StockGapEvent_pkey" PRIMARY KEY ("id")\n'
|
|
');\n'
|
|
'ALTER TABLE "StockGapEvent" ADD CONSTRAINT "StockGapEvent_tenantId_fkey"'
|
|
' FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id");\n'
|
|
)
|
|
|
|
r = extract(
|
|
[(m1 / "migration.sql").resolve(), (m2 / "migration.sql").resolve()],
|
|
root=tmp_path,
|
|
)
|
|
node_ids = {n["id"] for n in r["nodes"]}
|
|
|
|
# (a) the FK resolved cross-file onto the REAL Tenant definition node
|
|
tenant_ids = [i for i in node_ids if i.endswith("m1_migration_tenant")]
|
|
assert len(tenant_ids) == 1, f"expected one real Tenant node, got {tenant_ids}"
|
|
ref_targets = {e["target"] for e in r["edges"] if e["relation"] == "references"}
|
|
assert tenant_ids[0] in ref_targets, (
|
|
f"cross-file FK did not rewire onto {tenant_ids[0]}; "
|
|
f"references targets: {ref_targets}"
|
|
)
|
|
|
|
# (b) no dangling endpoints anywhere
|
|
for e in r["edges"]:
|
|
assert e["source"] in node_ids, f"dangling source: {e['source']}"
|
|
assert e["target"] in node_ids, f"dangling target: {e['target']}"
|
|
|
|
# (c) the absolute scan path never leaks into any id or endpoint
|
|
abs_slug = make_id(str(tmp_path.resolve()))
|
|
for i in node_ids:
|
|
assert abs_slug not in i, f"absolute path leaked into node id: {i}"
|
|
for e in r["edges"]:
|
|
assert abs_slug not in e["source"], f"absolute path leaked: {e['source']}"
|
|
assert abs_slug not in e["target"], f"absolute path leaked: {e['target']}"
|
|
|
|
def test_sql_alter_table_fk_edge():
|
|
"""ALTER TABLE ... FOREIGN KEY ... REFERENCES produces a references edge."""
|
|
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"]}
|
|
for e in fk_edges:
|
|
assert e["source"] in node_ids, f"dangling source: {e['source']}"
|
|
assert e["target"] in node_ids, f"dangling target: {e['target']}"
|
|
|
|
def test_sql_schema_qualified_names():
|
|
"""Schema-qualified table names (Schema.Table) are preserved."""
|
|
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_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"]}
|
|
for e in fk_edges:
|
|
assert e["source"] in node_ids, f"dangling source: {e['source']}"
|
|
assert e["target"] in node_ids, f"dangling target: {e['target']}"
|
|
|
|
def test_sql_plpgsql_functions_survive_parse_errors():
|
|
"""PL/pgSQL bodies make tree-sitter-sql emit ERROR nodes; the functions
|
|
must still be extracted (#1910), without cascading into later statements."""
|
|
r = _extract_sql_or_skip("sample_plpgsql.sql")
|
|
labels = [n["label"] for n in r["nodes"]]
|
|
# Both PL/pgSQL functions extracted, schema-qualified name kept whole
|
|
assert "exposed.important_function()" in labels
|
|
assert "tagged_quote_fn()" in labels
|
|
# Tables before and after the broken functions still extract
|
|
assert any("accounts" in l for l in labels)
|
|
assert any("audit_log" in l for l in labels)
|
|
# No spurious or empty nodes from the error recovery
|
|
for l in labels:
|
|
assert l, "empty node label"
|
|
assert l != "ERROR"
|
|
# Every function got a contains edge from the file node
|
|
contains_targets = {e["target"] for e in r["edges"] if e["relation"] == "contains"}
|
|
fn_ids = {n["id"] for n in r["nodes"] if n["label"].endswith("()")}
|
|
assert fn_ids <= contains_targets
|
|
|
|
def test_sql_plpgsql_clean_function_not_double_emitted():
|
|
"""A cleanly-parsed LANGUAGE sql function in the same file is emitted once."""
|
|
r = _extract_sql_or_skip("sample_plpgsql.sql")
|
|
labels = [n["label"] for n in r["nodes"]]
|
|
assert labels.count("plain_sql_fn()") == 1
|
|
# And nothing else is duplicated either
|
|
ids = [n["id"] for n in r["nodes"]]
|
|
assert len(ids) == len(set(ids))
|
|
|
|
def test_sql_quoted_plpgsql_routines_are_recovered():
|
|
"""#2180: quoted identifiers must not defeat the ERROR-node name recovery.
|
|
|
|
Generated PostgreSQL DDL (Supabase-style dumps, for example) quotes every
|
|
identifier: CREATE OR REPLACE FUNCTION "public"."fn"(...). The recovery
|
|
pattern matched a bare [\\w$.]+, which stops at the leading quote, so every
|
|
routine whose body also failed to parse -- RAISE, PERFORM, :=, IF..THEN,
|
|
bare NULL; -- was dropped with no warning and exit code 0. The same body
|
|
with an *unquoted* name recovered fine, which is why the drop looked like it
|
|
depended only on the body statement.
|
|
"""
|
|
r = _extract_sql_or_skip("sample_plpgsql_quoted.sql")
|
|
labels = [n["label"] for n in r["nodes"]]
|
|
for name in (
|
|
"raise_exception_fn",
|
|
"raise_notice_fn",
|
|
"perform_fn",
|
|
"assign_fn",
|
|
"if_then_fn",
|
|
"null_body_fn",
|
|
"quoted_proc",
|
|
):
|
|
assert f'"public"."{name}"()' in labels, f"{name} dropped from a quoted-DDL file (#2180)"
|
|
|
|
def test_sql_quoted_plpgsql_file_stays_clean():
|
|
"""The #2180 recovery must not add junk, duplicates, or drop the tables."""
|
|
r = _extract_sql_or_skip("sample_plpgsql_quoted.sql")
|
|
labels = [n["label"] for n in r["nodes"]]
|
|
# Tables before and after the unparseable routines still extract.
|
|
assert any("accounts" in l for l in labels)
|
|
assert any("audit_log" in l for l in labels)
|
|
# No empty or ERROR labels leaked out of the recovery.
|
|
for l in labels:
|
|
assert l, "empty node label"
|
|
assert l != "ERROR"
|
|
# Nothing emitted twice (a routine must not come from both paths).
|
|
ids = [n["id"] for n in r["nodes"]]
|
|
assert len(ids) == len(set(ids)), "duplicate node ids"
|
|
assert len(labels) == len(set(labels)), f"duplicate labels: {labels}"
|
|
# Every recovered routine is reachable from the file node.
|
|
contains_targets = {e["target"] for e in r["edges"] if e["relation"] == "contains"}
|
|
fn_ids = {n["id"] for n in r["nodes"] if n["label"].endswith("()")}
|
|
assert fn_ids <= contains_targets
|