#!/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: a recorded folder is never re-decided ---------------- from sorter.watch import existing_folder from sorter.models import Note as _Note def _n(chosen): 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}) if existing_folder(_n("300 areas/A")) == "300 areas/A": print("PASS guard: a recorded folder is treated as already decided") passed += 1 else: details.append("FAIL guard: recorded folder not detected") failed += 1 if existing_folder(_n("")) is None: print("PASS guard: no recorded folder means the watcher may suggest one") passed += 1 else: details.append("FAIL guard: empty folder wrongly treated as set") 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())