92ea55f77d
Two bugs found while filing the Obsidian Sorter note itself: 1. proposed_toc went stale when proposed_folder was edited by hand. The watcher now re-resolves the index line from the folder on every pass. 2. Re-proposing after a body edit clobbered a hand-edited proposed_folder. The watcher's own suggestion is now kept separately in proposed_folder_auto, so a difference marks a human choice, which wins. Adds two guard tests. 17/17 passing.
430 lines
17 KiB
Python
430 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Functional test suite for obsidian-sorter.
|
|
|
|
Runs against an ISOLATED COPY of the vault so the mirror is never touched.
|
|
Exercises the full pipeline: classify -> plan -> apply -> verify.
|
|
|
|
Usage:
|
|
python3 tests/run_tests.py [--vault ./test-vault] [--no-jev]
|
|
|
|
With --no-jev only the deterministic tier runs (faster, no API cost), which is
|
|
useful for testing the writer mechanics without model calls.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(REPO))
|
|
|
|
from sorter import frontmatter # noqa: E402
|
|
|
|
GENERIC_NOTE = """---
|
|
created: 2026-09-19 17:56
|
|
modified: 2026-09-19 17:56
|
|
type: note
|
|
tags: {tags}
|
|
aliases: []
|
|
---
|
|
# [[{title}]]
|
|
|
|
{body}
|
|
"""
|
|
|
|
GENERIC_WITH_USERDATA = """---
|
|
created: 2026-08-01 09:00
|
|
modified: 2026-08-02 10:00
|
|
type: note
|
|
tags:
|
|
- my-own-tag
|
|
- research
|
|
aliases:
|
|
- alt name
|
|
id: 1779253896-DPHX
|
|
areas:
|
|
- "[[Home Assistant]]"
|
|
---
|
|
# [[{title}]]
|
|
|
|
{body}
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------- test cases
|
|
|
|
CASES = [
|
|
# The BUCKET (project/area/family) is asserted strictly — it is reliable.
|
|
# The SUB-FOLDER is asserted only with --strict-folders, because JEV is
|
|
# probabilistic (~74% folder accuracy) and exact folders can vary run to run.
|
|
dict(
|
|
name="recipe -> family",
|
|
filename="Sourdough Bread Recipe.md",
|
|
body="My sourdough recipe: 500g flour, 350g water, 100g starter, 10g salt. "
|
|
"Bulk ferment 4 hours, shape, prove overnight, bake at 230C.",
|
|
expect_type="family",
|
|
expect_bucket="400 Personal Family",
|
|
expect_subfolder="450 Recipes",
|
|
),
|
|
dict(
|
|
name="unfinished electronics build -> project",
|
|
filename="Garage Door Sensor Build.md",
|
|
body="Plan to build a garage door sensor using an ESP32 and a reed switch. "
|
|
"TODO: order parts, write firmware. Needs to be built before winter.",
|
|
expect_type="project",
|
|
expect_bucket="200 projects",
|
|
expect_subfolder="250 Electronics",
|
|
),
|
|
dict(
|
|
name="ops documentation -> area",
|
|
filename="Nginx Reverse Proxy Notes.md",
|
|
body="Documentation for the nginx reverse proxy running on .13. It is live. "
|
|
"Configuration reference and TLS certificate renewal steps.",
|
|
expect_type="area",
|
|
expect_bucket="300 areas",
|
|
expect_subfolder="360 Dev-Ops",
|
|
),
|
|
dict(
|
|
name="family event -> family",
|
|
filename="Nan Birthday Party.md",
|
|
body="Photos and notes from Nan's 80th birthday at the community hall. "
|
|
"Family gathered for speeches and catering.",
|
|
expect_type="family",
|
|
expect_bucket="400 Personal Family",
|
|
expect_subfolder="460 Family Archive",
|
|
),
|
|
dict(
|
|
name="ai research -> area",
|
|
filename="Vector Database Comparison.md",
|
|
body="Research notes comparing vector database options for RAG: pgvector "
|
|
"vs chroma vs qdrant. Embeddings, recall, operational complexity.",
|
|
expect_type="area",
|
|
expect_bucket="300 areas",
|
|
expect_subfolder="355 AI Research",
|
|
),
|
|
# A tag is a human decision — it must be honoured every time.
|
|
dict(
|
|
name="tagged project -> project (tag wins)",
|
|
filename="Tagged Build Job.md",
|
|
body="Some reference notes about a thing.",
|
|
tags='[project]',
|
|
expect_type="project",
|
|
expect_bucket="200 projects",
|
|
expect_subfolder="200 projects",
|
|
),
|
|
]
|
|
|
|
|
|
# ------------------------------------------------------------------- helpers
|
|
|
|
def run(cmd: list[str], cwd: Path) -> subprocess.CompletedProcess:
|
|
return subprocess.run(cmd, cwd=str(cwd), capture_output=True, text=True)
|
|
|
|
|
|
def read_fm(path: Path) -> tuple[dict, str]:
|
|
text = path.read_text(encoding="utf-8")
|
|
fm, body, ok = frontmatter.parse(text)
|
|
return fm, body
|
|
|
|
|
|
def write_inbox(vault: Path, inbox: str, filename: str, content: str) -> Path:
|
|
p = vault / inbox / filename
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(content, encoding="utf-8")
|
|
return p
|
|
|
|
|
|
# ---------------------------------------------------------------------- main
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--vault", default=str(REPO / "test-vault"))
|
|
ap.add_argument("--inbox", default="010 inbox")
|
|
ap.add_argument("--no-jev", action="store_true")
|
|
ap.add_argument("--strict-folders", action="store_true",
|
|
help="also assert exact sub-folders (JEV is probabilistic; may be flaky)")
|
|
ap.add_argument("--keep", action="store_true", help="keep the temp vault")
|
|
args = ap.parse_args()
|
|
|
|
src = Path(args.vault).resolve()
|
|
if not src.is_dir():
|
|
print(f"error: vault not found: {src}", file=sys.stderr)
|
|
return 1
|
|
|
|
tmp = Path(tempfile.mkdtemp(prefix="sorter-test-"))
|
|
vault = tmp / "vault"
|
|
print(f"Isolated test vault: {vault}\n")
|
|
shutil.copytree(src, vault, symlinks=False,
|
|
ignore=shutil.ignore_patterns(".git", ".trash", ".venv",
|
|
"graphify-out*", ".obsidian"))
|
|
|
|
jev_flags = [] if args.no_jev else ["--jev"]
|
|
passed = failed = 0
|
|
details: list[str] = []
|
|
|
|
try:
|
|
# ---- inject notes -------------------------------------------------
|
|
for c in CASES:
|
|
write_inbox(vault, args.inbox, c["filename"], GENERIC_NOTE.format(
|
|
title=c["filename"][:-3], body=c["body"], tags=c.get("tags", "[]")))
|
|
|
|
# note with pre-existing user data (merge test)
|
|
write_inbox(vault, args.inbox, "Merge Test Note.md",
|
|
GENERIC_WITH_USERDATA.format(
|
|
title="Merge Test Note",
|
|
body="A recipe and family note that also carries user data."))
|
|
CASES.append(dict(
|
|
name="frontmatter merge",
|
|
filename="Merge Test Note.md",
|
|
expect_type=None, expect_folder_contains=None, expect_toc_section=None,
|
|
))
|
|
|
|
# ---- dry run must not write ---------------------------------------
|
|
before = set(p.name for p in (vault / args.inbox).glob("*.md")
|
|
if not p.name.startswith("_"))
|
|
r = run([sys.executable, "-m", "sorter", "--vault", str(vault),
|
|
"--inbox", args.inbox, "sort-inbox"] + jev_flags, REPO)
|
|
after = set(p.name for p in (vault / args.inbox).glob("*.md")
|
|
if not p.name.startswith("_"))
|
|
if before == after and r.returncode == 0:
|
|
print("PASS dry run leaves the vault untouched")
|
|
passed += 1
|
|
else:
|
|
print(f"FAIL dry run changed files (rc={r.returncode})")
|
|
print(r.stdout[-800:], r.stderr[-400:])
|
|
failed += 1
|
|
|
|
# ---- apply ---------------------------------------------------------
|
|
r = run([sys.executable, "-m", "sorter", "--vault", str(vault),
|
|
"--inbox", args.inbox, "sort-inbox"] + jev_flags + ["--apply", "--yes"], REPO)
|
|
if r.returncode != 0:
|
|
print(f"FAIL apply exited {r.returncode}")
|
|
print(r.stdout[-2000:])
|
|
print(r.stderr[-1000:])
|
|
return 1
|
|
|
|
# ---- verify each case ---------------------------------------------
|
|
for c in CASES:
|
|
name = c["name"]
|
|
fname = c["filename"]
|
|
problems: list[str] = []
|
|
|
|
# found somewhere outside the inbox?
|
|
matches = [p for p in vault.rglob(fname)
|
|
if args.inbox not in p.parts and ".trash" not in p.parts]
|
|
if not matches:
|
|
details.append(f"FAIL {name}: '{fname}' not filed anywhere")
|
|
failed += 1
|
|
continue
|
|
dest = matches[0]
|
|
fm, body = read_fm(dest)
|
|
|
|
if c.get("expect_type") and fm.get("type") != c["expect_type"]:
|
|
problems.append(f"type={fm.get('type')!r} expected {c['expect_type']!r}")
|
|
if fm.get("status") != "active":
|
|
problems.append(f"status={fm.get('status')!r}")
|
|
if not fm.get("title"):
|
|
problems.append("no title")
|
|
if re.search(r"\{\{.*?\}\}", json.dumps(fm)):
|
|
problems.append("unfilled template placeholder in frontmatter")
|
|
if not body.strip():
|
|
problems.append("body was emptied")
|
|
if c.get("expect_bucket"):
|
|
if c["expect_bucket"].lower() not in str(dest.parent).lower():
|
|
problems.append(f"folder={dest.parent.name!r} expected to be in "
|
|
f"bucket {c['expect_bucket']!r}")
|
|
if args.strict_folders and c.get("expect_subfolder"):
|
|
if c["expect_subfolder"].lower() not in str(dest.parent).lower():
|
|
problems.append(f"folder={dest.parent.name!r} expected "
|
|
f"{c['expect_subfolder']!r}")
|
|
if c.get("expect_toc_section"):
|
|
idx = vault / "100 Table of Contents"
|
|
listed = False
|
|
for f in idx.glob("*.md"):
|
|
if f"[[{fname[:-3]}]]" in f.read_text(encoding="utf-8"):
|
|
listed = True
|
|
if not listed:
|
|
problems.append(f"not listed in any index (expected ~{c['expect_toc_section']!r})")
|
|
|
|
if problems:
|
|
details.append(f"FAIL {name}: " + "; ".join(problems))
|
|
failed += 1
|
|
else:
|
|
print(f"PASS {name} -> {dest.relative_to(vault)}")
|
|
passed += 1
|
|
|
|
# ---- unit: safety guards (in-process, no API calls) -------------
|
|
from sorter.models import Proposal
|
|
from sorter.vault import Vault as _V
|
|
from sorter.writer import Writer as _W
|
|
|
|
uv = _V(str(vault))
|
|
uw = _W(uv)
|
|
dummy = Proposal(
|
|
source_rel="x", note_type="area",
|
|
target_folder="300 areas/360 Dev-Ops Network Computers",
|
|
toc_section="", template="Area.md", confidence=0.9, method="test",
|
|
)
|
|
|
|
# guard 1: Untitled is never moved
|
|
up = write_inbox(vault, args.inbox, "Untitled 99.md",
|
|
GENERIC_NOTE.format(title="Untitled 99", body="stray content", tags="[]"))
|
|
uplan = uw.plan(uv.read_note(up), dummy)
|
|
if not uplan.writable and "Untitled" in (uplan.blocked_reason or ""):
|
|
print("PASS guard: Untitled note is blocked")
|
|
passed += 1
|
|
else:
|
|
details.append("FAIL guard: Untitled note was NOT blocked")
|
|
failed += 1
|
|
|
|
# guard 2: existing destination blocks the plan
|
|
dest_dir = vault / "300 areas/360 Dev-Ops Network Computers"
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
(dest_dir / "Conflict Test.md").write_text(
|
|
"---\ntitle: Conflict Test\ntype: subject\nstatus: active\n---\n\n# Conflict\n",
|
|
encoding="utf-8")
|
|
cp = write_inbox(vault, args.inbox, "Conflict Test.md",
|
|
GENERIC_NOTE.format(title="Conflict Test", body="duplicate", tags="[]"))
|
|
cplan = uw.plan(uv.read_note(cp), dummy)
|
|
if not cplan.writable and "already exists" in (cplan.blocked_reason or ""):
|
|
print("PASS guard: existing destination blocks the plan")
|
|
passed += 1
|
|
else:
|
|
details.append(f"FAIL guard: destination collision not caught ({cplan.blocked_reason})")
|
|
failed += 1
|
|
|
|
# guard 3: .trash is never removed
|
|
trashed = uw.vault.root / ".trash"
|
|
if not trashed.exists():
|
|
trashed.mkdir(parents=True, exist_ok=True)
|
|
(trashed / "old.md").write_text("---\ntype: subject\n---\n\nx\n", encoding="utf-8")
|
|
tp = uw.vault.root / ".trash" / "old.md"
|
|
tnote = uv.read_note(tp)
|
|
if tnote is not None:
|
|
tplan = uw.plan(tnote, dummy)
|
|
_ok, log = uw.apply(tplan, source_path=str(tp))
|
|
if tp.exists() and any(".trash" in l for l in log):
|
|
print("PASS guard: .trash file is never removed")
|
|
passed += 1
|
|
else:
|
|
details.append("FAIL guard: .trash file was modified/removed")
|
|
failed += 1
|
|
|
|
# cleanup the injected guard fixtures
|
|
for p in (up, cp, dest_dir / "Conflict Test.md", dest_dir / "old.md", tp):
|
|
try:
|
|
p.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
# ---- watcher: the owner's hand-picked folder must survive ----------
|
|
from sorter.watch import user_selected_folder
|
|
from sorter.models import Note as _Note
|
|
|
|
def _n(chosen, auto):
|
|
return _Note(path="x", rel_path="010 inbox/x.md", filename="x.md",
|
|
stem="x", title="x", note_type="note", status=None,
|
|
fm={"proposed_folder": chosen, "proposed_folder_auto": auto})
|
|
|
|
if user_selected_folder(_n("300 areas/A", "200 projects/B")) == "300 areas/A":
|
|
print("PASS guard: a hand-edited folder is treated as the owner's choice")
|
|
passed += 1
|
|
else:
|
|
details.append("FAIL guard: hand-edited folder not detected")
|
|
failed += 1
|
|
|
|
if user_selected_folder(_n("200 projects/B", "200 projects/B")) is None:
|
|
print("PASS guard: an untouched proposal is not treated as customised")
|
|
passed += 1
|
|
else:
|
|
details.append("FAIL guard: untouched proposal wrongly seen as customised")
|
|
failed += 1
|
|
|
|
# ---- inbox cleared --------------------------------------------------
|
|
left = [p for p in (vault / args.inbox).glob("*.md")
|
|
if not p.name.startswith("_")]
|
|
if not left:
|
|
print("PASS inbox emptied")
|
|
passed += 1
|
|
else:
|
|
print(f"FAIL inbox still has: {[p.name for p in left]}")
|
|
failed += 1
|
|
|
|
# ---- merge assertions ----------------------------------------------
|
|
merge_matches = [p for p in vault.rglob("Merge Test Note.md")
|
|
if args.inbox not in p.parts]
|
|
if merge_matches:
|
|
fm, _ = read_fm(merge_matches[0])
|
|
mp = []
|
|
if "my-own-tag" not in (fm.get("tags") or []):
|
|
mp.append("user tag 'my-own-tag' lost")
|
|
if "research" not in (fm.get("tags") or []):
|
|
mp.append("user tag 'research' lost")
|
|
if fm.get("id") != "1779253896-DPHX":
|
|
mp.append("id lost")
|
|
if fm.get("created") != "2026-08-01 09:00":
|
|
mp.append(f"created changed to {fm.get('created')!r}")
|
|
if not fm.get("aliases"):
|
|
mp.append("aliases lost")
|
|
if mp:
|
|
details.append("FAIL frontmatter merge: " + "; ".join(mp))
|
|
failed += 1
|
|
else:
|
|
print("PASS frontmatter merge (tags, aliases, id, created preserved)")
|
|
passed += 1
|
|
|
|
# ---- idempotency ----------------------------------------------------
|
|
toc_before = {f.name: f.read_text(encoding="utf-8")
|
|
for f in (vault / "100 Table of Contents").glob("*.md")}
|
|
run([sys.executable, "-m", "sorter", "--vault", str(vault),
|
|
"--inbox", args.inbox, "sort-inbox"] + jev_flags + ["--apply", "--yes"], REPO)
|
|
toc_after = {f.name: f.read_text(encoding="utf-8")
|
|
for f in (vault / "100 Table of Contents").glob("*.md")}
|
|
if toc_before == toc_after:
|
|
print("PASS idempotent (second run changed no index file)")
|
|
passed += 1
|
|
else:
|
|
changed = [k for k in toc_before if toc_before[k] != toc_after.get(k)]
|
|
details.append(f"FAIL second run altered index files: {changed}")
|
|
failed += 1
|
|
|
|
# ---- duplicate check -------------------------------------------------
|
|
idx_text = "\n".join(f.read_text(encoding="utf-8")
|
|
for f in (vault / "100 Table of Contents").glob("*.md"))
|
|
dups = [c["filename"][:-3] for c in CASES
|
|
if idx_text.count(f"[[{c['filename'][:-3]}]]") > 1]
|
|
if dups:
|
|
details.append(f"FAIL duplicate index entries: {dups}")
|
|
failed += 1
|
|
else:
|
|
print("PASS no duplicate index entries")
|
|
passed += 1
|
|
|
|
finally:
|
|
if args.keep:
|
|
print(f"\n(temp vault kept at {vault})")
|
|
else:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
print(f"\n{'='*64}")
|
|
print(f"PASSED: {passed} FAILED: {failed}")
|
|
if details:
|
|
print("\nDetails:")
|
|
for d in details:
|
|
print(f" {d}")
|
|
print("=" * 64)
|
|
return 0 if failed == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|