tests/test_export_direction.py
Rama v8 · réplica de lectura del repositorio original.
"""Reloading graph.json must keep each edge's stored direction (#4066).
graph.json is written ``directed: false`` and carries true direction in each
link's arc order (#563), or in ``_src``/``_tgt`` markers on legacy
canonicalized files (#2309). A plain undirected ``node_link_graph`` load
re-orders each edge's endpoints by node-list position, so a link whose target
node precedes its source node comes back reversed.
query/path/explain (#2309, #2487), affected (#1174), merge-graphs (#2261) and
the MCP server already restore direction. These tests cover the remaining
reloaders: the export subcommands, the watch/update HTML re-render, the git
merge driver, which writes the committed graph.json, and `global add`, which
writes ~/.graphify/global-graph.json.
"""
from __future__ import annotations
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from unittest.mock import patch
PYTHON = sys.executable
# A fixed positive limit, so HTML rendering never depends on the caller's
# shell (GRAPHIFY_VIZ_NODE_LIMIT=0 disables it, e.g. on CI runners).
_VIZ_ENV = {"GRAPHIFY_VIZ_NODE_LIMIT": "5000"}
def _graph_json(tmp_path: Path) -> Path:
"""The `graphify extract` shape: undirected flag, direction in arc order.
``callee`` precedes ``caller`` in the node list, the exact condition under
which an undirected load flips the edge. ``helper`` -> ``util`` is listed
source-first, so it survives either way and guards against over-correction.
"""
out = tmp_path / "graphify-out"
out.mkdir()
data = {
"directed": False,
"multigraph": False,
"graph": {},
"nodes": [
{"id": "callee", "label": "callee()", "file_type": "code", "source_file": "a.py", "community": 0},
{"id": "caller", "label": "caller()", "file_type": "code", "source_file": "a.py", "community": 0},
{"id": "helper", "label": "helper()", "file_type": "code", "source_file": "a.py", "community": 0},
{"id": "util", "label": "util()", "file_type": "code", "source_file": "a.py", "community": 0},
# legacy canonicalized link: arc flipped, markers carry the truth
{"id": "legacy_b", "label": "legacy_b()", "file_type": "code", "source_file": "a.py", "community": 0},
{"id": "legacy_a", "label": "legacy_a()", "file_type": "code", "source_file": "a.py", "community": 0},
],
"links": [
{"source": "caller", "target": "callee", "relation": "calls", "confidence": "EXTRACTED"},
{"source": "helper", "target": "util", "relation": "calls", "confidence": "EXTRACTED"},
{"source": "legacy_b", "target": "legacy_a", "_src": "legacy_a", "_tgt": "legacy_b",
"relation": "calls", "confidence": "EXTRACTED"},
],
}
p = out / "graph.json"
p.write_text(json.dumps(data))
(out / ".graphify_analysis.json").write_text(json.dumps(
{"communities": {"0": ["callee", "caller", "helper", "util", "legacy_a", "legacy_b"]}}))
return p
TRUE_ARCS = {("caller", "callee"), ("helper", "util"), ("legacy_a", "legacy_b")}
def _run(args: list[str], cwd: Path, env: dict[str, str] | None = None) -> subprocess.CompletedProcess:
env = dict(env if env is not None else os.environ, GRAPHIFY_NO_AUTO_REFRESH="1", **_VIZ_ENV)
return subprocess.run([PYTHON, "-m", "graphify", *args], cwd=cwd,
capture_output=True, text=True, env=env)
def _html_arcs(html_path: Path) -> set[tuple[str, str]]:
html = html_path.read_text(encoding="utf-8")
return set(re.findall(r'"from": "([^"]+)", "to": "([^"]+)"', html))
def test_load_node_link_graph_stamps_stored_direction(tmp_path):
from graphify.paths import load_node_link_graph
G = load_node_link_graph(_graph_json(tmp_path))
arcs = {(d.get("_src", u), d.get("_tgt", v)) for u, v, d in G.edges(data=True)}
assert arcs == TRUE_ARCS
def test_export_html_draws_arrows_in_stored_direction(tmp_path):
_graph_json(tmp_path)
r = _run(["export", "html"], tmp_path)
assert r.returncode == 0, r.stderr
assert _html_arcs(tmp_path / "graphify-out" / "graph.html") == TRUE_ARCS
def test_export_cypher_merges_edges_in_stored_direction(tmp_path):
_graph_json(tmp_path)
r = _run(["export", "neo4j"], tmp_path)
assert r.returncode == 0, r.stderr
cypher = (tmp_path / "graphify-out" / "cypher.txt").read_text(encoding="utf-8")
arcs = set(re.findall(r"MATCH \(a \{id: '([^']+)'\}\), \(b \{id: '([^']+)'\}\)", cypher))
assert arcs == TRUE_ARCS
# Fake SDKs, imported by the subprocess through PYTHONPATH, that log the
# (src, tgt) of every relationship MERGE instead of talking to a database.
_FAKE_NEO4J = '''
import json, os
class _S:
def __enter__(self): return self
def __exit__(self, *a): return False
def run(self, q, **p):
if "src" in p:
with open(os.environ["FAKE_ARC_LOG"], "a") as f:
f.write(json.dumps([p["src"], p["tgt"]]) + "\\n")
class _D:
def session(self, **k): return _S()
def close(self): pass
class GraphDatabase:
@staticmethod
def driver(*a, **k): return _D()
'''
_FAKE_FALKORDB = '''
import json, os
class _G:
def query(self, q, params=None):
if params and "src" in params:
with open(os.environ["FAKE_ARC_LOG"], "a") as f:
f.write(json.dumps([params["src"], params["tgt"]]) + "\\n")
class FalkorDB:
def __init__(self, *a, **k): pass
def select_graph(self, name): return _G()
'''
def _push_arcs(tmp_path: Path, backend: str) -> set[tuple[str, str]]:
fakes = tmp_path / "fakes"
(fakes / backend).mkdir(parents=True)
(fakes / backend / "__init__.py").write_text(
_FAKE_NEO4J if backend == "neo4j" else _FAKE_FALKORDB)
log = tmp_path / "arcs.log"
env = {k: v for k, v in os.environ.items() if k not in ("NEO4J_PASSWORD", "FALKORDB_PASSWORD")}
env["PYTHONPATH"] = os.pathsep.join([str(fakes), env.get("PYTHONPATH", "")])
env["FAKE_ARC_LOG"] = str(log)
r = _run(["export", backend, "--push", "bolt://localhost:7687", "--password", "x"], tmp_path, env)
assert r.returncode == 0, r.stderr
return {tuple(json.loads(line)) for line in log.read_text().splitlines()}
def test_export_neo4j_push_sends_stored_direction(tmp_path):
_graph_json(tmp_path)
assert _push_arcs(tmp_path, "neo4j") == TRUE_ARCS
def test_export_falkordb_push_sends_stored_direction(tmp_path):
_graph_json(tmp_path)
assert _push_arcs(tmp_path, "falkordb") == TRUE_ARCS
def test_export_obsidian_canvas_edges_in_stored_direction(tmp_path):
"""JSON Canvas draws the arrowhead at ``toNode`` by default."""
_graph_json(tmp_path)
r = _run(["export", "obsidian"], tmp_path)
assert r.returncode == 0, r.stderr
canvas = json.loads((tmp_path / "graphify-out" / "obsidian" / "graph.canvas").read_text())
arcs = {(e["fromNode"][2:], e["toNode"][2:]) for e in canvas["edges"]}
assert arcs == TRUE_ARCS
def test_watch_html_reconcile_draws_stored_direction(tmp_path, monkeypatch):
"""`graphify update`, `watch` and the git hooks re-render graph.html here."""
from graphify.watch import _reconcile_graph_html
for key, value in _VIZ_ENV.items():
monkeypatch.setenv(key, value)
gp = _graph_json(tmp_path)
assert _reconcile_graph_html(gp.parent, json.loads(gp.read_text())) == "rendered"
assert _html_arcs(gp.parent / "graph.html") == TRUE_ARCS
def test_merge_driver_keeps_stored_direction_in_arc_order(tmp_path):
"""Even a merge of three identical copies used to rewrite graph.json reversed."""
gp = _graph_json(tmp_path)
base, cur, oth = (tmp_path / n for n in ("base.json", "cur.json", "oth.json"))
for p in (base, cur, oth):
p.write_text(gp.read_text())
r = _run(["merge-driver", str(base), str(cur), str(oth)], tmp_path)
assert r.returncode == 0, r.stderr
merged = json.loads(cur.read_text())
# Arc order is the direction (what to_json writes); no markers left behind.
assert {(l["source"], l["target"]) for l in merged["links"]} == TRUE_ARCS
assert not any("_src" in l or "_tgt" in l for l in merged["links"])
def _global_add(tmp_path: Path, *repos: tuple[Path, str]) -> dict:
"""Run `global add` for each (graph.json, tag) and return the saved global graph."""
from graphify.global_graph import global_add
gdir = tmp_path / ".graphify"
with patch("graphify.global_graph._GLOBAL_DIR", gdir), \
patch("graphify.global_graph._GLOBAL_GRAPH", gdir / "global-graph.json"), \
patch("graphify.global_graph._GLOBAL_MANIFEST", gdir / "global-manifest.json"):
for graph_path, tag in repos:
global_add(graph_path, tag)
return json.loads((gdir / "global-graph.json").read_text())
def _marker_arcs(data: dict) -> set[tuple[str, str]]:
return {(l.get("_src", l["source"]), l.get("_tgt", l["target"])) for l in data["links"]}
def test_global_add_persists_stored_direction(tmp_path):
saved = _global_add(tmp_path, (_graph_json(tmp_path), "proj"))
assert _marker_arcs(saved) == {(f"proj::{s}", f"proj::{t}") for s, t in TRUE_ARCS}
def test_global_add_dedup_remaps_direction_markers(tmp_path):
"""Two repos share an external `requests` node. B's copy is deduplicated
onto A::requests, so the markers on B's edge must follow it there."""
def repo(name: str, fn: str) -> Path:
# The external node is listed first, the order an undirected load flips.
data = {"directed": False, "multigraph": False, "graph": {},
"nodes": [{"id": "requests", "label": "requests"},
{"id": fn, "label": f"{fn}()", "source_file": f"{name}/x.py"}],
"links": [{"source": fn, "target": "requests", "relation": "calls"}]}
p = tmp_path / name / "graph.json"
p.parent.mkdir()
p.write_text(json.dumps(data))
return p
saved = _global_add(tmp_path, (repo("A", "fa"), "A"), (repo("B", "fb"), "B"))
arcs = _marker_arcs(saved)
assert arcs == {("A::fa", "A::requests"), ("B::fb", "A::requests")}
ids = {n["id"] for n in saved["nodes"]}
assert all(end in ids for arc in arcs for end in arc)