feat(p1): ULID identity, ASCII slugs, NFC, frontmatter round-trip, atomic vault I/O
Frontmatter uses ruamel round-trip so unknown keys, order and comments survive an edit (§4). Writes are tmp→fsync→os.replace with a pre-write history snapshot, and iter_files skips *.tmp so a crashed write is never visible (§14.6, §14.8).
This commit is contained in:
Markus committed
1 parent
d9b42e3432
commit
8f58119a74
9 files changed
+732
No files matched your search
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,54 @@
|
||||
"""Identity, slugs and Unicode normalisation (§4).
|
||||
|
||||
Identity is the ULID. The slug is cosmetic and always ASCII (§14.8).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
|
||||
from ulid import ULID
|
||||
|
||||
_ULID_RE = re.compile(r"^[0-9A-HJKMNP-TV-Z]{26}$")
|
||||
_TRANSLIT = str.maketrans(
|
||||
{
|
||||
"ä": "ae", "ö": "oe", "ü": "ue", "Ä": "ae", "Ö": "oe", "Ü": "ue",
|
||||
"ß": "ss", "å": "aa", "æ": "ae", "ø": "oe", "đ": "d", "ł": "l", "þ": "th",
|
||||
}
|
||||
)
|
||||
MAX_SLUG = 60
|
||||
|
||||
|
||||
def new_id() -> str:
|
||||
return str(ULID())
|
||||
|
||||
|
||||
def is_ulid(value: object) -> bool:
|
||||
return isinstance(value, str) and bool(_ULID_RE.match(value))
|
||||
|
||||
|
||||
def nfc(text: str) -> str:
|
||||
"""NFC-normalise. Everything read from or written to disk passes through here (§14.8)."""
|
||||
return unicodedata.normalize("NFC", text)
|
||||
|
||||
|
||||
def slugify(title: str, max_len: int = MAX_SLUG) -> str:
|
||||
s = nfc(title or "").strip().lower().translate(_TRANSLIT)
|
||||
s = unicodedata.normalize("NFKD", s)
|
||||
s = s.encode("ascii", "ignore").decode("ascii")
|
||||
s = re.sub(r"[^a-z0-9]+", "-", s).strip("-")
|
||||
s = re.sub(r"-{2,}", "-", s)
|
||||
if len(s) > max_len:
|
||||
s = s[:max_len].rstrip("-")
|
||||
return s or "note"
|
||||
|
||||
|
||||
def filename_for(title: str, note_id: str) -> str:
|
||||
return f"{slugify(title)}--{note_id}.md"
|
||||
|
||||
|
||||
def id_from_filename(name: str) -> str | None:
|
||||
stem = name[:-3] if name.endswith(".md") else name
|
||||
_, _, tail = stem.rpartition("--")
|
||||
return tail if is_ulid(tail) else None
|
||||
@@ -0,0 +1,233 @@
|
||||
"""Note format: YAML frontmatter + markdown body (§4).
|
||||
|
||||
Unknown frontmatter keys, their order and their comments must round-trip
|
||||
unchanged, so parsing uses ruamel's round-trip mode rather than a plain loader.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from ruamel.yaml import YAML
|
||||
from ruamel.yaml.comments import CommentedMap
|
||||
|
||||
from mdmem.config import NOTE_TYPES
|
||||
from mdmem.ids import filename_for, new_id, nfc
|
||||
|
||||
_FM_RE = re.compile(r"\A---[ \t]*\r?\n(.*?)\r?\n---[ \t]*\r?\n?", re.DOTALL)
|
||||
_WIKILINK_RE = re.compile(r"\[\[([^\[\]|#]+?)(?:#[^\[\]|]*)?(?:\|[^\[\]]*)?\]\]")
|
||||
_HEADING_RE = re.compile(r"^(#{1,6})\s+(.*?)\s*#*\s*$", re.MULTILINE)
|
||||
|
||||
CANONICAL_ORDER = [
|
||||
"id", "title", "type", "summary", "keywords", "tags", "relations",
|
||||
"supersedes", "superseded_by", "revises", "possible_duplicates",
|
||||
"observed_at", "valid_from", "valid_until", "source", "created", "updated",
|
||||
]
|
||||
|
||||
|
||||
def _yaml() -> YAML:
|
||||
y = YAML()
|
||||
y.preserve_quotes = True
|
||||
y.width = 4096
|
||||
y.indent(mapping=2, sequence=4, offset=2)
|
||||
return y
|
||||
|
||||
|
||||
def utcnow() -> str:
|
||||
return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
|
||||
def parse(text: str) -> tuple[CommentedMap, str]:
|
||||
"""Split a note into (frontmatter, body). Both are NFC-normalised."""
|
||||
text = nfc(text.replace("\r\n", "\n"))
|
||||
m = _FM_RE.match(text)
|
||||
if not m:
|
||||
return CommentedMap(), text
|
||||
raw = m.group(1)
|
||||
body = text[m.end() :]
|
||||
try:
|
||||
data = _yaml().load(raw)
|
||||
except Exception: # a hand-edited file with broken YAML must not kill the scan
|
||||
data = None
|
||||
if data is None:
|
||||
data = CommentedMap()
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("frontmatter must be a mapping")
|
||||
return data, body
|
||||
|
||||
|
||||
def dumps(fm: dict[str, Any], body: str) -> str:
|
||||
buf = io.StringIO()
|
||||
if fm:
|
||||
_yaml().dump(fm, buf)
|
||||
fm_text = buf.getvalue().rstrip("\n")
|
||||
body = nfc(body).rstrip("\n")
|
||||
return f"---\n{fm_text}\n---\n{body}\n"
|
||||
|
||||
|
||||
def _as_list(v: Any) -> list[str]:
|
||||
if v is None:
|
||||
return []
|
||||
if isinstance(v, str):
|
||||
return [v] if v.strip() else []
|
||||
if isinstance(v, (list, tuple)):
|
||||
return [str(x) for x in v if x is not None and str(x).strip()]
|
||||
return []
|
||||
|
||||
|
||||
@dataclass
|
||||
class Note:
|
||||
fm: CommentedMap
|
||||
body: str
|
||||
state: str = "inbox"
|
||||
path: Path | None = None
|
||||
file_hash: str = ""
|
||||
mtime: float = 0.0
|
||||
_errors: list[str] = field(default_factory=list)
|
||||
|
||||
# -- construction ------------------------------------------------------
|
||||
@classmethod
|
||||
def from_text(
|
||||
cls, text: str, state: str = "inbox", path: Path | None = None, **kw: Any
|
||||
) -> Note:
|
||||
fm, body = parse(text)
|
||||
return cls(fm=fm, body=body, state=state, path=path, **kw)
|
||||
|
||||
@classmethod
|
||||
def new(
|
||||
cls,
|
||||
*,
|
||||
title: str,
|
||||
body: str,
|
||||
type: str = "note",
|
||||
summary: str = "",
|
||||
keywords: list[str] | None = None,
|
||||
tags: list[str] | None = None,
|
||||
relations: dict[str, list[str]] | None = None,
|
||||
observed_at: str | None = None,
|
||||
valid_from: str | None = None,
|
||||
valid_until: str | None = None,
|
||||
source: dict[str, Any] | None = None,
|
||||
supersedes: list[str] | None = None,
|
||||
revises: str | None = None,
|
||||
note_id: str | None = None,
|
||||
state: str = "inbox",
|
||||
) -> Note:
|
||||
now = utcnow()
|
||||
fm = CommentedMap()
|
||||
fm["id"] = note_id or new_id()
|
||||
fm["title"] = nfc(title)
|
||||
fm["type"] = type if type in NOTE_TYPES else "note"
|
||||
fm["summary"] = nfc(summary or "")
|
||||
fm["keywords"] = [nfc(k) for k in (keywords or [])]
|
||||
fm["tags"] = [nfc(t) for t in (tags or [])]
|
||||
fm["relations"] = relations or {}
|
||||
fm["supersedes"] = supersedes or []
|
||||
fm["superseded_by"] = None
|
||||
fm["revises"] = revises
|
||||
fm["possible_duplicates"] = []
|
||||
fm["observed_at"] = observed_at or now
|
||||
fm["valid_from"] = valid_from
|
||||
fm["valid_until"] = valid_until
|
||||
fm["source"] = source or {}
|
||||
fm["created"] = now
|
||||
fm["updated"] = now
|
||||
return cls(fm=fm, body=nfc(body), state=state)
|
||||
|
||||
# -- accessors ---------------------------------------------------------
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
v = self.fm.get(key, default)
|
||||
return default if v is None else v
|
||||
|
||||
def set(self, key: str, value: Any) -> None:
|
||||
self.fm[key] = value
|
||||
|
||||
@property
|
||||
def id(self) -> str:
|
||||
return str(self.fm.get("id") or "")
|
||||
|
||||
@property
|
||||
def title(self) -> str:
|
||||
return str(self.fm.get("title") or "")
|
||||
|
||||
@property
|
||||
def type(self) -> str:
|
||||
return str(self.fm.get("type") or "note")
|
||||
|
||||
@property
|
||||
def summary(self) -> str:
|
||||
return str(self.fm.get("summary") or "")
|
||||
|
||||
@property
|
||||
def keywords(self) -> list[str]:
|
||||
return _as_list(self.fm.get("keywords"))
|
||||
|
||||
@property
|
||||
def tags(self) -> list[str]:
|
||||
return _as_list(self.fm.get("tags"))
|
||||
|
||||
@property
|
||||
def relations(self) -> dict[str, list[str]]:
|
||||
rel = self.fm.get("relations") or {}
|
||||
if not isinstance(rel, dict):
|
||||
return {}
|
||||
return {str(k): _as_list(v) for k, v in rel.items()}
|
||||
|
||||
@property
|
||||
def supersedes(self) -> list[str]:
|
||||
return _as_list(self.fm.get("supersedes"))
|
||||
|
||||
@property
|
||||
def superseded_by(self) -> str | None:
|
||||
v = self.fm.get("superseded_by")
|
||||
return str(v) if v else None
|
||||
|
||||
@property
|
||||
def revises(self) -> str | None:
|
||||
v = self.fm.get("revises")
|
||||
return str(v) if v else None
|
||||
|
||||
@property
|
||||
def possible_duplicates(self) -> list[str]:
|
||||
return _as_list(self.fm.get("possible_duplicates"))
|
||||
|
||||
@property
|
||||
def source(self) -> dict[str, Any]:
|
||||
v = self.fm.get("source")
|
||||
return dict(v) if isinstance(v, dict) else {}
|
||||
|
||||
@property
|
||||
def wikilinks(self) -> list[str]:
|
||||
seen: list[str] = []
|
||||
for m in _WIKILINK_RE.finditer(self.body):
|
||||
t = m.group(1).strip()
|
||||
if t and t not in seen:
|
||||
seen.append(t)
|
||||
return seen
|
||||
|
||||
def ts(self, key: str) -> str | None:
|
||||
v = self.fm.get(key)
|
||||
if v is None or v == "":
|
||||
return None
|
||||
if isinstance(v, datetime):
|
||||
return v.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
return str(v)
|
||||
|
||||
# -- serialisation -----------------------------------------------------
|
||||
def to_text(self) -> str:
|
||||
return dumps(self.fm, self.body)
|
||||
|
||||
def filename(self) -> str:
|
||||
return filename_for(self.title, self.id)
|
||||
|
||||
def touch(self) -> None:
|
||||
self.fm["updated"] = utcnow()
|
||||
|
||||
def headings(self) -> list[tuple[int, str, int]]:
|
||||
"""(level, text, start offset) for every ATX heading in the body."""
|
||||
return [(len(m.group(1)), m.group(2).strip(), m.start()) for m in _HEADING_RE.finditer(self.body)]
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Vault file I/O: atomic writes, history snapshots, lifecycle folders (§4).
|
||||
|
||||
Folder = lifecycle state. Identity is the ULID in the filename; the slug is cosmetic.
|
||||
All writes are tmp → fsync → os.replace, so a reader (or the watcher) never sees a
|
||||
partial file (§14.6).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import shutil
|
||||
from collections.abc import Iterator
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
from mdmem.config import STATES
|
||||
from mdmem.ids import id_from_filename, nfc
|
||||
from mdmem.notes import Note
|
||||
|
||||
TMP_SUFFIX = ".tmp"
|
||||
|
||||
|
||||
def sha256_text(text: str) -> str:
|
||||
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
class VaultError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class Vault:
|
||||
def __init__(self, root: Path, history_dir: Path) -> None:
|
||||
self.root = Path(root)
|
||||
self.history_dir = Path(history_dir)
|
||||
|
||||
# -- layout ------------------------------------------------------------
|
||||
def ensure_dirs(self) -> None:
|
||||
for state in STATES:
|
||||
(self.root / state).mkdir(parents=True, exist_ok=True)
|
||||
self.history_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def state_of(self, path: Path) -> str:
|
||||
try:
|
||||
rel = Path(path).resolve().relative_to(self.root.resolve())
|
||||
except ValueError as exc:
|
||||
raise VaultError(f"{path} is outside the vault") from exc
|
||||
state = rel.parts[0] if rel.parts else ""
|
||||
if state not in STATES:
|
||||
raise VaultError(f"{path} is not in a lifecycle folder")
|
||||
return state
|
||||
|
||||
def iter_files(self) -> Iterator[tuple[Path, str]]:
|
||||
for state in STATES:
|
||||
d = self.root / state
|
||||
if not d.is_dir():
|
||||
continue
|
||||
for p in sorted(d.rglob("*.md")):
|
||||
if p.name.endswith(TMP_SUFFIX) or ".history" in p.parts or not p.is_file():
|
||||
continue
|
||||
yield p, state
|
||||
|
||||
def find(self, note_id: str) -> Path | None:
|
||||
for p, _ in self.iter_files():
|
||||
if id_from_filename(p.name) == note_id:
|
||||
return p
|
||||
return None
|
||||
|
||||
# -- read / write ------------------------------------------------------
|
||||
def read(self, path: Path) -> Note:
|
||||
path = Path(path)
|
||||
raw = path.read_text(encoding="utf-8")
|
||||
note = Note.from_text(raw, state=self.state_of(path), path=path)
|
||||
note.file_hash = sha256_text(nfc(raw.replace("\r\n", "\n")))
|
||||
note.mtime = path.stat().st_mtime
|
||||
return note
|
||||
|
||||
def path_for(self, note: Note, state: str | None = None) -> Path:
|
||||
state = state or note.state
|
||||
if state not in STATES:
|
||||
raise VaultError(f"unknown state {state!r}")
|
||||
return self.root / state / note.filename()
|
||||
|
||||
def write(self, note: Note, state: str | None = None) -> Path:
|
||||
"""Write the note atomically, snapshotting whatever was there before."""
|
||||
state = state or note.state
|
||||
target = self.path_for(note, state)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
old_path = self.find(note.id)
|
||||
if old_path is not None and old_path.exists():
|
||||
self.snapshot(note.id, old_path.read_text(encoding="utf-8"))
|
||||
|
||||
text = note.to_text()
|
||||
self._atomic_write(target, text)
|
||||
|
||||
if old_path is not None and old_path != target:
|
||||
old_path.unlink(missing_ok=True)
|
||||
|
||||
note.state = state
|
||||
note.path = target
|
||||
note.file_hash = sha256_text(text)
|
||||
note.mtime = target.stat().st_mtime
|
||||
return target
|
||||
|
||||
def move(self, note: Note, state: str) -> Path:
|
||||
return self.write(note, state=state)
|
||||
|
||||
def delete(self, note_or_id: Note | str) -> None:
|
||||
note_id = note_or_id if isinstance(note_or_id, str) else note_or_id.id
|
||||
p = self.find(note_id)
|
||||
if p is not None:
|
||||
p.unlink(missing_ok=True)
|
||||
shutil.rmtree(self.history_dir / note_id, ignore_errors=True)
|
||||
|
||||
@staticmethod
|
||||
def _atomic_write(target: Path, text: str) -> None:
|
||||
tmp = target.with_name(target.name + TMP_SUFFIX)
|
||||
with open(tmp, "w", encoding="utf-8") as fh:
|
||||
fh.write(text)
|
||||
fh.flush()
|
||||
os.fsync(fh.fileno())
|
||||
os.replace(tmp, target)
|
||||
dir_fd = os.open(target.parent, os.O_RDONLY)
|
||||
try:
|
||||
os.fsync(dir_fd)
|
||||
finally:
|
||||
os.close(dir_fd)
|
||||
|
||||
# -- history -----------------------------------------------------------
|
||||
def snapshot(self, note_id: str, text: str) -> Path:
|
||||
d = self.history_dir / note_id
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
stamp = datetime.now(UTC).strftime("%Y-%m-%dT%H-%M-%S-%f")
|
||||
p = d / f"{stamp}.md"
|
||||
self._atomic_write(p, text)
|
||||
return p
|
||||
|
||||
def history(self, note_id: str) -> list[str]:
|
||||
d = self.history_dir / note_id
|
||||
if not d.is_dir():
|
||||
return []
|
||||
return sorted(p.stem for p in d.glob("*.md") if not p.name.endswith(TMP_SUFFIX))
|
||||
|
||||
def read_history(self, note_id: str, stamp: str) -> str:
|
||||
p = self.history_dir / note_id / f"{stamp}.md"
|
||||
if not p.is_file():
|
||||
raise VaultError(f"no revision {stamp} for {note_id}")
|
||||
return p.read_text(encoding="utf-8")
|
||||
|
||||
def writable(self) -> bool:
|
||||
try:
|
||||
self.root.mkdir(parents=True, exist_ok=True)
|
||||
probe = self.root / f".writecheck{TMP_SUFFIX}"
|
||||
probe.write_text("ok", encoding="utf-8")
|
||||
probe.unlink()
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
@@ -0,0 +1,26 @@
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def vault_root(tmp_path):
|
||||
root = tmp_path / "vault"
|
||||
root.mkdir()
|
||||
return root
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def settings(tmp_path, monkeypatch):
|
||||
from mdmem.config import Settings
|
||||
|
||||
for k in list(os.environ):
|
||||
if k.startswith(("EMBED_", "FUSION", "BM25_", "MIN_SCORE", "INBOX_", "FTS_")):
|
||||
monkeypatch.delenv(k, raising=False)
|
||||
monkeypatch.setenv("VAULT_DIR", str(tmp_path / "vault"))
|
||||
monkeypatch.setenv("INDEX_DB", str(tmp_path / "index.db"))
|
||||
monkeypatch.setenv("CACHE_DB", str(tmp_path / "cache.db"))
|
||||
monkeypatch.setenv("HISTORY_DIR", str(tmp_path / ".history"))
|
||||
monkeypatch.setenv("EMBED_DIM", "16")
|
||||
monkeypatch.setenv("MCP_TOKEN", "test-token")
|
||||
return Settings.from_env()
|
||||
@@ -0,0 +1,52 @@
|
||||
import unicodedata
|
||||
|
||||
import pytest
|
||||
from hypothesis import given
|
||||
from hypothesis import strategies as st
|
||||
|
||||
from mdmem.ids import filename_for, is_ulid, new_id, nfc, slugify
|
||||
|
||||
|
||||
def test_new_id_is_ulid():
|
||||
i = new_id()
|
||||
assert is_ulid(i) and len(i) == 26 and i == i.upper()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"title,expected",
|
||||
[
|
||||
("llama-swap routes local models", "llama-swap-routes-local-models"),
|
||||
("Größe der Bücher: ÄÖÜ", "groesse-der-buecher-aeoeue"),
|
||||
(" --- ", "note"),
|
||||
("", "note"),
|
||||
("Bifrost @ 10.0.0.59:9000", "bifrost-10-0-0-59-9000"),
|
||||
("日本語のみ", "note"),
|
||||
],
|
||||
)
|
||||
def test_slugify(title, expected):
|
||||
assert slugify(title) == expected
|
||||
|
||||
|
||||
def test_slug_is_ascii_and_bounded():
|
||||
s = slugify("x" * 500)
|
||||
assert s.isascii() and len(s) <= 60
|
||||
|
||||
|
||||
def test_filename_pairs_slug_and_ulid():
|
||||
i = new_id()
|
||||
fn = filename_for("Ein Test", i)
|
||||
assert fn == f"ein-test--{i}.md"
|
||||
|
||||
|
||||
def test_nfc_normalizes_decomposed_umlauts():
|
||||
decomposed = unicodedata.normalize("NFD", "Grüße")
|
||||
assert decomposed != "Grüße"
|
||||
assert nfc(decomposed) == "Grüße"
|
||||
|
||||
|
||||
@given(st.text())
|
||||
def test_slugify_never_raises_and_is_safe(s):
|
||||
out = slugify(s)
|
||||
assert out.isascii()
|
||||
assert "/" not in out and "\\" not in out and ".." not in out
|
||||
assert out == out.strip("-") or out == "note"
|
||||
@@ -0,0 +1,106 @@
|
||||
import unicodedata
|
||||
|
||||
from hypothesis import given
|
||||
from hypothesis import strategies as st
|
||||
|
||||
from mdmem.notes import Note, dumps, parse
|
||||
|
||||
SAMPLE = """---
|
||||
id: 01J9Z4K8X2M3N5P7Q9R1S3T5V7
|
||||
title: llama-swap routes local models behind Bifrost
|
||||
type: fact
|
||||
summary: Bifrost fronts llama-swap on evo-x2.
|
||||
keywords: [llama-swap, bifrost, evo-x2]
|
||||
tags: [homelab, ai]
|
||||
relations:
|
||||
runs_on: ["[[evo-x2]]"]
|
||||
weird_unknown_key: keep me # and this comment
|
||||
nested_unknown:
|
||||
a: 1
|
||||
b: [2, 3]
|
||||
created: 2026-09-11T08:00:00Z
|
||||
updated: 2026-09-11T08:00:00Z
|
||||
---
|
||||
llama-swap routes models; Bifrost fronts it. See [[local-ai-inference]].
|
||||
"""
|
||||
|
||||
|
||||
def test_parse_splits_frontmatter_and_body():
|
||||
fm, body = parse(SAMPLE)
|
||||
assert fm["title"].startswith("llama-swap")
|
||||
assert body.startswith("llama-swap routes models")
|
||||
assert "---" not in body
|
||||
|
||||
|
||||
def test_unknown_keys_and_order_round_trip():
|
||||
fm, body = parse(SAMPLE)
|
||||
out = dumps(fm, body)
|
||||
assert "weird_unknown_key: keep me" in out
|
||||
assert "# and this comment" in out
|
||||
fm2, body2 = parse(out)
|
||||
assert list(fm2.keys()) == list(fm.keys())
|
||||
assert fm2["nested_unknown"]["b"] == [2, 3]
|
||||
assert body2.strip() == body.strip()
|
||||
|
||||
|
||||
def test_note_accessors():
|
||||
n = Note.from_text(SAMPLE, state="active", path=None)
|
||||
assert n.id == "01J9Z4K8X2M3N5P7Q9R1S3T5V7"
|
||||
assert n.type == "fact"
|
||||
assert n.keywords == ["llama-swap", "bifrost", "evo-x2"]
|
||||
assert n.tags == ["homelab", "ai"]
|
||||
assert n.relations == {"runs_on": ["[[evo-x2]]"]}
|
||||
assert n.wikilinks == ["local-ai-inference"]
|
||||
|
||||
|
||||
def test_note_edit_preserves_unknown_keys():
|
||||
n = Note.from_text(SAMPLE, state="active", path=None)
|
||||
n.set("summary", "new summary")
|
||||
n.body = "changed"
|
||||
out = n.to_text()
|
||||
assert "weird_unknown_key: keep me" in out
|
||||
assert "new summary" in out
|
||||
assert out.rstrip().endswith("changed")
|
||||
|
||||
|
||||
def test_body_is_nfc_normalised_on_parse():
|
||||
text = "---\nid: x\n---\n" + unicodedata.normalize("NFD", "Grüße")
|
||||
_, body = parse(text)
|
||||
assert body.strip() == "Grüße"
|
||||
|
||||
|
||||
def test_new_note_has_canonical_key_order():
|
||||
n = Note.new(title="T", body="B", type="fact", summary="S", keywords=["k"])
|
||||
keys = list(n.fm.keys())
|
||||
assert keys[:5] == ["id", "title", "type", "summary", "keywords"]
|
||||
assert n.to_text().startswith("---\nid: ")
|
||||
|
||||
|
||||
def test_missing_frontmatter_is_tolerated():
|
||||
fm, body = parse("just a body\n")
|
||||
assert dict(fm) == {}
|
||||
assert body.strip() == "just a body"
|
||||
|
||||
|
||||
def test_wikilinks_with_alias_and_heading():
|
||||
n = Note.new(title="T", body="see [[target|alias]] and [[other#Head]]", type="note")
|
||||
assert n.wikilinks == ["target", "other"]
|
||||
|
||||
|
||||
@given(
|
||||
st.dictionaries(
|
||||
st.text(alphabet="abcdefghijklmnop_", min_size=1, max_size=8),
|
||||
st.one_of(st.text(max_size=20), st.integers(), st.booleans(), st.none()),
|
||||
max_size=6,
|
||||
),
|
||||
st.text(max_size=200),
|
||||
)
|
||||
def test_round_trip_property(extra, body):
|
||||
fm, _ = parse(SAMPLE)
|
||||
for k, v in extra.items():
|
||||
fm[k] = v
|
||||
out = dumps(fm, body)
|
||||
fm2, body2 = parse(out)
|
||||
assert set(fm2.keys()) == set(fm.keys())
|
||||
for k in extra:
|
||||
assert fm2[k] == fm[k]
|
||||
@@ -0,0 +1,102 @@
|
||||
import pytest
|
||||
|
||||
from mdmem.notes import Note
|
||||
from mdmem.vaultio import Vault
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def vault(tmp_path):
|
||||
v = Vault(tmp_path / "vault", tmp_path / ".history")
|
||||
v.ensure_dirs()
|
||||
return v
|
||||
|
||||
|
||||
def test_ensure_dirs_creates_lifecycle_folders(vault):
|
||||
for state in ("inbox", "active", "archive"):
|
||||
assert (vault.root / state).is_dir()
|
||||
|
||||
|
||||
def test_write_is_atomic_and_leaves_no_tmp(vault):
|
||||
n = Note.new(title="Ein Test", body="Körper", type="fact")
|
||||
p = vault.write(n)
|
||||
assert p.exists() and p.parent.name == "inbox"
|
||||
assert p.name.endswith(f"--{n.id}.md")
|
||||
assert list(vault.root.rglob("*.tmp")) == []
|
||||
assert "Körper" in p.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_overwrite_snapshots_previous_content_to_history(vault):
|
||||
n = Note.new(title="T", body="v1", type="fact")
|
||||
vault.write(n)
|
||||
assert vault.history(n.id) == [] # nothing existed before the first write
|
||||
n.body = "v2"
|
||||
vault.write(n)
|
||||
hist = vault.history(n.id)
|
||||
assert len(hist) == 1
|
||||
assert "v1" in vault.read_history(n.id, hist[0])
|
||||
|
||||
|
||||
def test_move_between_states_keeps_one_file(vault):
|
||||
n = Note.new(title="T", body="b", type="fact")
|
||||
vault.write(n)
|
||||
new_path = vault.move(n, "active")
|
||||
assert new_path.parent.name == "active"
|
||||
assert n.state == "active"
|
||||
assert len(list(vault.root.rglob("*.md"))) == 1
|
||||
|
||||
|
||||
def test_find_and_read(vault):
|
||||
n = Note.new(title="Grüße", body="hallo", type="fact")
|
||||
vault.write(n)
|
||||
found = vault.find(n.id)
|
||||
assert found is not None
|
||||
got = vault.read(found)
|
||||
assert got.id == n.id and got.state == "inbox" and got.body.strip() == "hallo"
|
||||
assert got.file_hash and got.mtime > 0
|
||||
|
||||
|
||||
def test_rename_on_title_change_removes_old_file(vault):
|
||||
n = Note.new(title="Old title", body="b", type="fact")
|
||||
old = vault.write(n)
|
||||
n.set("title", "New title")
|
||||
new = vault.write(n)
|
||||
assert new != old and not old.exists() and new.exists()
|
||||
assert len(list(vault.root.rglob("*.md"))) == 1
|
||||
|
||||
|
||||
def test_hard_delete_removes_file_and_history(vault):
|
||||
n = Note.new(title="T", body="v1", type="fact")
|
||||
vault.write(n)
|
||||
n.body = "v2"
|
||||
vault.write(n)
|
||||
vault.delete(n)
|
||||
assert vault.find(n.id) is None
|
||||
assert vault.history(n.id) == []
|
||||
|
||||
|
||||
def test_iter_files_skips_tmp_and_history(vault):
|
||||
n = Note.new(title="T", body="b", type="fact")
|
||||
vault.write(n)
|
||||
(vault.root / "inbox" / "junk.md.tmp").write_text("x")
|
||||
(vault.root / "inbox" / "notes.txt").write_text("x")
|
||||
found = list(vault.iter_files())
|
||||
assert len(found) == 1 and found[0][1] == "inbox"
|
||||
|
||||
|
||||
def test_partial_tmp_file_is_never_visible_as_a_note(vault):
|
||||
"""Crash simulation: a tmp file left behind must not be picked up (§14.6)."""
|
||||
(vault.root / "active" / "half-written--01J.md.tmp").write_text("---\nid: x\n---\nboom")
|
||||
assert list(vault.iter_files()) == []
|
||||
|
||||
|
||||
def test_read_normalises_nfd_from_disk(vault):
|
||||
import unicodedata
|
||||
|
||||
p = vault.root / "active" / "x--01JZZZZZZZZZZZZZZZZZZZZZZZ.md"
|
||||
p.write_text(
|
||||
"---\nid: 01JZZZZZZZZZZZZZZZZZZZZZZZ\ntitle: t\n---\n"
|
||||
+ unicodedata.normalize("NFD", "Grüße"),
|
||||
encoding="utf-8",
|
||||
)
|
||||
n = vault.read(p)
|
||||
assert n.body.strip() == "Grüße"
|
||||
Reference in new issue
Block a user