mirror of
https://github.com/safishamsi/graphify.git
synced 2026-07-12 10:27:11 +00:00
jerp86
71d7ab9d71
The semantic pass mints a document node <slug>_doc; the markdown quick-scan (extract_markdown) mints the bare <slug>. After a semantic build, a `graphify update` (AST path) re-runs the quick-scan and the graph ends up with BOTH — one document as two disconnected nodes, the file's edges split between them (semantic `references`/hyperedges on the _doc twin, quick-scan cross-links on the bare one). path/query traversals dead-end on the wrong twin; degree and communities split. build_from_json now reconciles the pair: when <slug> and <slug>_doc both exist with the same source_file and both are file_type=document, remap the bare node into the semantic _doc node (canonical, richer edges) and repoint its edges and hyperedges. Remap-induced self-loops are dropped; pre-existing ones are left alone. Gated to document twins for the same file, so a code symbol `foo` and an unrelated `foo_doc` never merge. Regression tests for both. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
973 lines
45 KiB
Python
973 lines
45 KiB
Python
import json
|
|
from pathlib import Path
|
|
import networkx as nx
|
|
from networkx.readwrite import json_graph
|
|
from graphify.build import build_from_json, build, build_merge, edge_data, edge_datas, dedupe_edges, dedupe_nodes
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def test_dedupe_edges_collapses_exact_parallels():
|
|
# #1317: --no-cluster / incremental update concatenate edge lists raw.
|
|
edges = [
|
|
{"source": "a", "target": "b", "relation": "calls", "source_location": "L1"},
|
|
{"source": "a", "target": "b", "relation": "calls", "source_location": "L9"}, # dup
|
|
{"source": "a", "target": "b", "relation": "imports"}, # different relation: kept
|
|
{"source": "b", "target": "c", "relation": "calls"},
|
|
]
|
|
out = dedupe_edges(edges)
|
|
keys = [(e["source"], e["target"], e["relation"]) for e in out]
|
|
assert keys == [("a", "b", "calls"), ("a", "b", "imports"), ("b", "c", "calls")]
|
|
# first occurrence wins (keeps L1, not L9)
|
|
assert out[0]["source_location"] == "L1"
|
|
|
|
|
|
def test_dedupe_edges_is_idempotent():
|
|
edges = [
|
|
{"source": "a", "target": "b", "relation": "calls"},
|
|
{"source": "a", "target": "b", "relation": "calls"},
|
|
]
|
|
once = dedupe_edges(edges)
|
|
twice = dedupe_edges(once + edges) # simulate a second `update` re-concatenating
|
|
assert len(once) == 1
|
|
assert len(twice) == 1
|
|
|
|
|
|
def test_dedupe_nodes_collapses_by_id_last_wins():
|
|
# #1327: a shared module anchor is emitted once per importing file; the
|
|
# --no-cluster raw writer must collapse same-id node dicts (#1317).
|
|
nodes = [
|
|
{"id": "foundation", "label": "Foundation", "type": "module", "source_file": "A.swift"},
|
|
{"id": "akit", "label": "AKit", "file_type": "code"},
|
|
{"id": "foundation", "label": "Foundation", "type": "module", "source_file": "B.swift"},
|
|
]
|
|
out = dedupe_nodes(nodes)
|
|
ids = [n["id"] for n in out]
|
|
assert ids == ["foundation", "akit"] # first-appearance order
|
|
# last writer wins on attributes
|
|
assert next(n for n in out if n["id"] == "foundation")["source_file"] == "B.swift"
|
|
|
|
def load_extraction():
|
|
return json.loads((FIXTURES / "extraction.json").read_text())
|
|
|
|
def test_build_from_json_node_count():
|
|
G = build_from_json(load_extraction())
|
|
assert G.number_of_nodes() == 4
|
|
|
|
def test_build_from_json_edge_count():
|
|
G = build_from_json(load_extraction())
|
|
assert G.number_of_edges() == 4
|
|
|
|
def test_nodes_have_label():
|
|
G = build_from_json(load_extraction())
|
|
assert G.nodes["n_transformer"]["label"] == "Transformer"
|
|
|
|
def test_edges_have_confidence():
|
|
G = build_from_json(load_extraction())
|
|
data = G.edges["n_attention", "n_concept_attn"]
|
|
assert data["confidence"] == "INFERRED"
|
|
|
|
def test_ambiguous_edge_preserved():
|
|
G = build_from_json(load_extraction())
|
|
data = G.edges["n_layernorm", "n_concept_attn"]
|
|
assert data["confidence"] == "AMBIGUOUS"
|
|
|
|
def test_legacy_node_source_canonicalized():
|
|
"""Legacy 'source' key on nodes is renamed to 'source_file' before graph build."""
|
|
ext = {"nodes": [{"id": "n1", "label": "A", "file_type": "code", "source": "a.py"}],
|
|
"edges": [], "input_tokens": 0, "output_tokens": 0}
|
|
G = build_from_json(ext)
|
|
assert "source_file" in G.nodes["n1"]
|
|
assert G.nodes["n1"]["source_file"] == "a.py"
|
|
assert "source" not in G.nodes["n1"]
|
|
|
|
|
|
def test_legacy_edge_from_to_canonicalized():
|
|
"""Legacy 'from'/'to' keys on edges are accepted alongside 'source'/'target'."""
|
|
ext = {"nodes": [{"id": "n1", "label": "A", "file_type": "code", "source_file": "a.py"},
|
|
{"id": "n2", "label": "B", "file_type": "code", "source_file": "b.py"}],
|
|
"edges": [{"from": "n1", "to": "n2", "relation": "calls",
|
|
"confidence": "EXTRACTED", "source_file": "a.py", "weight": 1.0}],
|
|
"input_tokens": 0, "output_tokens": 0}
|
|
G = build_from_json(ext)
|
|
assert G.number_of_edges() == 1
|
|
|
|
|
|
def test_source_file_backslash_normalized():
|
|
"""Windows backslash paths and POSIX paths for the same file must produce one node."""
|
|
extraction = {
|
|
"nodes": [
|
|
{"id": "n1", "label": "A", "file_type": "code", "source_file": "src\\middleware\\auth.py"},
|
|
{"id": "n2", "label": "B", "file_type": "code", "source_file": "src/middleware/auth.py"},
|
|
],
|
|
"edges": [],
|
|
"input_tokens": 0, "output_tokens": 0,
|
|
}
|
|
G = build_from_json(extraction)
|
|
sources = {G.nodes[n]["source_file"] for n in G.nodes()}
|
|
assert sources == {"src/middleware/auth.py"}
|
|
|
|
|
|
def test_edge_missing_source_file_backfilled_from_node():
|
|
"""#1279: a semantic/LLM edge lacking source_file must inherit it from its
|
|
source node rather than reach graph.json with no file reference."""
|
|
extraction = {
|
|
"nodes": [
|
|
{"id": "n1", "label": "A", "file_type": "concept", "source_file": "docs/a.md"},
|
|
{"id": "n2", "label": "B", "file_type": "concept", "source_file": "docs/b.md"},
|
|
],
|
|
# No source_file on the edge (as LLM output sometimes omits it).
|
|
"edges": [{"source": "n1", "target": "n2", "relation": "relates_to", "confidence": "INFERRED"}],
|
|
"input_tokens": 0, "output_tokens": 0,
|
|
}
|
|
G = build_from_json(extraction)
|
|
sf = edge_data(G, "n1", "n2").get("source_file")
|
|
assert sf == "docs/a.md" # backfilled from the source node
|
|
|
|
|
|
def test_build_merges_multiple_extractions():
|
|
ext1 = {"nodes": [{"id": "n1", "label": "A", "file_type": "code", "source_file": "a.py"}],
|
|
"edges": [], "input_tokens": 0, "output_tokens": 0}
|
|
ext2 = {"nodes": [{"id": "n2", "label": "B", "file_type": "document", "source_file": "b.md"}],
|
|
"edges": [{"source": "n1", "target": "n2", "relation": "references",
|
|
"confidence": "INFERRED", "source_file": "b.md", "weight": 1.0}],
|
|
"input_tokens": 0, "output_tokens": 0}
|
|
G = build([ext1, ext2])
|
|
assert G.number_of_nodes() == 2
|
|
assert G.number_of_edges() == 1
|
|
|
|
|
|
def test_none_file_type_defaults_to_concept(capsys):
|
|
"""Legacy nodes with file_type=None (e.g. preserved from older graph.json
|
|
by `_rebuild_code`) must not trigger 'invalid file_type None' warnings (#660)."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "n1", "label": "Stub", "file_type": None, "source_file": "a.py"},
|
|
{"id": "n2", "label": "Real", "file_type": "code", "source_file": "b.py"},
|
|
],
|
|
"edges": [],
|
|
"input_tokens": 0,
|
|
"output_tokens": 0,
|
|
}
|
|
G = build_from_json(ext)
|
|
err = capsys.readouterr().err
|
|
assert "invalid file_type" not in err
|
|
# The legacy node still exists in the graph and has been canonicalized
|
|
assert G.nodes["n1"]["file_type"] == "concept"
|
|
assert G.nodes["n2"]["file_type"] == "code"
|
|
|
|
|
|
def test_missing_file_type_defaults_to_concept(capsys):
|
|
"""Nodes missing file_type entirely should also be canonicalized to 'concept'."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "n1", "label": "Bare", "source_file": "a.py"},
|
|
],
|
|
"edges": [],
|
|
"input_tokens": 0,
|
|
"output_tokens": 0,
|
|
}
|
|
G = build_from_json(ext)
|
|
err = capsys.readouterr().err
|
|
assert "invalid file_type" not in err
|
|
assert "missing required field 'file_type'" not in err
|
|
assert G.nodes["n1"]["file_type"] == "concept"
|
|
|
|
|
|
def test_real_invalid_file_type_coerced_to_concept():
|
|
"""Unknown file_type values are coerced through the synonym mapper, falling
|
|
back to 'concept' for anything that isn't a known LLM synonym (#840)."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "n1", "label": "Bad", "file_type": "weird_type", "source_file": "a.py"},
|
|
],
|
|
"edges": [],
|
|
"input_tokens": 0,
|
|
"output_tokens": 0,
|
|
}
|
|
G = build_from_json(ext)
|
|
assert G.nodes["n1"]["file_type"] == "concept"
|
|
|
|
|
|
def test_file_type_synonym_mapping():
|
|
"""Known invalid file_type values map to their canonical equivalents."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "n1", "label": "MD", "file_type": "markdown", "source_file": "a.md"},
|
|
{"id": "n2", "label": "Tool", "file_type": "tool", "source_file": "b.py"},
|
|
{"id": "n3", "label": "Pat", "file_type": "pattern", "source_file": "c.md"},
|
|
],
|
|
"edges": [],
|
|
"input_tokens": 0,
|
|
"output_tokens": 0,
|
|
}
|
|
G = build_from_json(ext)
|
|
assert G.nodes["n1"]["file_type"] == "document"
|
|
assert G.nodes["n2"]["file_type"] == "code"
|
|
assert G.nodes["n3"]["file_type"] == "concept"
|
|
|
|
|
|
def test_ghost_merge_unique_located_node_still_merges():
|
|
"""#1145 ghost-merge: a semantic ghost collapses into the single AST node
|
|
sharing its (basename, label), and edges re-point to the AST node."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "ast_render", "label": "render", "file_type": "code",
|
|
"source_file": "src/app/index.ts", "source_location": "L10", "_origin": "ast"},
|
|
{"id": "ghost_render", "label": "render", "file_type": "code",
|
|
"source_file": "src/app/index.ts"},
|
|
{"id": "caller", "label": "main", "file_type": "code",
|
|
"source_file": "src/main.ts", "source_location": "L1", "_origin": "ast"},
|
|
],
|
|
"edges": [{"source": "caller", "target": "ghost_render", "relation": "calls",
|
|
"confidence": "EXTRACTED", "source_file": "src/main.ts", "weight": 1.0}],
|
|
"input_tokens": 0, "output_tokens": 0,
|
|
}
|
|
G = build_from_json(ext)
|
|
assert "ghost_render" not in G.nodes()
|
|
assert G.has_edge("caller", "ast_render")
|
|
|
|
|
|
def test_ghost_merge_skipped_on_basename_collision():
|
|
"""#1257: when two files with the same basename both define a symbol with the
|
|
same label, the (basename, label) key is ambiguous and the semantic ghost
|
|
must not be merged into an arbitrary one of them."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "a_render", "label": "render", "file_type": "code",
|
|
"source_file": "src/a/index.ts", "source_location": "L10", "_origin": "ast"},
|
|
{"id": "b_render", "label": "render", "file_type": "code",
|
|
"source_file": "src/b/index.ts", "source_location": "L20", "_origin": "ast"},
|
|
{"id": "ghost_render", "label": "render", "file_type": "code",
|
|
"source_file": "src/a/index.ts"},
|
|
{"id": "caller", "label": "main", "file_type": "code",
|
|
"source_file": "src/main.ts", "source_location": "L1", "_origin": "ast"},
|
|
],
|
|
"edges": [{"source": "caller", "target": "ghost_render", "relation": "calls",
|
|
"confidence": "EXTRACTED", "source_file": "src/main.ts", "weight": 1.0}],
|
|
"input_tokens": 0, "output_tokens": 0,
|
|
}
|
|
G = build_from_json(ext)
|
|
# The ghost survives: merging it into either a_render or b_render would
|
|
# pick an arbitrary winner (set iteration order over node_set).
|
|
assert "ghost_render" in G.nodes()
|
|
assert G.number_of_nodes() == 4
|
|
assert G.has_edge("caller", "ghost_render")
|
|
assert not G.has_edge("caller", "a_render")
|
|
assert not G.has_edge("caller", "b_render")
|
|
|
|
|
|
def test_ghost_merge_non_ast_different_files_both_survive():
|
|
"""#1753: two NON-AST (semantic) nodes sharing (basename, label) but from
|
|
DIFFERENT files are distinct concepts with no AST canonical twin. They must
|
|
not be merged into an arbitrary survivor (which flipped run-to-run with the
|
|
hash seed); both survive, mirroring the AST/AST guard (#1257)."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "dir_a_update_build_merge", "label": "build_merge() function",
|
|
"file_type": "concept", "source_file": "dir_a/update.md", "source_location": "L10"},
|
|
{"id": "dir_b_update_build_merge", "label": "build_merge() function",
|
|
"file_type": "concept", "source_file": "dir_b/update.md", "source_location": "L12"},
|
|
],
|
|
"edges": [],
|
|
}
|
|
G = build_from_json(ext, directed=False)
|
|
assert sorted(G.nodes()) == ["dir_a_update_build_merge", "dir_b_update_build_merge"]
|
|
|
|
|
|
def test_ghost_merge_non_ast_same_file_still_merges():
|
|
"""A genuine duplicate — two non-AST nodes with the SAME source_file and
|
|
label — is a real ghost and still collapses to one node (deterministically),
|
|
so #1753's fix doesn't leave same-file LLM duplicates behind."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "a_foo", "label": "Foo", "file_type": "concept",
|
|
"source_file": "x/doc.md", "source_location": "L1"},
|
|
{"id": "b_foo", "label": "Foo", "file_type": "concept",
|
|
"source_file": "x/doc.md", "source_location": "L2"},
|
|
],
|
|
"edges": [],
|
|
}
|
|
G = build_from_json(ext, directed=False)
|
|
assert G.number_of_nodes() == 1
|
|
|
|
|
|
def test_build_merge_preserves_call_edge_direction(tmp_path):
|
|
"""Regression for #760.
|
|
|
|
When the callee is defined before the caller in source, NetworkX's
|
|
undirected Graph stores edges in node-insertion order. Going through
|
|
node_link_graph() + edges() during build_merge previously flipped the
|
|
`calls` edge so that on the next save source/target were swapped.
|
|
|
|
build_merge must read the saved JSON's source/target verbatim instead
|
|
of round-tripping through NetworkX.
|
|
"""
|
|
from graphify.extract import extract_js
|
|
from graphify.export import to_json
|
|
|
|
# Callee `b` is defined before caller `a` so node insertion order
|
|
# is b, a. An undirected Graph then yields the edge as (b, a) on
|
|
# iteration, which is the wrong direction for `calls` (a calls b).
|
|
src = "function b() {}\nfunction a() { b(); }\n"
|
|
src_file = tmp_path / "x.js"
|
|
src_file.write_text(src)
|
|
|
|
extraction = extract_js(src_file)
|
|
assert "error" not in extraction
|
|
|
|
# Locate the `calls` edge in the raw extraction so we know the truth.
|
|
call_edges = [e for e in extraction["edges"] if e["relation"] == "calls"]
|
|
assert len(call_edges) == 1, "expected exactly one calls edge from the snippet"
|
|
truth_src = call_edges[0]["source"]
|
|
truth_tgt = call_edges[0]["target"]
|
|
|
|
nodes_by_id = {n["id"]: n for n in extraction["nodes"]}
|
|
assert nodes_by_id[truth_src]["label"].startswith("a")
|
|
assert nodes_by_id[truth_tgt]["label"].startswith("b")
|
|
|
|
# First build + save.
|
|
G1 = build([extraction], dedup=False)
|
|
graph_path = tmp_path / "graph.json"
|
|
communities: dict = {}
|
|
assert to_json(G1, communities, str(graph_path), force=True)
|
|
|
|
# Verify direction is correct in the freshly written JSON.
|
|
saved = json.loads(graph_path.read_text())
|
|
saved_calls = [e for e in saved.get("links", saved.get("edges", []))
|
|
if e.get("relation") == "calls"]
|
|
assert len(saved_calls) == 1
|
|
assert saved_calls[0]["source"] == truth_src
|
|
assert saved_calls[0]["target"] == truth_tgt
|
|
|
|
# Now simulate `--update` with no new chunks — load + re-save.
|
|
G2 = build_merge([], graph_path, dedup=False)
|
|
assert to_json(G2, communities, str(graph_path), force=True)
|
|
|
|
# The calls edge must still go a -> b, not b -> a.
|
|
reloaded = json.loads(graph_path.read_text())
|
|
reloaded_calls = [e for e in reloaded.get("links", reloaded.get("edges", []))
|
|
if e.get("relation") == "calls"]
|
|
assert len(reloaded_calls) == 1
|
|
assert reloaded_calls[0]["source"] == truth_src, (
|
|
f"calls edge source flipped after build_merge round-trip: "
|
|
f"expected {truth_src} (a), got {reloaded_calls[0]['source']}"
|
|
)
|
|
assert reloaded_calls[0]["target"] == truth_tgt, (
|
|
f"calls edge target flipped after build_merge round-trip: "
|
|
f"expected {truth_tgt} (b), got {reloaded_calls[0]['target']}"
|
|
)
|
|
|
|
|
|
def test_build_from_json_preserves_first_direction_on_bidirectional_pair(tmp_path):
|
|
"""Regression for #1061.
|
|
|
|
When an extraction emits two `calls` edges between the same pair in
|
|
opposite directions (mutual recursion, callbacks, event handlers, etc.),
|
|
nx.Graph collapses them into a single undirected edge. The deterministic
|
|
edge sort introduced in #1010 ordered edges by (source, target, relation),
|
|
so the lexicographically-later direction always wrote second and clobbered
|
|
the first edge's _src/_tgt — the surviving edge then exported with caller
|
|
and callee systematically swapped on every collision.
|
|
|
|
build_from_json must keep the first-seen direction for the surviving edge
|
|
instead of letting the second add_edge overwrite _src/_tgt.
|
|
"""
|
|
from graphify.export import to_json
|
|
|
|
# Lexicographic order of (src, tgt, rel) puts `a` < `z` first, so the sort
|
|
# processes `a -> z` BEFORE `z -> a`. Without the fix, the second write
|
|
# overwrites _src/_tgt and the exported edge becomes z -> a. With the fix,
|
|
# the first-seen `a -> z` direction is preserved.
|
|
extraction = {
|
|
"nodes": [
|
|
{"id": "a_handler", "label": "a", "file_type": "code", "source_file": "a.ts"},
|
|
{"id": "z_emitter", "label": "z", "file_type": "code", "source_file": "z.ts"},
|
|
],
|
|
"edges": [
|
|
{"source": "a_handler", "target": "z_emitter", "relation": "calls",
|
|
"confidence": "EXTRACTED", "source_file": "a.ts"},
|
|
{"source": "z_emitter", "target": "a_handler", "relation": "calls",
|
|
"confidence": "EXTRACTED", "source_file": "z.ts"},
|
|
],
|
|
"input_tokens": 0,
|
|
"output_tokens": 0,
|
|
}
|
|
G = build_from_json(extraction)
|
|
# Only one undirected edge between the pair survives, but its stored
|
|
# direction must be the first-seen one (a_handler -> z_emitter), not the
|
|
# lexicographically-later one (z_emitter -> a_handler).
|
|
assert G.number_of_edges() == 1
|
|
data = edge_data(G, "a_handler", "z_emitter")
|
|
assert data["_src"] == "a_handler"
|
|
assert data["_tgt"] == "z_emitter"
|
|
|
|
graph_path = tmp_path / "graph.json"
|
|
assert to_json(G, {}, str(graph_path), force=True)
|
|
saved = json.loads(graph_path.read_text())
|
|
saved_calls = [e for e in saved.get("links", saved.get("edges", []))
|
|
if e.get("relation") == "calls"]
|
|
assert len(saved_calls) == 1
|
|
assert saved_calls[0]["source"] == "a_handler", (
|
|
f"calls edge source flipped on bidirectional collision: "
|
|
f"expected a_handler, got {saved_calls[0]['source']}"
|
|
)
|
|
assert saved_calls[0]["target"] == "z_emitter", (
|
|
f"calls edge target flipped on bidirectional collision: "
|
|
f"expected z_emitter, got {saved_calls[0]['target']}"
|
|
)
|
|
|
|
|
|
# Regression tests for #796 — edge_data / edge_datas helpers must tolerate
|
|
# MultiGraph and MultiDiGraph, which networkx's node_link_graph() produces
|
|
# whenever the loaded JSON has multigraph: true. Plain G.edges[u, v] crashes
|
|
# on those with `ValueError: not enough values to unpack (expected 3, got 2)`.
|
|
|
|
def test_edge_data_simple_graph():
|
|
G = nx.Graph()
|
|
G.add_edge("a", "b", relation="calls", confidence="EXTRACTED")
|
|
d = edge_data(G, "a", "b")
|
|
assert isinstance(d, dict)
|
|
assert d["relation"] == "calls"
|
|
assert d["confidence"] == "EXTRACTED"
|
|
|
|
|
|
def test_edge_datas_simple_graph_returns_singleton_list():
|
|
G = nx.Graph()
|
|
G.add_edge("a", "b", relation="calls", confidence="EXTRACTED")
|
|
ds = edge_datas(G, "a", "b")
|
|
assert isinstance(ds, list)
|
|
assert len(ds) == 1
|
|
assert ds[0]["relation"] == "calls"
|
|
|
|
|
|
def test_edge_data_multigraph_with_parallel_edges():
|
|
G = nx.MultiGraph()
|
|
G.add_edge("a", "b", relation="calls", confidence="EXTRACTED")
|
|
G.add_edge("a", "b", relation="references", confidence="INFERRED")
|
|
d = edge_data(G, "a", "b")
|
|
assert isinstance(d, dict)
|
|
# First parallel edge wins; should be one of the two attribute dicts above.
|
|
assert d.get("relation") in ("calls", "references")
|
|
|
|
|
|
def test_edge_datas_multigraph_returns_all_parallel_edges():
|
|
G = nx.MultiGraph()
|
|
G.add_edge("a", "b", relation="calls", confidence="EXTRACTED")
|
|
G.add_edge("a", "b", relation="references", confidence="INFERRED")
|
|
ds = edge_datas(G, "a", "b")
|
|
assert isinstance(ds, list)
|
|
assert len(ds) == 2
|
|
relations = {e.get("relation") for e in ds}
|
|
assert relations == {"calls", "references"}
|
|
|
|
|
|
def test_edge_data_multidigraph():
|
|
G = nx.MultiDiGraph()
|
|
G.add_edge("a", "b", relation="calls")
|
|
G.add_edge("a", "b", relation="imports")
|
|
d = edge_data(G, "a", "b")
|
|
assert isinstance(d, dict)
|
|
assert d.get("relation") in ("calls", "imports")
|
|
ds = edge_datas(G, "a", "b")
|
|
assert len(ds) == 2
|
|
|
|
|
|
def test_edge_data_node_link_multigraph_roundtrip():
|
|
"""A node_link JSON with multigraph: true must load as MultiGraph and the
|
|
helpers must operate on it without raising the 3-tuple unpack ValueError."""
|
|
data = {
|
|
"directed": False,
|
|
"multigraph": True,
|
|
"graph": {},
|
|
"nodes": [
|
|
{"id": "a", "label": "A"},
|
|
{"id": "b", "label": "B"},
|
|
],
|
|
"links": [
|
|
{"source": "a", "target": "b", "relation": "calls", "confidence": "EXTRACTED"},
|
|
{"source": "a", "target": "b", "relation": "references", "confidence": "INFERRED"},
|
|
],
|
|
}
|
|
try:
|
|
G = json_graph.node_link_graph(data, edges="links")
|
|
except TypeError:
|
|
G = json_graph.node_link_graph(data)
|
|
assert isinstance(G, nx.MultiGraph)
|
|
# Plain G.edges[u, v] would raise here; the helper must not.
|
|
d = edge_data(G, "a", "b")
|
|
assert isinstance(d, dict)
|
|
assert d.get("relation") in ("calls", "references")
|
|
ds = edge_datas(G, "a", "b")
|
|
assert len(ds) == 2
|
|
|
|
|
|
def test_build_from_json_relativizes_absolute_source_file(tmp_path):
|
|
"""Semantic subagents emit absolute source_file paths; build_from_json must
|
|
relativize them to root so MCP traversal works correctly (#932)."""
|
|
root = tmp_path / "myproject"
|
|
root.mkdir()
|
|
abs_path = str(root / "docs" / "overview.md")
|
|
extraction = {
|
|
"nodes": [
|
|
{"id": "overview_intro", "label": "Intro", "source_file": abs_path, "file_type": "document"},
|
|
],
|
|
"edges": [
|
|
{"source": "overview_intro", "target": "overview_intro",
|
|
"relation": "self", "confidence": "EXTRACTED", "confidence_score": 1.0,
|
|
"source_file": abs_path},
|
|
],
|
|
}
|
|
G = build_from_json(extraction, root=root)
|
|
# The id-stem migration (#1504) re-keys the old short id to the full-path form.
|
|
sf = G.nodes["docs_overview_intro"]["source_file"]
|
|
assert not sf.startswith("/"), f"source_file still absolute: {sf}"
|
|
assert sf == "docs/overview.md"
|
|
|
|
|
|
def test_build_relativizes_absolute_source_file(tmp_path):
|
|
"""build() passes root through to build_from_json (#932)."""
|
|
root = tmp_path / "proj"
|
|
root.mkdir()
|
|
abs_path = str(root / "src" / "main.py")
|
|
extraction = {
|
|
"nodes": [{"id": "main_fn", "label": "main", "source_file": abs_path, "file_type": "code"}],
|
|
"edges": [],
|
|
}
|
|
G = build([extraction], root=root)
|
|
# #1504 re-keys main_fn (old stem "main") to the full-path form "src_main".
|
|
sf = G.nodes["src_main_fn"]["source_file"]
|
|
assert sf == "src/main.py"
|
|
|
|
|
|
def test_build_from_json_ambiguous_old_stem_alias_stays_dangling(tmp_path):
|
|
"""The #1504 old-stem alias (e.g. "ping.h" -> bare "ping") is meant to let a
|
|
stale-id edge from an un-re-extracted fragment still find its own file after
|
|
a rekey. But the old-stem form drops the extension and most of the path, so
|
|
two unrelated real files easily collapse onto the same bare alias (a C header
|
|
and a PHP script both named "ping", in different directories). A dangling
|
|
edge produced by an unrelated third file's own unscoped fallback id (e.g. the
|
|
C/C++ extractor's last-resort target for an #include it couldn't resolve to
|
|
a real path) must not silently ride that alias onto an arbitrary one of them
|
|
— it should stay dangling and get dropped, same as any other unresolvable
|
|
edge, rather than wire two unrelated files/languages together by accident."""
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
extraction = {
|
|
"nodes": [
|
|
# Ids given in their canonical (post-extract.py, extension-stripped)
|
|
# form, matching what a real graphify update run would already have
|
|
# produced before build_from_json assembles the final graph.
|
|
{"id": "dev_monitoring_ping", "label": "ping.h", "file_type": "code",
|
|
"source_file": "Dev/monitoring/ping.h"},
|
|
{"id": "www_pages_api_ping", "label": "ping.php", "file_type": "code",
|
|
"source_file": "www/pages/api/ping.php"},
|
|
{"id": "dev_poker_server", "label": "server.cpp", "file_type": "code",
|
|
"source_file": "Dev/poker/server.cpp"},
|
|
],
|
|
"edges": [
|
|
# The unscoped, deliberately-unresolved fallback edge a C/C++ #include
|
|
# resolver leaves behind when it can't find the header on disk.
|
|
{"source": "dev_poker_server", "target": "ping", "relation": "imports",
|
|
"confidence": "EXTRACTED", "source_file": "Dev/poker/server.cpp"},
|
|
],
|
|
}
|
|
G = build_from_json(extraction, root=root)
|
|
assert not G.has_edge("dev_poker_server", "dev_monitoring_ping")
|
|
assert not G.has_edge("dev_poker_server", "www_pages_api_ping")
|
|
|
|
|
|
def test_build_from_json_ambiguous_alias_detected_despite_header_impl_salting(tmp_path):
|
|
"""A same-directory .h/.cpp pair collides on their shared pre-extension id
|
|
and gets salted apart into ids like "tools_aolserver_utility_h_..." — no
|
|
longer a clean new_stem prefix. The ambiguity check must still recognize
|
|
the salted header as a legitimate claimant for the bare old-stem alias (by
|
|
label, not id shape), so a real collision with an unrelated same-named PHP
|
|
file is still caught instead of the header silently dropping out of the
|
|
race and leaving the PHP file as the lone "unambiguous" winner (this
|
|
reproduced against the real depot: Tools/aolserver/utility.h and .cpp,
|
|
salted apart, let wwwapi.masque.com/pages/utility.php win the bare
|
|
"utility" alias uncontested)."""
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
extraction = {
|
|
"nodes": [
|
|
{"id": "tools_aolserver_utility_h_tools_aolserver_utility", "label": "utility.h",
|
|
"file_type": "code", "source_file": "Tools/aolserver/utility.h"},
|
|
{"id": "tools_aolserver_utility_cpp_tools_aolserver_utility", "label": "utility.cpp",
|
|
"file_type": "code", "source_file": "Tools/aolserver/utility.cpp"},
|
|
{"id": "wwwapi_masque_com_pages_utility", "label": "utility.php",
|
|
"file_type": "code", "source_file": "wwwapi.masque.com/pages/utility.php"},
|
|
{"id": "dev_poker_server", "label": "server.cpp", "file_type": "code",
|
|
"source_file": "Dev/poker/server.cpp"},
|
|
],
|
|
"edges": [
|
|
{"source": "dev_poker_server", "target": "utility", "relation": "imports",
|
|
"confidence": "EXTRACTED", "source_file": "Dev/poker/server.cpp"},
|
|
],
|
|
}
|
|
G = build_from_json(extraction, root=root)
|
|
assert not G.has_edge("dev_poker_server", "wwwapi_masque_com_pages_utility")
|
|
assert not G.has_edge("dev_poker_server", "tools_aolserver_utility_h_tools_aolserver_utility")
|
|
|
|
|
|
def test_build_from_json_unambiguous_old_stem_alias_still_resolves(tmp_path):
|
|
"""Companion to the ambiguous case above: when exactly one real file claims
|
|
an old-stem alias, a dangling edge to that bare alias should still resolve
|
|
to it — the #1504 migration-compat behavior this index exists for."""
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
extraction = {
|
|
"nodes": [
|
|
{"id": "dev_monitoring_utility", "label": "utility.h", "file_type": "code",
|
|
"source_file": "Dev/monitoring/utility.h"},
|
|
{"id": "dev_poker_server", "label": "server.cpp", "file_type": "code",
|
|
"source_file": "Dev/poker/server.cpp"},
|
|
],
|
|
"edges": [
|
|
{"source": "dev_poker_server", "target": "utility", "relation": "imports",
|
|
"confidence": "EXTRACTED", "source_file": "Dev/poker/server.cpp"},
|
|
],
|
|
}
|
|
G = build_from_json(extraction, root=root)
|
|
assert G.has_edge("dev_poker_server", "dev_monitoring_utility")
|
|
|
|
|
|
def test_build_from_json_relative_source_file_unchanged(tmp_path):
|
|
"""Already-relative source_file paths must not be modified."""
|
|
extraction = {
|
|
"nodes": [{"id": "foo_bar", "label": "bar", "source_file": "src/foo.py", "file_type": "code"}],
|
|
"edges": [],
|
|
}
|
|
G = build_from_json(extraction, root=tmp_path)
|
|
# source_file must be untouched; the id is re-keyed to the full-path form (#1504).
|
|
assert G.nodes["src_foo_bar"]["source_file"] == "src/foo.py"
|
|
|
|
|
|
def test_build_merge_prune_absolute_paths_match_relative_nodes(tmp_path):
|
|
"""#1007: manifest stores absolute paths, graph nodes store relative paths.
|
|
prune_sources with absolute paths must still remove the right nodes and edges."""
|
|
import networkx as nx
|
|
|
|
root = tmp_path / "corpus"
|
|
root.mkdir()
|
|
graph_path = tmp_path / "graph.json"
|
|
|
|
# Simulate a graph with relative source_file paths (as built normally)
|
|
chunk = {"nodes": [
|
|
{"id": "n1", "label": "login", "file_type": "code", "source_file": "module_a/auth.py"},
|
|
{"id": "n2", "label": "format_date", "file_type": "code", "source_file": "module_b/utils.py"},
|
|
], "edges": [
|
|
{"source": "n1", "target": "n2", "relation": "calls", "confidence": "EXTRACTED",
|
|
"source_file": "module_b/utils.py", "weight": 1.0},
|
|
]}
|
|
G0 = build([chunk], dedup=False)
|
|
graph_path.write_text(json.dumps(nx.node_link_data(G0, edges="edges")), encoding="utf-8")
|
|
|
|
# prune_sources from manifest — absolute paths (what detect_incremental emits)
|
|
deleted_abs = [str(root / "module_b" / "utils.py")]
|
|
G1 = build_merge([], graph_path, prune_sources=deleted_abs, dedup=False, root=root)
|
|
|
|
node_labels = {d["label"] for _, d in G1.nodes(data=True)}
|
|
assert "format_date" not in node_labels, "stale node from deleted file should be pruned"
|
|
assert "login" in node_labels, "unrelated node must survive"
|
|
# Edge from deleted file must also be gone
|
|
assert G1.number_of_edges() == 0, "edge from deleted source_file should be pruned"
|
|
|
|
|
|
def test_build_merge_prune_windows_backslash_paths(tmp_path):
|
|
"""#1007: prune_sources with Windows-style backslash absolute paths must still match."""
|
|
import networkx as nx
|
|
|
|
root = tmp_path / "corpus"
|
|
root.mkdir()
|
|
graph_path = tmp_path / "graph.json"
|
|
|
|
chunk = {"nodes": [
|
|
{"id": "n1", "label": "parse_date", "file_type": "code", "source_file": "module_b/utils.py"},
|
|
], "edges": []}
|
|
G0 = build([chunk], dedup=False)
|
|
graph_path.write_text(json.dumps(nx.node_link_data(G0, edges="edges")), encoding="utf-8")
|
|
|
|
# Simulate Windows manifest path with backslashes
|
|
win_path = str(root / "module_b" / "utils.py").replace("/", "\\")
|
|
G1 = build_merge([], graph_path, prune_sources=[win_path], dedup=False, root=root)
|
|
|
|
node_labels = {d["label"] for _, d in G1.nodes(data=True)}
|
|
assert "parse_date" not in node_labels, "node should be pruned even with backslash path"
|
|
|
|
|
|
def test_build_merge_replaces_changed_file_stale_edges(tmp_path):
|
|
"""Re-extracting a CHANGED file must REPLACE its prior nodes/edges, not
|
|
accumulate them. build_merge previously only grew the graph, so an edge that
|
|
disappeared from a file's new version survived forever (only exact-duplicate
|
|
edges collapsed). The new-chunk source_file may be an absolute win32 path
|
|
while the stored graph keeps relative posix — both forms must match."""
|
|
import networkx as nx
|
|
|
|
root = tmp_path / "corpus"
|
|
root.mkdir()
|
|
graph_path = tmp_path / "graph.json"
|
|
|
|
# First build: changed.md contributed A, B and edge A->B; keep.md is unrelated.
|
|
chunk0 = {"nodes": [
|
|
{"id": "A", "label": "A", "file_type": "document", "source_file": "changed.md"},
|
|
{"id": "B", "label": "B", "file_type": "document", "source_file": "changed.md"},
|
|
{"id": "K", "label": "K", "file_type": "document", "source_file": "keep.md"},
|
|
], "edges": [
|
|
{"source": "A", "target": "B", "relation": "references", "confidence": "EXTRACTED",
|
|
"source_file": "changed.md", "weight": 1.0},
|
|
{"source": "K", "target": "A", "relation": "references", "confidence": "EXTRACTED",
|
|
"source_file": "keep.md", "weight": 1.0},
|
|
]}
|
|
G0 = build([chunk0], dedup=False)
|
|
graph_path.write_text(json.dumps(nx.node_link_data(G0, edges="edges")), encoding="utf-8")
|
|
|
|
# changed.md edited: re-extraction now yields A, C and edge A->C (B dropped).
|
|
# source_file arrives as an absolute win32-style path (as detect emits on Windows).
|
|
abs_changed = str(root / "changed.md").replace("/", "\\")
|
|
new_chunk = {"nodes": [
|
|
{"id": "A", "label": "A", "file_type": "document", "source_file": abs_changed},
|
|
{"id": "C", "label": "C", "file_type": "document", "source_file": abs_changed},
|
|
], "edges": [
|
|
{"source": "A", "target": "C", "relation": "references", "confidence": "EXTRACTED",
|
|
"source_file": abs_changed, "weight": 1.0},
|
|
]}
|
|
G1 = build_merge([new_chunk], graph_path, dedup=False, root=root)
|
|
|
|
labels = {d["label"] for _, d in G1.nodes(data=True)}
|
|
edges = {(u, v) for u, v in G1.edges()}
|
|
|
|
# Stale contribution from the old version of changed.md is gone.
|
|
assert "B" not in labels, "stale node from changed file's old version must be dropped"
|
|
assert ("A", "B") not in edges and ("B", "A") not in edges, "stale edge must be dropped"
|
|
# Fresh contribution is present.
|
|
assert "C" in labels, "re-extracted node must be present"
|
|
assert ("A", "C") in edges, "re-extracted edge must be present"
|
|
# An unchanged file is untouched.
|
|
assert "K" in labels, "unchanged file's node must survive"
|
|
assert ("K", "A") in edges, "unchanged file's edge must survive"
|
|
|
|
|
|
def test_build_merge_root_collapses_convention_drift(tmp_path):
|
|
"""Skill contract: the extraction subagent must emit source_file as the
|
|
verbatim path from FILE_LIST AND the caller must pass root= (the build root).
|
|
Then build_merge canonicalizes the new chunk to the same relative base as the
|
|
stored graph, so re-extraction REPLACES the prior node (incl. stale nodes for
|
|
that file) instead of accumulating a duplicate. Without root, a drifted
|
|
relative base (e.g. a bare basename from a different run) mismatches and the
|
|
graph duplicates. Engine is unchanged — this pins the prompt/root contract."""
|
|
import networkx as nx
|
|
|
|
root = tmp_path
|
|
graph_path = tmp_path / "graphify-out" / "graph.json"
|
|
graph_path.parent.mkdir(parents=True)
|
|
|
|
# Stored graph: nested project-relative convention + a STALE node for the same
|
|
# file that the re-extraction no longer emits.
|
|
stored = {"nodes": [
|
|
{"id": "wiki_overview_overview", "label": "Overview", "file_type": "document",
|
|
"source_file": "docs/wiki/overview.md"},
|
|
{"id": "wiki_overview_stale", "label": "Stale", "file_type": "document",
|
|
"source_file": "docs/wiki/overview.md"},
|
|
], "edges": []}
|
|
G0 = build([stored], dedup=False)
|
|
saved = json.dumps(nx.node_link_data(G0, edges="edges"))
|
|
graph_path.write_text(saved, encoding="utf-8")
|
|
|
|
# BUG: --update drifted to a bare basename and no root was passed. Different
|
|
# base -> source_file replace misses -> stale + duplicate both survive.
|
|
drift = {"nodes": [
|
|
{"id": "overview_overview", "label": "Overview", "file_type": "document",
|
|
"source_file": "overview.md"},
|
|
], "edges": []}
|
|
G_bug = build_merge([drift], graph_path, dedup=False)
|
|
assert G_bug.number_of_nodes() == 3, "mismatched base must NOT replace -> stale+dup remain"
|
|
|
|
# FIX: subagent emits the verbatim path; caller passes root (the build root).
|
|
graph_path.write_text(saved, encoding="utf-8")
|
|
abs_overview = str(root / "docs" / "wiki" / "overview.md")
|
|
fixed = {"nodes": [
|
|
{"id": "wiki_overview_overview", "label": "Overview", "file_type": "document",
|
|
"source_file": abs_overview},
|
|
], "edges": []}
|
|
G_ok = build_merge([fixed], graph_path, prune_sources=None, dedup=False, root=root)
|
|
assert G_ok.number_of_nodes() == 1, "verbatim path + root must collapse to one node"
|
|
# #1504 re-keys the author-chosen short ids to the canonical full-path stem.
|
|
assert "docs_wiki_overview_stale" not in G_ok, "stale node for the re-extracted file must be dropped"
|
|
assert G_ok.nodes["docs_wiki_overview_overview"]["source_file"] == "docs/wiki/overview.md", \
|
|
"new chunk must be canonicalized to the stored relative base"
|
|
|
|
|
|
def test_build_merge_rejects_oversized_existing_graph(monkeypatch, tmp_path):
|
|
"""#F4: build_merge must refuse to read an existing graph.json that
|
|
exceeds the size cap, rather than json.loads-ing it into memory."""
|
|
import pytest
|
|
|
|
graph_path = tmp_path / "graph.json"
|
|
graph_path.write_text(json.dumps({"nodes": [], "links": []}), encoding="utf-8")
|
|
monkeypatch.setattr("graphify.security._MAX_GRAPH_FILE_BYTES", 8)
|
|
with pytest.raises(ValueError, match="exceeds"):
|
|
build_merge([], graph_path, dedup=False)
|
|
|
|
|
|
def test_build_from_json_skips_non_hashable_node_id():
|
|
# A malformed LLM extraction can emit a list-valued id; build_from_json must
|
|
# skip it (NetworkX add_node would otherwise raise unhashable type) and still
|
|
# build the graph from the well-formed nodes.
|
|
extraction = {
|
|
"nodes": [
|
|
{"id": "a", "label": "A", "file_type": "code", "source_file": "a.py"},
|
|
{"id": ["x", "y"], "label": "B", "file_type": "code", "source_file": "b.py"},
|
|
{"label": "C", "file_type": "code", "source_file": "c.py"}, # missing id
|
|
],
|
|
"edges": [],
|
|
}
|
|
G = build_from_json(extraction)
|
|
assert set(G.nodes()) == {"a"}
|
|
|
|
|
|
def test_build_from_json_skips_edge_with_non_hashable_endpoint():
|
|
# A list-valued edge endpoint must be skipped rather than crash the
|
|
# `not in node_set` membership test. The well-formed edge survives.
|
|
extraction = {
|
|
"nodes": [
|
|
{"id": "a", "label": "A", "file_type": "code", "source_file": "a.py"},
|
|
{"id": "b", "label": "B", "file_type": "code", "source_file": "b.py"},
|
|
],
|
|
"edges": [
|
|
{"source": "a", "target": ["b", "c"], "relation": "calls",
|
|
"confidence": "INFERRED", "source_file": "a.py"},
|
|
{"source": "a", "target": "b", "relation": "imports",
|
|
"confidence": "EXTRACTED", "source_file": "a.py"},
|
|
],
|
|
}
|
|
G = build_from_json(extraction)
|
|
assert G.number_of_nodes() == 2
|
|
assert G.number_of_edges() == 1
|
|
assert G.has_edge("a", "b")
|
|
|
|
|
|
# ── #1504 migration: legacy-id detection + re-key source_file contract ──────────
|
|
|
|
def test_graph_has_legacy_ids_detects_old_scheme():
|
|
"""The read-only-consumer nudge (query/serve) flags a pre-#1504 graph and
|
|
leaves a canonical one alone."""
|
|
from graphify.build import graph_has_legacy_ids
|
|
old = [{"id": "api_readme", "source_file": "docs/v1/api/README.md", "type": "document", "source_location": "L1"}]
|
|
new = [{"id": "docs_v1_api_readme", "source_file": "docs/v1/api/README.md", "type": "document", "source_location": "L1"}]
|
|
assert graph_has_legacy_ids(old, root=".") is True
|
|
assert graph_has_legacy_ids(new, root=".") is False
|
|
# sourceless / top-level file nodes don't false-positive
|
|
assert graph_has_legacy_ids([{"id": "setup", "source_file": "setup.py", "source_location": "L1"}], root=".") is False
|
|
assert graph_has_legacy_ids([{"id": "x", "label": "y"}], root=".") is False
|
|
# package/dir-scoped SYMBOL ids (Go's _make_id(pkg_dir, name) -> "sub_thing") must
|
|
# NOT false-positive: not file-level (no L1), so ignored even though "sub_thing"
|
|
# coincides with the old file-stem form of pkg/sub/thing.go.
|
|
go_symbol = [{"id": "sub_thing", "source_file": "pkg/sub/thing.go", "type": "code", "source_location": "L3"}]
|
|
assert graph_has_legacy_ids(go_symbol, root=".") is False
|
|
|
|
|
|
def test_semantic_rekey_relative_vs_absolute_source_file():
|
|
"""Re-key contract: a relative source_file is migrated; an absolute one is left
|
|
untouched (it can't be relativized, so its on-disk path must not leak into IDs)."""
|
|
from graphify.build import _semantic_id_remap
|
|
rel = [{"id": "api_readme", "source_file": "docs/v1/api/README.md", "type": "document"}]
|
|
assert _semantic_id_remap(rel, ".") == {"api_readme": "docs_v1_api_readme"}
|
|
# absolute path with no resolvable root → skipped, not remapped to an abs-path id
|
|
ab = [{"id": "api_readme", "source_file": "/abs/docs/v1/api/README.md", "type": "document"}]
|
|
assert _semantic_id_remap(ab, None) == {}
|
|
|
|
|
|
def test_cross_language_imports_references_are_dropped():
|
|
"""#1749: an `imports`/`references` edge must not bind across a language
|
|
family. A Python `import time` that resolved by bare stem onto a `time.ts`
|
|
file node welds the two language halves together at a phantom edge; the spec
|
|
forbids this for `calls` and it is equally invalid here."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "backend_worker_py", "label": "worker.py", "file_type": "code",
|
|
"source_file": "backend/worker.py", "source_location": "L1", "_origin": "ast"},
|
|
{"id": "src_time_ts", "label": "time.ts", "file_type": "code",
|
|
"source_file": "src/time.ts", "source_location": "L1", "_origin": "ast"},
|
|
{"id": "src_util_ts", "label": "util.ts", "file_type": "code",
|
|
"source_file": "src/util.ts", "source_location": "L1", "_origin": "ast"},
|
|
],
|
|
"edges": [
|
|
# phantom: Python file importing a TS file (cross-language)
|
|
{"source": "backend_worker_py", "target": "src_time_ts", "relation": "imports",
|
|
"confidence": "EXTRACTED", "source_file": "backend/worker.py", "weight": 1.0},
|
|
# legit: TS importing TS (same family) must survive
|
|
{"source": "src_time_ts", "target": "src_util_ts", "relation": "imports",
|
|
"confidence": "EXTRACTED", "source_file": "src/time.ts", "weight": 1.0},
|
|
],
|
|
}
|
|
G = build_from_json(ext, directed=False)
|
|
assert not G.has_edge("backend_worker_py", "src_time_ts"), "cross-language import must be dropped"
|
|
assert G.has_edge("src_time_ts", "src_util_ts"), "same-family (TS->TS) import must survive"
|
|
|
|
|
|
def test_cross_family_reference_to_unknown_ext_is_kept():
|
|
"""The #1749 guard only drops when BOTH endpoints are known code languages,
|
|
so a reference from a config/manifest (unknown ext) to a code file is kept."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "pkg_json", "label": "package.json", "file_type": "code",
|
|
"source_file": "package.json", "source_location": "L1", "_origin": "ast"},
|
|
{"id": "src_app_ts", "label": "app.ts", "file_type": "code",
|
|
"source_file": "src/app.ts", "source_location": "L1", "_origin": "ast"},
|
|
],
|
|
"edges": [
|
|
{"source": "pkg_json", "target": "src_app_ts", "relation": "references",
|
|
"confidence": "EXTRACTED", "source_file": "package.json", "weight": 1.0},
|
|
],
|
|
}
|
|
G = build_from_json(ext, directed=False)
|
|
assert G.has_edge("pkg_json", "src_app_ts"), "config->code reference (unknown ext) must be kept"
|
|
|
|
|
|
def test_markdown_doc_twin_merges_into_semantic_doc_node():
|
|
"""#1799: the markdown quick-scan's bare `<slug>` doc node and the semantic
|
|
`<slug>_doc` node for the same file must collapse to one node, with edges
|
|
consolidated — otherwise a document is two disconnected halves and traversals
|
|
dead-end on the wrong twin."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "docs_readme_doc", "label": "README", "file_type": "document",
|
|
"source_file": "docs/readme.md", "source_location": "L1"},
|
|
{"id": "docs_readme", "label": "readme.md", "file_type": "document",
|
|
"source_file": "docs/readme.md", "source_location": "L1"},
|
|
{"id": "code_auth", "label": "auth", "file_type": "code",
|
|
"source_file": "auth.py", "source_location": "L1"},
|
|
{"id": "docs_guide", "label": "guide.md", "file_type": "document",
|
|
"source_file": "docs/guide.md", "source_location": "L1"},
|
|
],
|
|
"edges": [
|
|
{"source": "docs_readme_doc", "target": "code_auth", "relation": "references",
|
|
"source_file": "docs/readme.md", "confidence": "INFERRED", "weight": 1.0},
|
|
{"source": "docs_guide", "target": "docs_readme", "relation": "references",
|
|
"source_file": "docs/guide.md", "confidence": "EXTRACTED", "weight": 1.0},
|
|
],
|
|
}
|
|
G = build_from_json(ext, directed=False)
|
|
assert "docs_readme" not in G.nodes() # bare twin merged away
|
|
assert "docs_readme_doc" in G.nodes() # semantic node is canonical
|
|
assert G.has_edge("docs_guide", "docs_readme_doc") # quick-scan edge repointed
|
|
assert G.has_edge("docs_readme_doc", "code_auth") # semantic edge kept
|
|
|
|
|
|
def test_doc_twin_merge_does_not_touch_code_symbols():
|
|
"""#1799 guard: a code symbol `foo` and an unrelated `foo_doc` (not
|
|
file_type=document) must NOT merge, even sharing a source_file."""
|
|
ext = {
|
|
"nodes": [
|
|
{"id": "m_foo", "label": "foo", "file_type": "code",
|
|
"source_file": "m.py", "source_location": "L1"},
|
|
{"id": "m_foo_doc", "label": "foo rationale", "file_type": "rationale",
|
|
"source_file": "m.py", "source_location": "L2"},
|
|
],
|
|
"edges": [],
|
|
}
|
|
G = build_from_json(ext, directed=False)
|
|
assert {"m_foo", "m_foo_doc"} <= set(G.nodes())
|