mirror of
https://github.com/Graphify-Labs/graphify.git
synced 2026-09-14 19:34:09 +08:00
L4XB
ab80fa7413
The Rust extractor gated every non-function item on struct_item,
enum_item and trait_item, so tree-sitter-rust's static_item and
const_item never produced a node: a constant existed in the graph only
through the files that referenced it, never from the Rust that defines
it (0 of 783 names in a jdx/mise snapshot).
Add a branch for both node types: a module-level declaration gets a
node with a file-level contains edge, an associated const inside an
impl is attributed to the impl like a method, and the declared type is
referenced the way a struct field type is.
Fixes #3471
(cherry picked from commit 70ff464064)
1420 lines
62 KiB
Python
1420 lines
62 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_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}"
|
|
)
|
|
|
|
|
|
# ── 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_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_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
|