Navegador

feat: text-based graph export/import in JSONL format Deterministic JSONL export with sorted nodes/edges for clean git diffs. Round-trip import recreates the full graph. CLI: navegador export/import. Closes #25

lmata 2026-03-23 04:54 trunk

Commit 57ba118128bc1f74408cc365e81b50ba643e0eb69de55b617ad465443bb9bdf6

Parent c4536b2a00c4b40…

4 files changed +42 +179 +56 +252

~ navegador/cli/commands.py + navegador/graph/export.py ~ tests/test_cli.py + tests/test_export.py

M navegador/cli/commands.py

+42

		--- navegador/cli/commands.py
		+++ navegador/cli/commands.py
		@@ -643,10 +643,52 @@
643	643	table.add_column("Count", justify="right", style="green")
644	644	for k, v in stats.items():
645	645	table.add_row(k.capitalize(), str(v))
646	646	console.print(table)
647	647
	648	+
	649	+# ── Export / Import ──────────────────────────────────────────────────────────
	650	+
	651	+
	652	+@main.command("export")
	653	+@click.argument("output", type=click.Path())
	654	+@DB_OPTION
	655	+@click.option("--json", "as_json", is_flag=True, help="Output stats as JSON.")
	656	+def export_cmd(output: str, db: str, as_json: bool):
	657	+ """Export the graph to a text-based JSONL file (git-friendly)."""
	658	+ from navegador.graph.export import export_graph
	659	+
	660	+ store = _get_store(db)
	661	+ stats = export_graph(store, output)
	662	+
	663	+ if as_json:
	664	+ click.echo(json.dumps(stats, indent=2))
	665	+ else:
	666	+ console.print(
	667	+ f"[green]Exported[/green] {stats['nodes']} nodes, {stats['edges']} edges → {output}"
	668	+ )
	669	+
	670	+
	671	+@main.command("import")
	672	+@click.argument("input_path", type=click.Path(exists=True))
	673	+@DB_OPTION
	674	+@click.option("--no-clear", is_flag=True, help="Don't wipe graph before importing.")
	675	+@click.option("--json", "as_json", is_flag=True, help="Output stats as JSON.")
	676	+def import_cmd(input_path: str, db: str, no_clear: bool, as_json: bool):
	677	+ """Import a graph from a JSONL export file."""
	678	+ from navegador.graph.export import import_graph
	679	+
	680	+ store = _get_store(db)
	681	+ stats = import_graph(store, input_path, clear=not no_clear)
	682	+
	683	+ if as_json:
	684	+ click.echo(json.dumps(stats, indent=2))
	685	+ else:
	686	+ console.print(
	687	+ f"[green]Imported[/green] {stats['nodes']} nodes, {stats['edges']} edges ← {input_path}"
	688	+ )
	689	+
648	690
649	691	# ── Schema migrations ────────────────────────────────────────────────────────
650	692
651	693
652	694	@main.command()
653	695
654	696	ADDED navegador/graph/export.py

	--- navegador/cli/commands.py
	+++ navegador/cli/commands.py
	@@ -643,10 +643,52 @@
643	table.add_column("Count", justify="right", style="green")
644	for k, v in stats.items():
645	table.add_row(k.capitalize(), str(v))
646	console.print(table)
647










































648
649	# ── Schema migrations ────────────────────────────────────────────────────────
650
651
652	@main.command()
653
654	DDED navegador/graph/export.py

	--- navegador/cli/commands.py
	+++ navegador/cli/commands.py
	@@ -643,10 +643,52 @@
643	table.add_column("Count", justify="right", style="green")
644	for k, v in stats.items():
645	table.add_row(k.capitalize(), str(v))
646	console.print(table)
647
648
649	# ── Export / Import ──────────────────────────────────────────────────────────
650
651
652	@main.command("export")
653	@click.argument("output", type=click.Path())
654	@DB_OPTION
655	@click.option("--json", "as_json", is_flag=True, help="Output stats as JSON.")
656	def export_cmd(output: str, db: str, as_json: bool):
657	"""Export the graph to a text-based JSONL file (git-friendly)."""
658	from navegador.graph.export import export_graph
659
660	store = _get_store(db)
661	stats = export_graph(store, output)
662
663	if as_json:
664	click.echo(json.dumps(stats, indent=2))
665	else:
666	console.print(
667	f"[green]Exported[/green] {stats['nodes']} nodes, {stats['edges']} edges → {output}"
668	)
669
670
671	@main.command("import")
672	@click.argument("input_path", type=click.Path(exists=True))
673	@DB_OPTION
674	@click.option("--no-clear", is_flag=True, help="Don't wipe graph before importing.")
675	@click.option("--json", "as_json", is_flag=True, help="Output stats as JSON.")
676	def import_cmd(input_path: str, db: str, no_clear: bool, as_json: bool):
677	"""Import a graph from a JSONL export file."""
678	from navegador.graph.export import import_graph
679
680	store = _get_store(db)
681	stats = import_graph(store, input_path, clear=not no_clear)
682
683	if as_json:
684	click.echo(json.dumps(stats, indent=2))
685	else:
686	console.print(
687	f"[green]Imported[/green] {stats['nodes']} nodes, {stats['edges']} edges ← {input_path}"
688	)
689
690
691	# ── Schema migrations ────────────────────────────────────────────────────────
692
693
694	@main.command()
695
696	DDED navegador/graph/export.py

A navegador/graph/export.py

+179

		--- a/navegador/graph/export.py
		+++ b/navegador/graph/export.py
		@@ -0,0 +1,179 @@
	1	+"""
	2	+Text-based graph export and import for navegador.
	3	+
	4	+Exports the full graph to a deterministic JSON Lines format (.jsonl)
	5	+suitable for committing to version control. Each line is a self-contained
	6	+JSON object — either a node or an edge.
	7	+
	8	+Format:
	9	+ {"kind": "node", "label": "Function", "props": {"name": "foo", ...}}
	10	+ {"kind": "edge", "type": "CALLS", "from": {...}, "to": {...}}
	11	+"""
	12	+
	13	+import json
	14	+import logging
	15	+from pathlib import Path
	16	+
	17	+from navegador.graph.store import GraphStore
	18	+
	19	+logger = logging.getLogger(__name__)
	20	+
	21	+
	22	+def export_graph(store: GraphStore, output_path: str \| Path) -> dict[str, int]:
	23	+ """
	24	+ Export the full graph to a JSONL file.
	25	+
	26	+ Returns:
	27	+ Dict with counts: nodes, edges.
	28	+ """
	29	+ output_path = Path(output_path)
	30	+ output_path.parent.mkdir(parents=True, exist_ok=True)
	31	+
	32	+ nodes = _export_nodes(store)
	33	+ edges = _export_edges(store)
	34	+
	35	+ # Sort for deterministic output
	36	+ nodes.sort(key=lambda n: (n["label"], json.dumps(n["props"], sort_kkey=lambda e: (e["type"], e["type"],
	37	+ th)
	38	+ json.dum))
	39	+
	40	+ with output_path.open("w", encoding="utf-8") as f:
	41	+ for node in nodes:
	42	+ f.write(json.dumps(node, sort_keys=True) + "\n")
	43	+ for edge in edges:
	44	+ f.write(json.dumps(edge, sort_keys=True) + "\n")
	45	+
	46	+ logger.info("Exported %d nodes, %d edges to %s", len(nodes), len(edges), output_path)
	47	+ return {"nodes": len(nodes), "edges": len(edges)}
	48	+
	49	+
	50	+def import_graph(store: GraphStore, input_path: str \| Path, clear: bool = True) -> dict[str, int]:
	51	+ """
	52	+ Import a graph from a JSONL file.
	53	+
	54	+ Args:
	55	+ store: Target GraphStore.
	56	+ input_path: Path to the JSONL file.
	57	+ clear: If True (default), wipe the graph before importing.
	58	+
	59	+ Returns:
	60	+ Dict with counts: nodes, edges.
	61	+ """
	62	+ input_path = Path(input_path)
	63	+ if not input_path.exists():
	64	+ raise FileNotFoundError(f"Export file not found: {input_path}")
	65	+
	66	+ if clear:
	67	+ store.clear()
	68	+
	69	+ node_count = 0
	70	+ edge_count = 0
	71	+
	72	+ with input_path.open("r", encoding="utf-8") as f:
	73	+ for line in f:
	74	+ line = line.strip()
	75	+ if not line:
	76	+ continue
	77	+ record = json.loads(line)
	78	+
	79	+ if record["kind"] == "node":
	80	+ _import_node(store, record)
	81	+ node_count += 1
	82	+ elif record["kind"] == "edge":
	83	+ _import_edge(store, record)
	84	+ edge_count += 1
	85	+
	86	+ logger.info("Imported %d nodes, %d edges from %s", node_count, edge_count, input_path)
	87	+ return {"nodes": node_count, "edges": edge_count}
	88	+
	89	+
	90	+def _export_nodes(store: GraphStore) -> list[dict]:
	91	+ """Export all nodes with their labels and properties."""
	92	+ result = store.query(
	93	+ _keys=True),
	94	+ json.dumps(e["to"], sort_keys=True),
	95	+
	96	+ )
	97	+ )
	98	+
	99	+ with output_path.open("w", encoding="utf-8") as f:
	100	+ for node in nodes:
	101	+ f.write(json.dumps(node, sort_keys=True) + "\n")
	102	+ for edge in edges:
	103	+ f.write(json.dumps(edge, sort_keys=True) + "\n")
	104	+
	105	+ logger.info("Exported %d nodes, %d edges to %s", len(nodes), len(edges), output_path)
	106	+ return {"nodes": len(nodes), "edges": len(edges)}
	107	+
	108	+
	109	+def import_graph(store: GraphStore, input_path: str \| Path, clear: bool = True) -> dict[str, int]:
	110	+ """
	111	+ Import a graph from a JSONL file.
	112	+
	113	+ Args:
	114	+ store: Target GraphStore.
	115	+ input_path: Path to the JSONL file.
	116	+ clear: If True (default), wipe the graph before importing.
	117	+
	118	+ Returns:
	119	+ Dict with counts: no{
	120	+th)
	121	+ d import for navegador.
	122	+
	123	+Exports"""
	124	+Text-based graph export aise FileNotFoundError(f"Export file not found: {input_path}")
	125	+
	126	+ if r:
	127	+ store.clear()
	128	+
	129	+ node_count = 0
	130	+ edge_count = 0
	131	+
	132	+}tf-8") as f:
	133	+ for line in f:
	134	+ line = line.strip()
	135	+ if not line:
	136	+ continue
	137	+ record = json.loads(line)
	138	+
	139	+ if record["kind"] == "node":
	140	+ _import_node(store, record)
	141	+ node_count += 1
	142	+ elif record["kind"] == "edge":
	143	+ _import_edge(store, record)
	144	+ edge_count += 1
	145	+
	146	+ logger.info("Imported %d nodes, %d edges from %s", node_count, edge_count, input_path)
	147	+ return {"nodes": node_count, "edges": edge_count}
	148	+
	149	+
	150	+def _export_nodes(store: GraphStore) -> list[dict]:
	151	+ """Export all nodes with their labels and properties."""
	152	+ result = store.query("MATCH (n) RETURN labels(n)[0] AS label, properties(n) AS props")
	153	+ nodes = []
	154	+ for row in result.result_set or []:
	155	+ label = row[0]
	156	+ props = row[1] if isinstance(row[1], dict) else {}
	157	+ nodes.append({"kind": "node", "label": lafcord["to"]
	158	+
	159	+ from_key = "namef"name: $to_name"
	160	+
	161	+ params = {
	162	+ "from_name": from_info["name"],
	163	+ "to_name": to_info["name"],
	164	+ }
	165	+
	166	+ if from_info.get("path"):
	167	+ from_key += ", file_path: $from_path"
	168	+ params["from_path"] = from_info["path"]
	169	+
	170	+ if to_info.get("path"):
	171	+ to_key += ", path: $to_path"
	172	+ params["to_path"] = to_info["path"]
	173	+
	174	+ cypher = (
	175	+ f"MATCH (a:{from_info['label']} {{{from_key}}}), "
	176	+ f"(b:{to_info['label']} {{{to_key}}}) "
	177	+ f"MERGE (a)-[r:{edge_type}]->(b)"
	178	+ )
	179	+ s

	--- a/navegador/graph/export.py
	+++ b/navegador/graph/export.py
	@@ -0,0 +1,179 @@

	--- a/navegador/graph/export.py
	+++ b/navegador/graph/export.py
	@@ -0,0 +1,179 @@
1	"""
2	Text-based graph export and import for navegador.
3
4	Exports the full graph to a deterministic JSON Lines format (.jsonl)
5	suitable for committing to version control. Each line is a self-contained
6	JSON object — either a node or an edge.
7
8	Format:
9	{"kind": "node", "label": "Function", "props": {"name": "foo", ...}}
10	{"kind": "edge", "type": "CALLS", "from": {...}, "to": {...}}
11	"""
12
13	import json
14	import logging
15	from pathlib import Path
16
17	from navegador.graph.store import GraphStore
18
19	logger = logging.getLogger(__name__)
20
21
22	def export_graph(store: GraphStore, output_path: str \| Path) -> dict[str, int]:
23	"""
24	Export the full graph to a JSONL file.
25
26	Returns:
27	Dict with counts: nodes, edges.
28	"""
29	output_path = Path(output_path)
30	output_path.parent.mkdir(parents=True, exist_ok=True)
31
32	nodes = _export_nodes(store)
33	edges = _export_edges(store)
34
35	# Sort for deterministic output
36	nodes.sort(key=lambda n: (n["label"], json.dumps(n["props"], sort_kkey=lambda e: (e["type"], e["type"],
37	th)
38	json.dum))
39
40	with output_path.open("w", encoding="utf-8") as f:
41	for node in nodes:
42	f.write(json.dumps(node, sort_keys=True) + "\n")
43	for edge in edges:
44	f.write(json.dumps(edge, sort_keys=True) + "\n")
45
46	logger.info("Exported %d nodes, %d edges to %s", len(nodes), len(edges), output_path)
47	return {"nodes": len(nodes), "edges": len(edges)}
48
49
50	def import_graph(store: GraphStore, input_path: str \| Path, clear: bool = True) -> dict[str, int]:
51	"""
52	Import a graph from a JSONL file.
53
54	Args:
55	store: Target GraphStore.
56	input_path: Path to the JSONL file.
57	clear: If True (default), wipe the graph before importing.
58
59	Returns:
60	Dict with counts: nodes, edges.
61	"""
62	input_path = Path(input_path)
63	if not input_path.exists():
64	raise FileNotFoundError(f"Export file not found: {input_path}")
65
66	if clear:
67	store.clear()
68
69	node_count = 0
70	edge_count = 0
71
72	with input_path.open("r", encoding="utf-8") as f:
73	for line in f:
74	line = line.strip()
75	if not line:
76	continue
77	record = json.loads(line)
78
79	if record["kind"] == "node":
80	_import_node(store, record)
81	node_count += 1
82	elif record["kind"] == "edge":
83	_import_edge(store, record)
84	edge_count += 1
85
86	logger.info("Imported %d nodes, %d edges from %s", node_count, edge_count, input_path)
87	return {"nodes": node_count, "edges": edge_count}
88
89
90	def _export_nodes(store: GraphStore) -> list[dict]:
91	"""Export all nodes with their labels and properties."""
92	result = store.query(
93	_keys=True),
94	json.dumps(e["to"], sort_keys=True),
95
96	)
97	)
98
99	with output_path.open("w", encoding="utf-8") as f:
100	for node in nodes:
101	f.write(json.dumps(node, sort_keys=True) + "\n")
102	for edge in edges:
103	f.write(json.dumps(edge, sort_keys=True) + "\n")
104
105	logger.info("Exported %d nodes, %d edges to %s", len(nodes), len(edges), output_path)
106	return {"nodes": len(nodes), "edges": len(edges)}
107
108
109	def import_graph(store: GraphStore, input_path: str \| Path, clear: bool = True) -> dict[str, int]:
110	"""
111	Import a graph from a JSONL file.
112
113	Args:
114	store: Target GraphStore.
115	input_path: Path to the JSONL file.
116	clear: If True (default), wipe the graph before importing.
117
118	Returns:
119	Dict with counts: no{
120	th)
121	d import for navegador.
122
123	Exports"""
124	Text-based graph export aise FileNotFoundError(f"Export file not found: {input_path}")
125
126	if r:
127	store.clear()
128
129	node_count = 0
130	edge_count = 0
131
132	}tf-8") as f:
133	for line in f:
134	line = line.strip()
135	if not line:
136	continue
137	record = json.loads(line)
138
139	if record["kind"] == "node":
140	_import_node(store, record)
141	node_count += 1
142	elif record["kind"] == "edge":
143	_import_edge(store, record)
144	edge_count += 1
145
146	logger.info("Imported %d nodes, %d edges from %s", node_count, edge_count, input_path)
147	return {"nodes": node_count, "edges": edge_count}
148
149
150	def _export_nodes(store: GraphStore) -> list[dict]:
151	"""Export all nodes with their labels and properties."""
152	result = store.query("MATCH (n) RETURN labels(n)[0] AS label, properties(n) AS props")
153	nodes = []
154	for row in result.result_set or []:
155	label = row[0]
156	props = row[1] if isinstance(row[1], dict) else {}
157	nodes.append({"kind": "node", "label": lafcord["to"]
158
159	from_key = "namef"name: $to_name"
160
161	params = {
162	"from_name": from_info["name"],
163	"to_name": to_info["name"],
164	}
165
166	if from_info.get("path"):
167	from_key += ", file_path: $from_path"
168	params["from_path"] = from_info["path"]
169
170	if to_info.get("path"):
171	to_key += ", path: $to_path"
172	params["to_path"] = to_info["path"]
173
174	cypher = (
175	f"MATCH (a:{from_info['label']} {{{from_key}}}), "
176	f"(b:{to_info['label']} {{{to_key}}}) "
177	f"MERGE (a)-[r:{edge_type}]->(b)"
178	)
179	s

M tests/test_cli.py

+56

		--- tests/test_cli.py
		+++ tests/test_cli.py
		@@ -638,10 +638,66 @@
638	638	assert result.exit_code == 0
639	639	MockPI.return_value.ingest_batch.assert_called_once()
640	640
641	641
642	642	# ── mcp command (lines 538-549) ───────────────────────────────────────────────
	643	+
	644	+# ── export / import ──────────────────────────────────────────────────────────
	645	+
	646	+class TestExportCommand:
	647	+ def test_export_success(self):
	648	+ runner = CliRunner()
	649	+ with runner.isolated_filesystem():
	650	+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
	651	+ patch("navegador.graph.export.export_graph", return_value={"nodes": 10, "edges": 5}):
	652	+ result = runner.invoke(main, ["export", "graph.jsonl"])
	653	+ assert result.exit_code == 0
	654	+ assert "10 nodes" in result.output
	655	+
	656	+ def test_export_json(self):
	657	+ runner = CliRunner()
	658	+ with runner.isolated_filesystem():
	659	+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
	660	+ patch("navegador.graph.export.export_graph", return_value={"nodes": 10, "edges": 5}):
	661	+ result = runner.invoke(main, ["export", "graph.jsonl", "--json"])
	662	+ assert result.exit_code == 0
	663	+ data = json.loads(result.output)
	664	+ assert data["nodes"] == 10
	665	+
	666	+
	667	+class TestImportCommand:
	668	+ def test_import_success(self):
	669	+ runner = CliRunner()
	670	+ with runner.isolated_filesystem():
	671	+ Path("graph.jsonl").write_text("")
	672	+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
	673	+ patch("navegador.graph.export.import_graph", return_value={"nodes": 10, "edges": 5}):
	674	+ result = runner.invoke(main, ["import", "graph.jsonl"])
	675	+ assert result.exit_code == 0
	676	+ assert "10 nodes" in result.output
	677	+
	678	+ def test_import_json(self):
	679	+ runner = CliRunner()
	680	+ with runner.isolated_filesystem():
	681	+ Path("graph.jsonl").write_text("")
	682	+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
	683	+ patch("navegador.graph.export.import_graph", return_value={"nodes": 8, "edges": 3}):
	684	+ result = runner.invoke(main, ["import", "graph.jsonl", "--json"])
	685	+ assert result.exit_code == 0
	686	+ data = json.loads(result.output)
	687	+ assert data["nodes"] == 8
	688	+
	689	+ def test_import_no_clear(self):
	690	+ runner = CliRunner()
	691	+ with runner.isolated_filesystem():
	692	+ Path("graph.jsonl").write_text("")
	693	+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
	694	+ patch("navegador.graph.export.import_graph", return_value={"nodes": 0, "edges": 0}) as mock_imp:
	695	+ runner.invoke(main, ["import", "graph.jsonl", "--no-clear"])
	696	+ mock_imp.assert_called_once()
	697	+ assert mock_imp.call_args[1]["clear"] is False
	698	+
643	699
644	700	# ── migrate ──────────────────────────────────────────────────────────────────
645	701
646	702	class TestMigrateCommand:
647	703	def test_migrate_applies_migrations(self):
648	704
649	705	ADDED tests/test_export.py

	--- tests/test_cli.py
	+++ tests/test_cli.py
	@@ -638,10 +638,66 @@
638	assert result.exit_code == 0
639	MockPI.return_value.ingest_batch.assert_called_once()
640
641
642	# ── mcp command (lines 538-549) ───────────────────────────────────────────────
























































643
644	# ── migrate ──────────────────────────────────────────────────────────────────
645
646	class TestMigrateCommand:
647	def test_migrate_applies_migrations(self):
648
649	DDED tests/test_export.py

	--- tests/test_cli.py
	+++ tests/test_cli.py
	@@ -638,10 +638,66 @@
638	assert result.exit_code == 0
639	MockPI.return_value.ingest_batch.assert_called_once()
640
641
642	# ── mcp command (lines 538-549) ───────────────────────────────────────────────
643
644	# ── export / import ──────────────────────────────────────────────────────────
645
646	class TestExportCommand:
647	def test_export_success(self):
648	runner = CliRunner()
649	with runner.isolated_filesystem():
650	with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
651	patch("navegador.graph.export.export_graph", return_value={"nodes": 10, "edges": 5}):
652	result = runner.invoke(main, ["export", "graph.jsonl"])
653	assert result.exit_code == 0
654	assert "10 nodes" in result.output
655
656	def test_export_json(self):
657	runner = CliRunner()
658	with runner.isolated_filesystem():
659	with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
660	patch("navegador.graph.export.export_graph", return_value={"nodes": 10, "edges": 5}):
661	result = runner.invoke(main, ["export", "graph.jsonl", "--json"])
662	assert result.exit_code == 0
663	data = json.loads(result.output)
664	assert data["nodes"] == 10
665
666
667	class TestImportCommand:
668	def test_import_success(self):
669	runner = CliRunner()
670	with runner.isolated_filesystem():
671	Path("graph.jsonl").write_text("")
672	with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
673	patch("navegador.graph.export.import_graph", return_value={"nodes": 10, "edges": 5}):
674	result = runner.invoke(main, ["import", "graph.jsonl"])
675	assert result.exit_code == 0
676	assert "10 nodes" in result.output
677
678	def test_import_json(self):
679	runner = CliRunner()
680	with runner.isolated_filesystem():
681	Path("graph.jsonl").write_text("")
682	with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
683	patch("navegador.graph.export.import_graph", return_value={"nodes": 8, "edges": 3}):
684	result = runner.invoke(main, ["import", "graph.jsonl", "--json"])
685	assert result.exit_code == 0
686	data = json.loads(result.output)
687	assert data["nodes"] == 8
688
689	def test_import_no_clear(self):
690	runner = CliRunner()
691	with runner.isolated_filesystem():
692	Path("graph.jsonl").write_text("")
693	with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
694	patch("navegador.graph.export.import_graph", return_value={"nodes": 0, "edges": 0}) as mock_imp:
695	runner.invoke(main, ["import", "graph.jsonl", "--no-clear"])
696	mock_imp.assert_called_once()
697	assert mock_imp.call_args[1]["clear"] is False
698
699
700	# ── migrate ──────────────────────────────────────────────────────────────────
701
702	class TestMigrateCommand:
703	def test_migrate_applies_migrations(self):
704
705	DDED tests/test_export.py

A tests/test_export.py

+252

		--- a/tests/test_export.py
		+++ b/tests/test_export.py
		@@ -0,0 +1,252 @@
	1	+"""Tests for navegador.graph.export — text-based graph export and import."""
	2	+
	3	+import json
	4	+import tempfile
	5	+from pathlib import Path
	6	+from unittest.mock import MagicMock
	7	+
	8	+import pytest
	9	+
	10	+from navegador.graph.export import (
	11	+ _export_edges,
	12	+ _export_nodes,
	13	+ _import_edge,
	14	+ _import_node,
	15	+ export_graph,
	16	+ import_graph,
	17	+)
	18	+
	19	+
	20	+def _mock_store(nodes=None, edges=None):
	21	+ store = MagicMock()
	22	+
	23	+ def query_side_effect(cypher, params=None):
	24	+ result = MagicMock()
	25	+ if "labels(n)" in cypher and "properties" in cypher:
	26	+ result.result_set = nodes or []
	27	+ elif "type(r)" in cypher:
	28	+ result.result_set = edges or []
	29	+ elif "DETACH DELETE" in cypher:
	30	+ result.result_set = []
	31	+ else:
	32	+ result.result_set = []
	33	+ return result
	34	+
	35	+ store.query.side_effect = query_side_effect
	36	+ store.clear = MagicMock()
	37	+ return store
	38	+
	39	+
	40	+# ── export_graph ─────────────────────────────────────────────────────────────
	41	+
	42	+class TestExportGraph:
	43	+ def test_creates_output_file(self):
	44	+ store = _mock_store(nodes=[], edges=[])
	45	+ with tempfile.TemporaryDirectory() as tmpdir:
	46	+ output = Path(tmpdir) / "graph.jsonl"
	47	+ export_graph(store, output)
	48	+ assert output.exists()
	49	+
	50	+ def test_returns_counts(self):
	51	+ nodes = [["Function", {"name": "foo", "file_path": "app.py"}]]
	52	+ store = _mock_store(nodes=nodes, edges=[])
	53	+ with tempfile.TemporaryDirectory() as tmpdir:
	54	+ stats = export_graph(store, Path(tmpdir) / "graph.jsonl")
	55	+ assert stats["nodes"] == 1
	56	+ assert stats["edges"] == 0
	57	+
	58	+ def test_writes_valid_jsonl(self):
	59	+ nodes = [
	60	+ ["Function", {"name": "foo", "file_path": "app.py"}],
	61	+ ["Class", {"name": "Bar", "file_path": "bar.py"}],
	62	+ ]
	63	+ edges = [["CALLS", "Function", "foo", "app.py", "Function", "bar", "bar.py"]]
	64	+ store = _mock_store(nodes=nodes, edges=edges)
	65	+ with tempfile.TemporaryDirectory() as tmpdir:
	66	+ output = Path(tmpdir) / "graph.jsonl"
	67	+ export_graph(store, output)
	68	+ lines = output.read_text().strip().split("\n")
	69	+ assert len(lines) == 3 # 2 nodes + 1 edge
	70	+ for line in lines:
	71	+ record = json.loads(line)
	72	+ assert record["kind"] in ("node", "edge")
	73	+
	74	+ def test_output_is_sorted(self):
	75	+ nodes = [
	76	+ ["Function", {"name": "z_func", "file_path": "z.py"}],
	77	+ ["Class", {"name": "a_class", "file_path": "a.py"}],
	78	+ ]
	79	+ store = _mock_store(nodes=nodes, edges=[])
	80	+ with tempfile.TemporaryDirectory() as tmpdir:
	81	+ output = Path(tmpdir) / "graph.jsonl"
	82	+ export_graph(store, output)
	83	+ lines = output.read_text().strip().split("\n")
	84	+ labels = [json.loads(line)["label"] for line in lines]
	85	+ # Class comes before Function alphabetically
	86	+ assert labels[0] == "Class"
	87	+ assert labels[1] == "Function"
	88	+
	89	+ def test_creates_parent_dirs(self):
	90	+ store = _mock_store(nodes=[], edges=[])
	91	+ with tempfile.TemporaryDirectory() as tmpdir:
	92	+ output = Path(tmpdir) / "sub" / "dir" / "graph.jsonl"
	93	+ export_graph(store, output)
	94	+ assert output.exists()
	95	+
	96	+
	97	+# ── import_graph ─────────────────────────────────────────────────────────────
	98	+
	99	+class TestImportGraph:
	100	+ def test_raises_on_missing_file(self):
	101	+ store = MagicMock()
	102	+ with pytest.raises(FileNotFoundError):
	103	+ import_graph(store, "/nonexistent/graph.jsonl")
	104	+
	105	+ def test_clears_graph_by_default(self):
	106	+ store = MagicMock()
	107	+ store.query.return_value = MagicMock(result_set=[])
	108	+ with tempfile.TemporaryDirectory() as tmpdir:
	109	+ f = Path(tmpdir) / "graph.jsonl"
	110	+ f.write_text("")
	111	+ import_graph(store, f)
	112	+ store.clear.assert_called_once()
	113	+
	114	+ def test_no_clear_flag(self):
	115	+ store = MagicMock()
	116	+ store.query.return_value = MagicMock(result_set=[])
	117	+ with tempfile.TemporaryDirectory() as tmpdir:
	118	+ f = Path(tmpdir) / "graph.jsonl"
	119	+ f.write_text("")
	120	+ import_graph(store, f, clear=False)
	121	+ store.clear.assert_not_called()
	122	+
	123	+ def test_imports_nodes(self):
	124	+ store = MagicMock()
	125	+ store.query.return_value = MagicMock(result_set=[])
	126	+ with tempfile.TemporaryDirectory() as tmpdir:
	127	+ f = Path(tmpdir) / "graph.jsonl"
	128	+ node = {"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": "app.py"}}
	129	+ f.write_text(json.dumps(node) + "\n")
	130	+ stats = import_graph(store, f)
	131	+ assert stats["nodes"] == 1
	132	+
	133	+ def test_imports_edges(self):
	134	+ store = MagicMock()
	135	+ store.query.return_value = MagicMock(result_set=[])
	136	+ with tempfile.TemporaryDirectory() as tmpdir:
	137	+ f = Path(tmpdir) / "graph.jsonl"
	138	+ edge = {
	139	+ "kind": "edge",
	140	+ "type": "CALLS",
	141	+ "from": {"label": "Function", "name": "foo", "path": "app.py"},
	142	+ "to": {"label": "Function", "name": "bar", "path": "bar.py"},
	143	+ }
	144	+ f.write_text(json.dumps(edge) + "\n")
	145	+ stats = import_graph(store, f)
	146	+ assert stats["edges"] == 1
	147	+
	148	+ def test_returns_counts(self):
	149	+ store = MagicMock()
	150	+ store.query.return_value = MagicMock(result_set=[])
	151	+ with tempfile.TemporaryDirectory() as tmpdir:
	152	+ f = Path(tmpdir) / "graph.jsonl"
	153	+ lines = [
	154	+ json.dumps({"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": "app.py"}}),
	155	+ json.dumps({"kind": "node", "label": "Class", "props": {"name": "Bar", "file_path": "bar.py"}}),
	156	+ json.dumps({"kind": "edge", "type": "CALLS",
	157	+ "from": {"label": "Function", "name": "foo", "path": ""},
	158	+ "to": {"label": "Class", "name": "Bar", "path": ""}}),
	159	+ ]
	160	+ f.write_text("\n".join(lines) + "\n")
	161	+ stats = import_graph(store, f)
	162	+ assert stats["nodes"] == 2
	163	+ assert stats["edges"] == 1
	164	+
	165	+ def test_skips_blank_lines(self):
	166	+ store = MagicMock()
	167	+ store.query.return_value = MagicMock(result_set=[])
	168	+ with tempfile.TemporaryDirectory() as tmpdir:
	169	+ f = Path(tmpdir) / "graph.jsonl"
	170	+ node = json.dumps({"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": ""}})
	171	+ f.write_text(f"\n{node}\n\n")
	172	+ stats = import_graph(store, f)
	173	+ assert stats["nodes"] == 1
	174	+
	175	+
	176	+# ── _export_nodes / _export_edges ────────────────────────────────────────────
	177	+
	178	+class TestExportHelpers:
	179	+ def test_export_nodes_handles_non_dict_props(self):
	180	+ store = MagicMock()
	181	+ store.query.return_value = MagicMock(result_set=[["Function", "not_a_dict"]])
	182	+ nodes = _export_nodes(store)
	183	+ assert len(nodes) == 1
	184	+ assert nodes[0]["props"] == {}
	185	+
	186	+ def test_export_edges_returns_structured_data(self):
	187	+ store = MagicMock()
	188	+ store.query.return_value = MagicMock(
	189	+ result_set=[["CALLS", "Function", "foo", "app.py", "Function", "bar", "bar.py"]]
	190	+ )
	191	+ edges = _export_edges(store)
	192	+ assert len(edges) == 1
	193	+ assert edges[0]["type"] == "CALLS"
	194	+ assert edges[0]["from"]["name"] == "foo"
	195	+ assert edges[0]["to"]["name"] == "bar"
	196	+
	197	+
	198	+# ── _import_node / _import_edge ──────────────────────────────────────────────
	199	+
	200	+class TestImportHelpers:
	201	+ def test_import_node_adds_missing_file_path(self):
	202	+ store = MagicMock()
	203	+ store.query.return_value = MagicMock(result_set=[])
	204	+ record = {"kind": "node", "label": "Concept", "props": {"name": "JWT"}}
	205	+ _import_node(store, record)
	206	+ store.query.assert_called_once()
	207	+ cypher = store.query.call_args[0][0]
	208	+ assert "MERGE" in cypher
	209	+
	210	+ def test_import_node_adds_missing_name(self):
	211	+ store = MagicMock()
	212	+ store.query.return_value = MagicMock(result_set=[])
	213	+ record = {"kind": "node", "label": "Domain", "props": {"description": "Auth domain"}}
	214	+ _import_node(store, record)
	215	+ # Should have added name="" to props
	216	+ store.query.assert_called_once()
	217	+ params = store.query.call_args[0][1]
	218	+ assert params["name"] == ""
	219	+
	220	+ def test_import_node_uses_path_key_for_repos(self):
	221	+ store = MagicMock()
	222	+ store.query.return_value = MagicMock(result_set=[])
	223	+ record = {"kind": "node", "label": "Repository", "props": {"name": "myrepo", "path": "/code/myrepo"}}
	224	+ _import_node(store, record)
	225	+ cypher = store.query.call_args[0][0]
	226	+ assert "path" in cypher
	227	+
	228	+ def test_import_edge_with_paths(self):
	229	+ store = MagicMock()
	230	+ store.query.return_value = MagicMock(result_set=[])
	231	+ record = {
	232	+ "kind": "edge",
	233	+ "type": "CALLS",
	234	+ "from": {"label": "Function", "name": "foo", "path": "app.py"},
	235	+ "to": {"label": "Function", "name": "bar", "path": "bar.py"},
	236	+ }
	237	+ _import_edge(store, record)
	238	+ store.query.assert_called_once()
	239	+
	240	+ def test_import_edge_without_paths(self):
	241	+ store = MagicMock()
	242	+ store.query.return_value = MagicMock(result_set=[])
	243	+ record = {
	244	+ "kind": "edge",
	245	+ "type": "RELATED_TO",
	246	+ "from": {"label": "Concept", "name": "JWT", "path": ""},
	247	+ "to": {"label": "Concept", "name": "OAuth", "path": ""},
	248	+ }
	249	+ _import_edge(store, record)
	250	+ store.query.assert_called_once()
	251	+ cypher = store.query.call_args[0][0]
	252	+ assert "file_path" not in cypher

	--- a/tests/test_export.py
	+++ b/tests/test_export.py
	@@ -0,0 +1,252 @@

	--- a/tests/test_export.py
	+++ b/tests/test_export.py
	@@ -0,0 +1,252 @@
1	"""Tests for navegador.graph.export — text-based graph export and import."""
2
3	import json
4	import tempfile
5	from pathlib import Path
6	from unittest.mock import MagicMock
7
8	import pytest
9
10	from navegador.graph.export import (
11	_export_edges,
12	_export_nodes,
13	_import_edge,
14	_import_node,
15	export_graph,
16	import_graph,
17	)
18
19
20	def _mock_store(nodes=None, edges=None):
21	store = MagicMock()
22
23	def query_side_effect(cypher, params=None):
24	result = MagicMock()
25	if "labels(n)" in cypher and "properties" in cypher:
26	result.result_set = nodes or []
27	elif "type(r)" in cypher:
28	result.result_set = edges or []
29	elif "DETACH DELETE" in cypher:
30	result.result_set = []
31	else:
32	result.result_set = []
33	return result
34
35	store.query.side_effect = query_side_effect
36	store.clear = MagicMock()
37	return store
38
39
40	# ── export_graph ─────────────────────────────────────────────────────────────
41
42	class TestExportGraph:
43	def test_creates_output_file(self):
44	store = _mock_store(nodes=[], edges=[])
45	with tempfile.TemporaryDirectory() as tmpdir:
46	output = Path(tmpdir) / "graph.jsonl"
47	export_graph(store, output)
48	assert output.exists()
49
50	def test_returns_counts(self):
51	nodes = [["Function", {"name": "foo", "file_path": "app.py"}]]
52	store = _mock_store(nodes=nodes, edges=[])
53	with tempfile.TemporaryDirectory() as tmpdir:
54	stats = export_graph(store, Path(tmpdir) / "graph.jsonl")
55	assert stats["nodes"] == 1
56	assert stats["edges"] == 0
57
58	def test_writes_valid_jsonl(self):
59	nodes = [
60	["Function", {"name": "foo", "file_path": "app.py"}],
61	["Class", {"name": "Bar", "file_path": "bar.py"}],
62	]
63	edges = [["CALLS", "Function", "foo", "app.py", "Function", "bar", "bar.py"]]
64	store = _mock_store(nodes=nodes, edges=edges)
65	with tempfile.TemporaryDirectory() as tmpdir:
66	output = Path(tmpdir) / "graph.jsonl"
67	export_graph(store, output)
68	lines = output.read_text().strip().split("\n")
69	assert len(lines) == 3 # 2 nodes + 1 edge
70	for line in lines:
71	record = json.loads(line)
72	assert record["kind"] in ("node", "edge")
73
74	def test_output_is_sorted(self):
75	nodes = [
76	["Function", {"name": "z_func", "file_path": "z.py"}],
77	["Class", {"name": "a_class", "file_path": "a.py"}],
78	]
79	store = _mock_store(nodes=nodes, edges=[])
80	with tempfile.TemporaryDirectory() as tmpdir:
81	output = Path(tmpdir) / "graph.jsonl"
82	export_graph(store, output)
83	lines = output.read_text().strip().split("\n")
84	labels = [json.loads(line)["label"] for line in lines]
85	# Class comes before Function alphabetically
86	assert labels[0] == "Class"
87	assert labels[1] == "Function"
88
89	def test_creates_parent_dirs(self):
90	store = _mock_store(nodes=[], edges=[])
91	with tempfile.TemporaryDirectory() as tmpdir:
92	output = Path(tmpdir) / "sub" / "dir" / "graph.jsonl"
93	export_graph(store, output)
94	assert output.exists()
95
96
97	# ── import_graph ─────────────────────────────────────────────────────────────
98
99	class TestImportGraph:
100	def test_raises_on_missing_file(self):
101	store = MagicMock()
102	with pytest.raises(FileNotFoundError):
103	import_graph(store, "/nonexistent/graph.jsonl")
104
105	def test_clears_graph_by_default(self):
106	store = MagicMock()
107	store.query.return_value = MagicMock(result_set=[])
108	with tempfile.TemporaryDirectory() as tmpdir:
109	f = Path(tmpdir) / "graph.jsonl"
110	f.write_text("")
111	import_graph(store, f)
112	store.clear.assert_called_once()
113
114	def test_no_clear_flag(self):
115	store = MagicMock()
116	store.query.return_value = MagicMock(result_set=[])
117	with tempfile.TemporaryDirectory() as tmpdir:
118	f = Path(tmpdir) / "graph.jsonl"
119	f.write_text("")
120	import_graph(store, f, clear=False)
121	store.clear.assert_not_called()
122
123	def test_imports_nodes(self):
124	store = MagicMock()
125	store.query.return_value = MagicMock(result_set=[])
126	with tempfile.TemporaryDirectory() as tmpdir:
127	f = Path(tmpdir) / "graph.jsonl"
128	node = {"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": "app.py"}}
129	f.write_text(json.dumps(node) + "\n")
130	stats = import_graph(store, f)
131	assert stats["nodes"] == 1
132
133	def test_imports_edges(self):
134	store = MagicMock()
135	store.query.return_value = MagicMock(result_set=[])
136	with tempfile.TemporaryDirectory() as tmpdir:
137	f = Path(tmpdir) / "graph.jsonl"
138	edge = {
139	"kind": "edge",
140	"type": "CALLS",
141	"from": {"label": "Function", "name": "foo", "path": "app.py"},
142	"to": {"label": "Function", "name": "bar", "path": "bar.py"},
143	}
144	f.write_text(json.dumps(edge) + "\n")
145	stats = import_graph(store, f)
146	assert stats["edges"] == 1
147
148	def test_returns_counts(self):
149	store = MagicMock()
150	store.query.return_value = MagicMock(result_set=[])
151	with tempfile.TemporaryDirectory() as tmpdir:
152	f = Path(tmpdir) / "graph.jsonl"
153	lines = [
154	json.dumps({"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": "app.py"}}),
155	json.dumps({"kind": "node", "label": "Class", "props": {"name": "Bar", "file_path": "bar.py"}}),
156	json.dumps({"kind": "edge", "type": "CALLS",
157	"from": {"label": "Function", "name": "foo", "path": ""},
158	"to": {"label": "Class", "name": "Bar", "path": ""}}),
159	]
160	f.write_text("\n".join(lines) + "\n")
161	stats = import_graph(store, f)
162	assert stats["nodes"] == 2
163	assert stats["edges"] == 1
164
165	def test_skips_blank_lines(self):
166	store = MagicMock()
167	store.query.return_value = MagicMock(result_set=[])
168	with tempfile.TemporaryDirectory() as tmpdir:
169	f = Path(tmpdir) / "graph.jsonl"
170	node = json.dumps({"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": ""}})
171	f.write_text(f"\n{node}\n\n")
172	stats = import_graph(store, f)
173	assert stats["nodes"] == 1
174
175
176	# ── _export_nodes / _export_edges ────────────────────────────────────────────
177
178	class TestExportHelpers:
179	def test_export_nodes_handles_non_dict_props(self):
180	store = MagicMock()
181	store.query.return_value = MagicMock(result_set=[["Function", "not_a_dict"]])
182	nodes = _export_nodes(store)
183	assert len(nodes) == 1
184	assert nodes[0]["props"] == {}
185
186	def test_export_edges_returns_structured_data(self):
187	store = MagicMock()
188	store.query.return_value = MagicMock(
189	result_set=[["CALLS", "Function", "foo", "app.py", "Function", "bar", "bar.py"]]
190	)
191	edges = _export_edges(store)
192	assert len(edges) == 1
193	assert edges[0]["type"] == "CALLS"
194	assert edges[0]["from"]["name"] == "foo"
195	assert edges[0]["to"]["name"] == "bar"
196
197
198	# ── _import_node / _import_edge ──────────────────────────────────────────────
199
200	class TestImportHelpers:
201	def test_import_node_adds_missing_file_path(self):
202	store = MagicMock()
203	store.query.return_value = MagicMock(result_set=[])
204	record = {"kind": "node", "label": "Concept", "props": {"name": "JWT"}}
205	_import_node(store, record)
206	store.query.assert_called_once()
207	cypher = store.query.call_args[0][0]
208	assert "MERGE" in cypher
209
210	def test_import_node_adds_missing_name(self):
211	store = MagicMock()
212	store.query.return_value = MagicMock(result_set=[])
213	record = {"kind": "node", "label": "Domain", "props": {"description": "Auth domain"}}
214	_import_node(store, record)
215	# Should have added name="" to props
216	store.query.assert_called_once()
217	params = store.query.call_args[0][1]
218	assert params["name"] == ""
219
220	def test_import_node_uses_path_key_for_repos(self):
221	store = MagicMock()
222	store.query.return_value = MagicMock(result_set=[])
223	record = {"kind": "node", "label": "Repository", "props": {"name": "myrepo", "path": "/code/myrepo"}}
224	_import_node(store, record)
225	cypher = store.query.call_args[0][0]
226	assert "path" in cypher
227
228	def test_import_edge_with_paths(self):
229	store = MagicMock()
230	store.query.return_value = MagicMock(result_set=[])
231	record = {
232	"kind": "edge",
233	"type": "CALLS",
234	"from": {"label": "Function", "name": "foo", "path": "app.py"},
235	"to": {"label": "Function", "name": "bar", "path": "bar.py"},
236	}
237	_import_edge(store, record)
238	store.query.assert_called_once()
239
240	def test_import_edge_without_paths(self):
241	store = MagicMock()
242	store.query.return_value = MagicMock(result_set=[])
243	record = {
244	"kind": "edge",
245	"type": "RELATED_TO",
246	"from": {"label": "Concept", "name": "JWT", "path": ""},
247	"to": {"label": "Concept", "name": "OAuth", "path": ""},
248	}
249	_import_edge(store, record)
250	store.query.assert_called_once()
251	cypher = store.query.call_args[0][0]
252	assert "file_path" not in cypher

Navegador

Keyboard Shortcuts