skills/ar9av__obsidian-wiki
1
0
Fork 0
mirror of https://github.com/Ar9av/obsidian-wiki.git synced 2026-09-14 20:36:34 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
feat/queryable-graph-layer
ar9av__obsidian-wiki/tests/test_graph_analysis.py
T
ar9av 382c3da29c feat(graph): bridges, cohesion, paths, neighbourhoods and a betweenness cache 
Brings the graph analyser up to the algorithm family graphify runs on code
graphs, in pure Python (no networkx/numpy — the package stays dependency-free):

- `bridges` — exact Brandes betweenness centrality. The pages that sit on the
  most shortest paths; removing one fragments the vault.
- `cohesion` per community — intra-cluster edge density; < 0.15 with 5+ pages
  marks a fragmented cluster.
- `--path A B` — shortest wikilink chain between two pages.
- `--around PAGE --depth N --direction in|out|both` — N-hop neighbourhood;
  `--direction in` is the blast radius if the page were renamed or removed.
- `--diff-against` / `--snapshot` — compare against a previous `_insights.md`
  snapshot (added/removed pages and links, newly connected, lost incoming).
- `suggested_questions` — derived from bridges, sink hubs, isolates and
  low-cohesion clusters.

Two correctness fixes fall out of this:

- Vault bookkeeping files (`index`, `log`, `hot`, `_insights` at the vault
  root, plus `_meta/` and `_readouts/`) are no longer graph nodes. They link
  to nearly every page, so they dominated every degree and centrality ranking.
  A nested `concepts/index.md` is still an ordinary page.
- Pages are now iterated in sorted order rather than filesystem order, so
  results are reproducible across machines. That exposed a latent flaw in the
  label-propagation fallback: a sequential sweep cascades one label along
  chains and collapsed a barbell graph into a single community. Nodes are now
  visited in a seeded shuffle — deterministic, and clusters survive.

Betweenness is O(V*E) and the only expensive step in the module (~43s on a
5,000-page vault, versus 0.13s to parse it and 0.03s for communities), so it
is memoised in `.graph-cache.json`. The cache key is a hash of the graph
topology rather than file mtimes, which makes a stale hit impossible by
construction: retargeting any link changes the key, while editing prose
without touching links keeps the cached result valid. The file is only
written when the computation actually exceeded 0.5s, so small and medium
vaults never accumulate one; it is bounded to 3 keys, written atomically,
and ignored if corrupt.

Sampled betweenness (k pivots) was measured as an alternative and rejected:
25-50x faster but only 60-90% top-10 overlap and wrong ordering even on a
38-page vault, which is not acceptable for a ranked list a human reads.
2026-08-16 23:10:06 -07:00

696 lines
27 KiB
Python
Raw Permalink Blame History

"""Tests for vault graph analysis: community detection, god nodes, surprising connections."""
import json
import subprocess
import sys
import textwrap
from pathlib import Path
import pytest
from obsidian_wiki.graph_analysis import (
analyse_vault,
dead_ends,
detect_communities_greedy,
god_nodes,
isolated,
parse_vault_graph,
surprising_connections,
)
# ---------------------------------------------------------------------------
# Fixtures — synthetic vault
# ---------------------------------------------------------------------------
@pytest.fixture
def vault(tmp_path):
v = tmp_path / "vault"
v.mkdir()
return v
def _page(vault: Path, name: str, links: list[str], tags: list[str] | None = None) -> Path:
"""Write a minimal wiki page with wikilinks."""
lines = ["---", f"title: {name}"]
if tags:
lines.append(f"tags: [{', '.join(tags)}]")
lines += ["---", f"# {name}", ""]
lines += [f"[[{lnk}]]" for lnk in links]
p = vault / f"{name}.md"
p.write_text("\n".join(lines) + "\n")
return p
@pytest.fixture
def simple_vault(vault):
"""
A → B → C
A → C
D (isolated)
E → F (dead-end cluster)
"""
_page(vault, "a", ["b", "c"], tags=["concepts"])
_page(vault, "b", ["c"], tags=["concepts"])
_page(vault, "c", [], tags=["references"])
_page(vault, "d", [], tags=["entities"])
_page(vault, "e", ["f"])
_page(vault, "f", [])
return vault
# ---------------------------------------------------------------------------
# parse_vault_graph
# ---------------------------------------------------------------------------
class TestParseVaultGraph:
def test_reads_wikilinks(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
assert "b" in outgoing["a"]
assert "c" in outgoing["a"]
def test_all_pages_present_as_keys(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
assert set(outgoing.keys()) == {"a", "b", "c", "d", "e", "f"}
def test_reads_tags(self, simple_vault):
_, tags = parse_vault_graph(simple_vault)
assert "concepts" in tags.get("a", [])
def test_empty_vault(self, vault):
outgoing, tags = parse_vault_graph(vault)
assert outgoing == {}
def test_self_links_ignored(self, vault):
_page(vault, "selfref", ["selfref"])
outgoing, _ = parse_vault_graph(vault)
assert "selfref" not in outgoing.get("selfref", [])
def test_links_to_nonexistent_pages_excluded(self, vault):
_page(vault, "orphan", ["doesnotexist"])
outgoing, _ = parse_vault_graph(vault)
assert outgoing["orphan"] == []
# ---------------------------------------------------------------------------
# god_nodes
# ---------------------------------------------------------------------------
class TestGodNodes:
def test_c_is_top_hub(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
result = god_nodes(outgoing, top_n=3)
# c has 2 in-links (from a and b), so should be in top 3
top_pages = {r["page"] for r in result}
assert "c" in top_pages
def test_degree_sum(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
result = god_nodes(outgoing)
for node in result:
assert node["degree"] == node["in_degree"] + node["out_degree"]
def test_respects_top_n(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
result = god_nodes(outgoing, top_n=2)
assert len(result) <= 2
# ---------------------------------------------------------------------------
# dead_ends / isolated
# ---------------------------------------------------------------------------
class TestDeadEndsIsolated:
def test_dead_ends(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
de = dead_ends(outgoing)
assert "c" in de
assert "f" in de
assert "a" not in de
def test_isolated(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
iso = isolated(outgoing)
assert "d" in iso
assert "a" not in iso
assert "c" not in iso # c has incoming links so not isolated
# ---------------------------------------------------------------------------
# Community detection
# ---------------------------------------------------------------------------
class TestCommunityDetection:
def test_returns_list_of_sets(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
comms = detect_communities_greedy(outgoing)
assert isinstance(comms, list)
for c in comms:
assert isinstance(c, set)
def test_all_nodes_assigned(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
comms = detect_communities_greedy(outgoing)
all_nodes = set(outgoing.keys())
assigned = set()
for c in comms:
assigned |= c
assert assigned == all_nodes
def test_no_overlap(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
comms = detect_communities_greedy(outgoing)
seen = set()
for c in comms:
assert c.isdisjoint(seen), "Node appears in two communities"
seen |= c
def test_connected_cluster_grouped(self, vault):
# a-b-c tightly connected, x-y-z separate — should land in different communities
_page(vault, "a", ["b", "c"])
_page(vault, "b", ["a", "c"])
_page(vault, "c", ["a", "b"])
_page(vault, "x", ["y", "z"])
_page(vault, "y", ["x", "z"])
_page(vault, "z", ["x", "y"])
outgoing, _ = parse_vault_graph(vault)
comms = detect_communities_greedy(outgoing)
# At least 2 communities
assert len(comms) >= 2
def test_empty_graph(self, vault):
outgoing, _ = parse_vault_graph(vault)
comms = detect_communities_greedy(outgoing)
assert comms == []
# ---------------------------------------------------------------------------
# Surprising connections
# ---------------------------------------------------------------------------
class TestSurprisingConnections:
def test_cross_community_edge_found(self, vault):
# Build two tight clusters with one bridge
_page(vault, "a", ["b", "c", "x"])
_page(vault, "b", ["a", "c"])
_page(vault, "c", ["a", "b"])
_page(vault, "x", ["y", "z"])
_page(vault, "y", ["x", "z"])
_page(vault, "z", ["x", "y"])
outgoing, _ = parse_vault_graph(vault)
comms = detect_communities_greedy(outgoing)
sc = surprising_connections(outgoing, comms)
sources = {s["source"] for s in sc}
targets = {s["target"] for s in sc}
assert sources | targets # at least one cross-community edge found
def test_no_intra_community_edges(self, vault):
_page(vault, "a", ["b"])
_page(vault, "b", ["a"])
outgoing, _ = parse_vault_graph(vault)
comms = [{"a", "b"}] # one community
sc = surprising_connections(outgoing, comms)
assert sc == []
def test_scores_positive(self, vault):
_page(vault, "a", ["b", "x"])
_page(vault, "b", ["a"])
_page(vault, "x", ["y"])
_page(vault, "y", ["x"])
outgoing, _ = parse_vault_graph(vault)
comms = detect_communities_greedy(outgoing)
sc = surprising_connections(outgoing, comms)
for item in sc:
assert item["score"] > 0
# ---------------------------------------------------------------------------
# analyse_vault (integration)
# ---------------------------------------------------------------------------
class TestAnalyseVault:
def test_returns_all_keys(self, simple_vault):
result = analyse_vault(simple_vault)
assert set(result.keys()) == {
"god_nodes", "bridges", "communities", "surprising_connections",
"suggested_questions", "dead_ends", "isolated", "stats",
}
def test_stats_correct(self, simple_vault):
result = analyse_vault(simple_vault)
assert result["stats"]["pages"] == 6
assert result["stats"]["edges"] > 0
def test_empty_vault(self, vault):
result = analyse_vault(vault)
assert result["stats"]["pages"] == 0
def test_json_serialisable(self, simple_vault):
result = analyse_vault(simple_vault)
json.dumps(result) # must not raise
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
class TestGraphAnalyseCLI:
def test_outputs_json(self, simple_vault):
proc = subprocess.run(
[sys.executable, "-m", "obsidian_wiki.cli", "graph-analyse", str(simple_vault)],
capture_output=True, text=True,
)
assert proc.returncode == 0
data = json.loads(proc.stdout)
assert "god_nodes" in data
def test_pretty_flag(self, simple_vault):
proc = subprocess.run(
[sys.executable, "-m", "obsidian_wiki.cli", "graph-analyse", str(simple_vault), "--pretty"],
capture_output=True, text=True,
)
assert proc.returncode == 0
assert "\n " in proc.stdout
def test_top_flag(self, simple_vault):
proc = subprocess.run(
[sys.executable, "-m", "obsidian_wiki.cli", "graph-analyse", str(simple_vault), "--top", "3"],
capture_output=True, text=True,
)
assert proc.returncode == 0
data = json.loads(proc.stdout)
assert len(data["god_nodes"]) <= 3
def test_missing_vault_exits_nonzero(self, tmp_path):
proc = subprocess.run(
[sys.executable, "-m", "obsidian_wiki.cli", "graph-analyse", str(tmp_path / "nope")],
capture_output=True, text=True,
)
assert proc.returncode != 0
# ---------------------------------------------------------------------------
# graphify-derived algorithms: betweenness, cohesion, path, neighbourhood,
# diff, suggested questions
# ---------------------------------------------------------------------------
from obsidian_wiki.graph_analysis import ( # noqa: E402
betweenness_centrality,
bridge_pages,
cohesion_score,
graph_diff,
load_snapshot,
neighborhood,
shortest_path,
snapshot,
suggest_questions,
)
@pytest.fixture
def barbell_vault(vault):
"""Two triangles joined through a single bridge page `m`.
a-b-c clique — m — x-y-z clique
"""
_page(vault, "a", ["b", "c", "m"], tags=["left"])
_page(vault, "b", ["a", "c"], tags=["left"])
_page(vault, "c", ["a", "b"], tags=["left"])
_page(vault, "m", ["a", "x"], tags=["mid"])
_page(vault, "x", ["m", "y", "z"], tags=["right"])
_page(vault, "y", ["x", "z"], tags=["right"])
_page(vault, "z", ["x", "y"], tags=["right"])
return vault
class TestBetweenness:
def test_bridge_has_highest_betweenness(self, barbell_vault):
outgoing, _ = parse_vault_graph(barbell_vault)
bc = betweenness_centrality(outgoing)
assert max(bc, key=bc.get) == "m"
# leaf-of-clique nodes lie on no shortest paths
assert bc["b"] == 0.0 and bc["y"] == 0.0
def test_normalised_range(self, barbell_vault):
outgoing, _ = parse_vault_graph(barbell_vault)
for v in betweenness_centrality(outgoing).values():
assert 0.0 <= v <= 1.0
def test_matches_star_exact_value(self, vault):
# Star: hub h with 4 leaves — hub betweenness = 1.0 normalised
_page(vault, "h", ["p", "q", "r", "s"])
for leaf in "pqrs":
_page(vault, leaf, [])
outgoing, _ = parse_vault_graph(vault)
bc = betweenness_centrality(outgoing)
assert bc["h"] == pytest.approx(1.0)
def test_bridge_pages_reports_connected_communities(self, barbell_vault):
outgoing, _ = parse_vault_graph(barbell_vault)
comms = detect_communities_greedy(outgoing)
bridges = bridge_pages(outgoing, comms, top_n=3)
assert bridges[0]["page"] == "m"
assert bridges[0]["betweenness"] > 0
assert "connects" in bridges[0]
def test_empty(self):
assert betweenness_centrality({}) == {}
assert bridge_pages({}, [], 5) == []
class TestCohesion:
def test_clique_is_one(self, barbell_vault):
outgoing, _ = parse_vault_graph(barbell_vault)
assert cohesion_score(outgoing, {"a", "b", "c"}) == 1.0
def test_no_internal_links_is_zero(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
assert cohesion_score(outgoing, {"d", "f", "c"}) == 0.0
def test_singleton_is_one(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
assert cohesion_score(outgoing, {"d"}) == 1.0
def test_analyse_includes_cohesion(self, barbell_vault):
result = analyse_vault(barbell_vault)
for c in result["communities"]:
assert 0.0 <= c["cohesion"] <= 1.0
class TestShortestPath:
def test_finds_path_across_bridge(self, barbell_vault):
outgoing, _ = parse_vault_graph(barbell_vault)
path = shortest_path(outgoing, "b", "y")
assert path[0] == "b" and path[-1] == "y"
assert "m" in path
assert len(path) == 5 # b-a-m-x-y
def test_directed_respects_link_direction(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
assert shortest_path(outgoing, "a", "c", directed=True) == ["a", "c"]
assert shortest_path(outgoing, "c", "a", directed=True) is None
assert shortest_path(outgoing, "c", "a") == ["c", "a"]
def test_unreachable_and_unknown(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
assert shortest_path(outgoing, "a", "d") is None
assert shortest_path(outgoing, "a", "nope") is None
assert shortest_path(outgoing, "a", "a") == ["a"]
def test_slug_normalisation(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
assert shortest_path(outgoing, "A", "B") == ["a", "b"]
class TestNeighborhood:
def test_depth_layers(self, barbell_vault):
outgoing, _ = parse_vault_graph(barbell_vault)
hits = neighborhood(outgoing, "m", depth=1)
assert {h["page"] for h in hits} == {"a", "x"}
hits2 = neighborhood(outgoing, "m", depth=2)
assert {h["page"] for h in hits2} == {"a", "x", "b", "c", "y", "z"}
assert all(h["depth"] in (1, 2) for h in hits2)
def test_incoming_blast_radius(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
# who would break if `c` were renamed? a and b link to it
hits = neighborhood(outgoing, "c", depth=1, direction="in")
assert {h["page"] for h in hits} == {"a", "b"}
assert neighborhood(outgoing, "c", depth=1, direction="out") == []
def test_unknown_seed(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
assert neighborhood(outgoing, "nope") == []
class TestGraphDiff:
def test_added_removed(self):
old = {"nodes": ["a", "b", "c"], "edges": [["a", "b"]]}
new = {"nodes": ["a", "b", "d"], "edges": [["a", "b"], ["a", "d"], ["d", "b"]]}
d = graph_diff(old, new)
assert d["added_pages"] == ["d"]
assert d["removed_pages"] == ["c"]
assert ["a", "d"] in d["added_edges"]
assert d["removed_edges"] == []
assert d["summary"] == {"pages": 0, "edges": 2}
def test_newly_connected_and_lost_incoming(self):
old = {"nodes": ["a", "b", "c"], "edges": [["a", "b"]]}
new = {"nodes": ["a", "b", "c"], "edges": [["a", "c"]]}
d = graph_diff(old, new)
assert d["newly_connected"] == ["c"]
assert d["lost_incoming"] == ["b"]
def test_snapshot_roundtrip_via_insights_comment(self, simple_vault, tmp_path):
outgoing, _ = parse_vault_graph(simple_vault)
snap = snapshot(outgoing)
insights = tmp_path / "_insights.md"
insights.write_text("# Insights\n\nstuff\n\n<!-- GRAPH_SNAPSHOT: " + json.dumps(snap) + " -->\n")
loaded = load_snapshot(insights)
assert loaded == snap
assert graph_diff(loaded, snap)["summary"] == {"pages": 0, "edges": 0}
def test_load_snapshot_missing(self, tmp_path):
p = tmp_path / "x.md"
p.write_text("no snapshot here")
assert load_snapshot(p) is None
assert load_snapshot(tmp_path / "absent.md") is None
def test_analyse_with_previous(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
prev = snapshot(outgoing)
_page(simple_vault, "g", ["a"])
result = analyse_vault(simple_vault, previous_snapshot=prev, include_snapshot=True)
assert result["diff"]["added_pages"] == ["g"]
assert "snapshot" in result
class TestSuggestQuestions:
def test_bridge_question_first(self, barbell_vault):
outgoing, _ = parse_vault_graph(barbell_vault)
comms = detect_communities_greedy(outgoing)
labels = {i: f"c{i}" for i in range(len(comms))}
bridges = bridge_pages(outgoing, comms)
qs = suggest_questions(outgoing, comms, labels, bridges)
assert qs
assert qs[0]["type"] == "bridge_node" and "`m`" in qs[0]["question"]
def test_isolated_question(self, simple_vault):
outgoing, _ = parse_vault_graph(simple_vault)
qs = suggest_questions(outgoing, [], {}, [])
assert any(q["type"] == "isolated_nodes" for q in qs)
def test_no_signal(self):
qs = suggest_questions({"a": ["b"], "b": ["a"]}, [{"a", "b"}], {0: "x"}, [])
assert qs == [] or qs[0]["type"] in ("no_signal", "isolated_nodes")
def test_included_in_analyse(self, barbell_vault):
result = analyse_vault(barbell_vault)
assert "suggested_questions" in result and "bridges" in result
assert result["stats"]["density"] > 0
class TestSurprisingDedup:
def test_note_and_pair_dedup(self, vault):
# Hub h in community 0 links to 3 different pages of community 1;
# community 1 also has one link to community 2. Pair (1,2) must not
# be pushed out by the three (0,1) edges.
_page(vault, "h", ["p", "q", "r"])
_page(vault, "p", ["q"]); _page(vault, "q", ["p"]); _page(vault, "r", ["p", "s"])
_page(vault, "s", ["t"]); _page(vault, "t", ["s"])
outgoing, _ = parse_vault_graph(vault)
comms = [{"h"}, {"p", "q", "r"}, {"s", "t"}]
sc = surprising_connections(outgoing, comms, top_n=2)
assert all("note" in x for x in sc)
pairs = {tuple(sorted(x["note"].replace("bridges community ", "").split(" -> community "))) for x in sc}
assert pairs == {("0", "1"), ("1", "2")}
class TestGraphAnalyseCLIQueryModes:
def _run(self, *argv):
return subprocess.run(
[sys.executable, "-m", "obsidian_wiki.cli", "graph-analyse", *argv],
capture_output=True, text=True,
)
def test_path(self, barbell_vault):
proc = self._run(str(barbell_vault), "--path", "b", "y")
assert proc.returncode == 0, proc.stderr
data = json.loads(proc.stdout)
assert data["hops"] == 4 and "m" in data["path"]
def test_around(self, barbell_vault):
proc = self._run(str(barbell_vault), "--around", "m", "--depth", "1")
assert proc.returncode == 0, proc.stderr
data = json.loads(proc.stdout)
assert {p["page"] for p in data["pages"]} == {"a", "x"}
def test_diff_against_and_snapshot(self, simple_vault, tmp_path):
proc = self._run(str(simple_vault), "--snapshot")
snap = json.loads(proc.stdout)["snapshot"]
prev = tmp_path / "_insights.md"
prev.write_text("<!-- GRAPH_SNAPSHOT: " + json.dumps(snap) + " -->")
_page(simple_vault, "new", ["a"])
proc = self._run(str(simple_vault), "--diff-against", str(prev))
assert proc.returncode == 0, proc.stderr
assert json.loads(proc.stdout)["diff"]["added_pages"] == ["new"]
class TestUniqueCommunityLabels:
def test_labels_disambiguated(self, vault):
# two disconnected clusters both dominated by tag `x`
_page(vault, "a", ["b"], tags=["x", "left"]); _page(vault, "b", ["a"], tags=["x"])
_page(vault, "c", ["d"], tags=["x", "right"]); _page(vault, "d", ["c"], tags=["x"])
result = analyse_vault(vault)
labels = [c["label"] for c in result["communities"]]
assert len(labels) == len(set(labels))
assert "x" in labels and any(l.startswith("x/") for l in labels)
# ---------------------------------------------------------------------------
# Shared page selection + betweenness cache
# ---------------------------------------------------------------------------
from obsidian_wiki.graph_analysis import ( # noqa: E402
CACHE_FILENAME,
SKIP_ROOT_FILES,
betweenness_cached,
graph_fingerprint,
is_wiki_page,
iter_pages,
)
class TestPageSelection:
def test_root_bookkeeping_excluded(self, vault):
_page(vault, "a", ["b"]); _page(vault, "b", [])
for name in SKIP_ROOT_FILES:
(vault / f"{name}.md").write_text("---\ntitle: x\n---\n[[a]]\n")
out, _ = parse_vault_graph(vault)
assert set(out) == {"a", "b"}
def test_nested_index_is_a_real_page(self, vault):
"""Only ROOT index/log/hot are bookkeeping — concepts/index.md is content."""
(vault / "concepts").mkdir()
(vault / "concepts" / "index.md").write_text("---\ntitle: Concept Index\n---\n[[a]]\n")
_page(vault, "a", [])
out, _ = parse_vault_graph(vault)
assert "index" in out
assert is_wiki_page(vault / "concepts" / "index.md", vault)
assert not is_wiki_page(vault / "index.md", vault)
def test_skip_dirs_excluded(self, vault):
for d in ("_raw", "_meta", "_readouts", "_staging"):
(vault / d).mkdir()
(vault / d / "note.md").write_text("---\ntitle: x\n---\n")
_page(vault, "a", [])
assert [p.stem for p in iter_pages(vault)] == ["a"]
def test_iter_pages_is_sorted(self, vault):
for n in ("zebra", "alpha", "middle"):
_page(vault, n, [])
assert [p.stem for p in iter_pages(vault)] == ["alpha", "middle", "zebra"]
def test_path_outside_vault_is_not_a_page(self, vault, tmp_path):
assert not is_wiki_page(tmp_path / "elsewhere.md", vault)
class TestGraphFingerprint:
def test_stable_across_calls(self, simple_vault):
out, _ = parse_vault_graph(simple_vault)
assert graph_fingerprint(out) == graph_fingerprint(out)
def test_changes_when_an_edge_changes(self, simple_vault):
out, _ = parse_vault_graph(simple_vault)
before = graph_fingerprint(out)
_page(simple_vault, "e", ["f", "a"]) # add an edge
out2, _ = parse_vault_graph(simple_vault)
assert graph_fingerprint(out2) != before
def test_insensitive_to_key_order(self, simple_vault):
out, _ = parse_vault_graph(simple_vault)
shuffled = {k: out[k] for k in reversed(list(out))}
assert graph_fingerprint(shuffled) == graph_fingerprint(out)
class TestBetweennessCache:
def test_matches_uncached(self, barbell_vault):
out, _ = parse_vault_graph(barbell_vault)
exact = betweenness_centrality(out)
cached = betweenness_cached(out, barbell_vault)
assert all(abs(exact[k] - cached[k]) < 1e-12 for k in exact)
def test_no_cache_file_for_small_vault(self, barbell_vault):
"""Tiny graphs recompute in microseconds — don't litter the vault."""
out, _ = parse_vault_graph(barbell_vault)
betweenness_cached(out, barbell_vault)
assert not (barbell_vault / CACHE_FILENAME).exists()
def test_none_vault_still_works(self, barbell_vault):
out, _ = parse_vault_graph(barbell_vault)
assert betweenness_cached(out, None) == betweenness_centrality(out)
def test_corrupt_cache_fails_open(self, barbell_vault):
(barbell_vault / CACHE_FILENAME).write_text("{ not json")
out, _ = parse_vault_graph(barbell_vault)
assert betweenness_cached(out, barbell_vault) == betweenness_centrality(out)
def test_hit_is_used_when_present(self, barbell_vault):
"""A planted cache entry is returned verbatim — proves the key matches."""
out, _ = parse_vault_graph(barbell_vault)
key = f"{graph_fingerprint(out)}:1"
planted = {n: 0.5 for n in out}
(barbell_vault / CACHE_FILENAME).write_text(
json.dumps({"version": 1, "betweenness": {key: planted}})
)
assert betweenness_cached(out, barbell_vault) == planted
def test_stale_key_is_ignored(self, barbell_vault):
out, _ = parse_vault_graph(barbell_vault)
(barbell_vault / CACHE_FILENAME).write_text(
json.dumps({"version": 1, "betweenness": {"someotherkey:1": {n: 9.0 for n in out}}})
)
assert betweenness_cached(out, barbell_vault) == betweenness_centrality(out)
def test_unwritable_vault_does_not_raise(self, barbell_vault, tmp_path):
out, _ = parse_vault_graph(barbell_vault)
assert betweenness_cached(out, tmp_path / "does" / "not" / "exist")
class TestCommunityQuality:
def test_barbell_splits_into_two(self, barbell_vault):
"""A sequential label-propagation sweep collapses this into one group."""
out, _ = parse_vault_graph(barbell_vault)
comms = detect_communities_greedy(out)
assert len(comms) == 2
assert {"x", "y", "z"} in [set(c) for c in comms]
def test_deterministic_across_runs(self, barbell_vault):
out, _ = parse_vault_graph(barbell_vault)
sigs = {
json.dumps(sorted(sorted(c) for c in detect_communities_greedy(out)))
for _ in range(10)
}
assert len(sigs) == 1
class TestGitignoreBackfill:
def test_new_gitignore_ignores_cache(self):
from obsidian_wiki.sync import GITIGNORE_CONTENT
assert ".graph-cache.json" in GITIGNORE_CONTENT
def test_existing_gitignore_is_never_edited_only_advised(self, tmp_path, monkeypatch):
"""The user's .gitignore is theirs — advise, don't mutate."""
import obsidian_wiki.sync as sync
v = tmp_path / "vault"; v.mkdir()
(v / ".gitignore").write_text(".env\n_raw/\n")
monkeypatch.setattr(sync, "_git", lambda *a, **k: type("P", (), {"returncode": 0, "stderr": "", "stdout": ""})())
msgs = sync.configure_sync(v, "git@example.com:me/vault.git")
assert (v / ".gitignore").read_text() == ".env\n_raw/\n" # byte-identical
assert any(".graph-cache.json" in m for m in msgs)
def test_no_advice_when_already_ignored(self, tmp_path, monkeypatch):
import obsidian_wiki.sync as sync
v = tmp_path / "vault"; v.mkdir()
(v / ".gitignore").write_text(".graph-cache.json\n")
monkeypatch.setattr(sync, "_git", lambda *a, **k: type("P", (), {"returncode": 0, "stderr": "", "stdout": ""})())
assert not any(".graph-cache.json" in m for m in sync.configure_sync(v, "git@example.com:me/vault.git"))
Reference in New Issue View Git Blame Copy Permalink