tests/test_dedup_shrink_refuses_force_write.py
Rama v8 · réplica de lectura del repositorio original.
"""A complete `graphify extract` must not force-write a shrink that dedup caused (#3774).
A full build sets force=True, and to_json then skips the #479 node-count check.
Dedup can merge a same-file duplicate and replace graph.json with no message.
A deleted file is a different shrink: that write still happens.
`--allow-dedup-shrink` is the only flag that accepts the dedup shrink.
"""
from __future__ import annotations
import json
import pytest
import graphify.__main__ as mainmod
def _notes(corpus):
readme = str(corpus / "README.md")
return [
{
"id": "readme_decisions_a",
"label": "Decisions",
"file_type": "document",
"source_file": readme,
"source_location": "L1",
"_origin": "semantic",
},
{
"id": "readme_decisions_b",
"label": "Decisions",
"file_type": "document",
"source_file": readme,
"source_location": "L2",
"_origin": "semantic",
},
{
"id": "readme_overview",
"label": "Overview",
"file_type": "document",
"source_file": readme,
"source_location": "L3",
"_origin": "semantic",
},
]
def _write_graph(path, nodes):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps({"nodes": nodes, "links": []}),
encoding="utf-8",
)
def _node_ids(path):
data = json.loads(path.read_text(encoding="utf-8"))
return [n["id"] for n in data["nodes"]]
def _corpus(tmp_path):
(tmp_path / "README.md").write_text("# Notes\nThe entry point overview.\n", encoding="utf-8")
(tmp_path / "GUIDE.md").write_text("# Guide\nHow to use the thing.\n", encoding="utf-8")
return tmp_path
def _arm(monkeypatch, tmp_path, *, nodes, extra_argv=()):
corpus = _corpus(tmp_path)
out_dir = tmp_path / "out"
graph_path = out_dir / "graphify-out" / "graph.json"
_write_graph(graph_path, nodes)
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-fake-key")
def _stub_corpus(paths, **kwargs):
on_chunk = kwargs.get("on_chunk_done")
if on_chunk:
on_chunk(0, 1, {"nodes": [], "edges": [], "hyperedges": []})
return {
"nodes": nodes,
"edges": [],
"hyperedges": [],
"input_tokens": 10,
"output_tokens": 5,
}
monkeypatch.setattr("graphify.llm.extract_corpus_parallel", _stub_corpus)
monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None)
monkeypatch.setattr(
mainmod.sys, "argv",
["graphify", "extract", str(corpus), "--backend", "claude",
"--out", str(out_dir), *extra_argv],
)
return graph_path
def _run():
try:
mainmod.main()
except SystemExit as exc:
return exc.code
return 0
def test_complete_run_refuses_a_same_file_dedup_shrink(monkeypatch, tmp_path, capsys):
"""Two same-file duplicates plus one keeper. Dedup merges the pair.
The file on disk must keep all three nodes."""
graph_path = _arm(monkeypatch, tmp_path, nodes=_notes(tmp_path))
code = _run()
err = capsys.readouterr().err
assert code not in (None, 0)
assert _node_ids(graph_path) == [
"readme_decisions_a",
"readme_decisions_b",
"readme_overview",
]
assert "3" in err and "2" in err
# the message attributes the shrink to the actual dedup contribution
# (1 merged node), not the whole 3->2 delta
assert "merged 1" in err
assert "--allow-dedup-shrink" in err
def test_allow_dedup_shrink_writes_the_smaller_graph(monkeypatch, tmp_path):
graph_path = _arm(
monkeypatch, tmp_path, nodes=_notes(tmp_path),
extra_argv=["--allow-dedup-shrink"],
)
code = _run()
ids = _node_ids(graph_path)
assert code in (None, 0)
assert len(ids) == 2
assert "readme_overview" in ids
assert not {"readme_decisions_a", "readme_decisions_b"} <= set(ids)
raw = graph_path.read_text(encoding="utf-8")
assert "_dedup_collapsed" not in raw
assert "_pruned_node_count" not in raw
def test_a_deleted_file_still_writes(monkeypatch, tmp_path):
"""The shrink is one missing file. Dedup merges nothing. The write proceeds."""
corpus = tmp_path
readme = str(corpus / "README.md")
gone = str(corpus / "gone.md")
nodes = [
{
"id": "readme_overview",
"label": "Overview",
"file_type": "document",
"source_file": readme,
"source_location": "L1",
"_origin": "semantic",
},
{
"id": "gone_notes",
"label": "Gone notes",
"file_type": "document",
"source_file": gone,
"source_location": "L1",
"_origin": "semantic",
},
]
fresh = [nodes[0]]
graph_path = _arm(monkeypatch, tmp_path, nodes=fresh)
# Replace the seeded graph with both files. The stub still returns only README.
_write_graph(graph_path, nodes)
code = _run()
ids = _node_ids(graph_path)
assert code in (None, 0)
assert "gone_notes" not in ids
assert "readme_overview" in ids
def test_to_json_does_not_store_shrink_accounting(tmp_path):
"""build() records the counters on the graph object. to_json must not
write them. Callers other than extract follow that path (#3774)."""
from graphify.build import build
from graphify.export import to_json
G = build([{"nodes": _notes(tmp_path), "edges": [], "hyperedges": []}])
assert G.graph.get("_dedup_collapsed") == 1
out = tmp_path / "graph.json"
assert to_json(G, {}, str(out), force=True) is True
raw = out.read_text(encoding="utf-8")
assert "_dedup_collapsed" not in raw
assert "_pruned_node_count" not in raw
def test_pruned_node_count_ignores_a_stub_created_this_run(tmp_path):
"""A source-less stub minted this run and then orphaned by the prune is
not a saved node. It must not excuse a dedup shrink."""
from graphify.build import build_merge
root = tmp_path / "repo"
root.mkdir()
kept = "README.md"
gone = "gone.md"
disk = [
{"id": "kept", "label": "Overview", "file_type": "document", "source_file": kept},
{"id": "gone", "label": "Gone", "file_type": "document", "source_file": gone},
]
graph_path = root / "graphify-out" / "graph.json"
_write_graph(graph_path, disk)
stub_edge = {"source": "gone", "target": "fresh_stub", "source_file": gone}
chunk = {
"nodes": [
{"id": "kept", "label": "Overview", "file_type": "document", "source_file": kept},
{"id": "fresh_stub", "label": "Path", "file_type": "code"},
],
"edges": [stub_edge],
"hyperedges": [],
}
G = build_merge([chunk], graph_path, prune_sources=[gone], dedup=False, root=root)
assert "gone" not in G
assert "fresh_stub" not in G
assert G.graph.get("_pruned_node_count") == 1