diff --git a/main.py b/main.py index fa2dd4b..a68ef5f 100644 --- a/main.py +++ b/main.py @@ -39,7 +39,7 @@ def run_webdav(host, port, data_dir, username, password): def run_api(host, port, data_dir): import uvicorn - from nexanote.storage import FileNoteStore, run_migration + from nexanote.storage import create_store, run_migration from nexanote.api.routes import create_app data_dir = Path(data_dir) @@ -50,7 +50,9 @@ def run_api(host, port, data_dir): logger.info(report.summary()) if report.backup_path: logger.info(f"Legacy SQLite backup kept at: {report.backup_path}") - db = FileNoteStore(data_dir) + # `create_store` reads the on-disk mode marker / NEXANOTE_STORAGE_MODE + # env var so users can opt into the plain-Markdown backend. + db = create_store(data_dir) app = create_app(db) logger.info(f"API REST démarrée sur http://{host}:{port}") diff --git a/nexanote/storage/__init__.py b/nexanote/storage/__init__.py index 3276bc9..b87897b 100644 --- a/nexanote/storage/__init__.py +++ b/nexanote/storage/__init__.py @@ -2,27 +2,68 @@ NexaNote — Storage package. EN: Public entry points: - - FileNoteStore primary storage as of v1.0.0 (file-based) - - run_migration SQLite → file migration helper - - NexaNoteDB legacy SQLite store, kept for migration + - FileNoteStore YAML-frontmatter backend (default). + - PlainMarkdownNoteStore plain Markdown + JSON-sidecar backend. + - create_store factory that picks the backend from the + on-disk marker / env var. + - run_migration legacy SQLite → file-store migration. + - migrate_yaml_to_plain YAML store → plain Markdown migration. + - NexaNoteDB legacy SQLite store, kept for migration. +FR: Points d'entrée publics du package de stockage. """ -from nexanote.storage.export import export_all, export_note, sanitize_filename +from nexanote.storage.backend import ( + DEFAULT_MODE, + ENV_STORAGE_MODE, + MODE_MARKER, + MODE_PLAIN, + MODE_YAML, + BackendInfo, + NoteStore, + create_store, + detect_mode, + write_mode_marker, +) +from nexanote.storage.export import ( + AutoExportConfig, + AutoExporter, + export_all, + export_note, + sanitize_filename, +) from nexanote.storage.file_store import FileNoteStore from nexanote.storage.legacy_db import NexaNoteDB from nexanote.storage.migration import ( MigrationReport, + PlainMigrationReport, + migrate_yaml_to_plain, needs_migration, run_migration, ) +from nexanote.storage.plain_store import PlainMarkdownNoteStore __all__ = [ + "AutoExportConfig", + "AutoExporter", + "BackendInfo", + "DEFAULT_MODE", + "ENV_STORAGE_MODE", "FileNoteStore", - "NexaNoteDB", + "MODE_MARKER", + "MODE_PLAIN", + "MODE_YAML", "MigrationReport", - "needs_migration", - "run_migration", + "NexaNoteDB", + "NoteStore", + "PlainMarkdownNoteStore", + "PlainMigrationReport", + "create_store", + "detect_mode", "export_all", "export_note", + "migrate_yaml_to_plain", + "needs_migration", + "run_migration", "sanitize_filename", + "write_mode_marker", ] diff --git a/nexanote/storage/backend.py b/nexanote/storage/backend.py new file mode 100644 index 0000000..736e571 --- /dev/null +++ b/nexanote/storage/backend.py @@ -0,0 +1,151 @@ +""" +NexaNote — Storage backend factory / Sélecteur de backend de stockage. + +EN: Two backends ship side-by-side: + * ``FileNoteStore`` (default) — YAML frontmatter inside the + Markdown file itself, one + file per note keyed by id. + * ``PlainMarkdownNoteStore`` — Pure Markdown body in + ``.md`` plus a + ``<Title>.json`` sidecar for + metadata. Direct Obsidian + vault drop-in. + + The active backend for a given data directory is recorded in a small + ``.nexanote_storage_mode`` marker file. ``create_store`` reads the + marker, falls back to the ``NEXANOTE_STORAGE_MODE`` env var, and + finally to ``yaml`` when nothing is set. This keeps existing data + directories on the YAML backend unless the user explicitly switches. + +FR: Deux backends cohabitent — frontmatter YAML ou Markdown propre + + sidecar JSON. Le mode actif est enregistré dans un marqueur, avec + repli sur la variable d'env ``NEXANOTE_STORAGE_MODE`` puis ``yaml``. +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Optional, Union + +from nexanote.storage.file_store import FileNoteStore +from nexanote.storage.plain_store import PlainMarkdownNoteStore + +if TYPE_CHECKING: + from nexanote.storage.export import AutoExportConfig + +logger = logging.getLogger("nexanote.storage.backend") + +# EN: Marker filename inside the data directory. Recording the mode on +# disk avoids accidental backend swaps when the env var is missing. +# FR: Marqueur dans le data dir — évite un changement de backend accidentel. +MODE_MARKER = ".nexanote_storage_mode" + +ENV_STORAGE_MODE = "NEXANOTE_STORAGE_MODE" + +MODE_YAML = "yaml" +MODE_PLAIN = "plain" +DEFAULT_MODE = MODE_YAML + +NoteStore = Union[FileNoteStore, PlainMarkdownNoteStore] + + +@dataclass(frozen=True) +class BackendInfo: + """Resolved backend selection for a data directory.""" + + mode: str + source: str # "marker", "env", or "default" + + +def detect_mode( + data_dir: Path, + env: Optional[dict] = None, +) -> BackendInfo: + """ + EN: Resolve which backend to open for `data_dir`. Marker file wins; + env var is the next choice; otherwise the default (`yaml`) is used. + FR: Détermine le backend à utiliser. Marqueur > env > défaut (`yaml`). + """ + data_dir = Path(data_dir) + marker = data_dir / MODE_MARKER + if marker.exists(): + try: + mode = marker.read_text(encoding="utf-8").strip().lower() + except OSError as exc: + logger.warning(f"could not read storage mode marker {marker}: {exc}") + mode = "" + if mode in (MODE_YAML, MODE_PLAIN): + return BackendInfo(mode=mode, source="marker") + logger.warning( + f"storage mode marker contains unknown value {mode!r} — falling back" + ) + + source_env = env if env is not None else os.environ + env_mode = (source_env.get(ENV_STORAGE_MODE) or "").strip().lower() + if env_mode in (MODE_YAML, MODE_PLAIN): + return BackendInfo(mode=env_mode, source="env") + + return BackendInfo(mode=DEFAULT_MODE, source="default") + + +def write_mode_marker(data_dir: Path, mode: str) -> None: + """Pin `data_dir` to `mode` so future opens don't drift.""" + if mode not in (MODE_YAML, MODE_PLAIN): + raise ValueError(f"unknown storage mode: {mode!r}") + data_dir = Path(data_dir) + data_dir.mkdir(parents=True, exist_ok=True) + marker = data_dir / MODE_MARKER + marker.write_text(mode + "\n", encoding="utf-8") + + +def create_store( + data_dir: Path, + mode: Optional[str] = None, + auto_export: Optional["AutoExportConfig"] = None, +) -> NoteStore: + """ + EN: Open the right backend for `data_dir`. Pass `mode` to force a + choice; otherwise `detect_mode` is used. The chosen mode is + persisted via a marker on first use so subsequent opens stay + consistent even if the env var disappears. + FR: Ouvre le bon backend pour `data_dir`. Persiste le choix dans un + marqueur pour rester stable d'un démarrage à l'autre. + """ + data_dir = Path(data_dir) + if mode is None: + info = detect_mode(data_dir) + mode = info.mode + logger.info( + f"Storage backend: {mode} (source={info.source}, dir={data_dir})" + ) + else: + logger.info(f"Storage backend: {mode} (forced, dir={data_dir})") + + # Record the choice so a later run without env vars stays consistent. + marker = data_dir / MODE_MARKER + if not marker.exists(): + try: + write_mode_marker(data_dir, mode) + except OSError as exc: + logger.warning(f"could not write storage mode marker: {exc}") + + if mode == MODE_PLAIN: + return PlainMarkdownNoteStore(data_dir, auto_export=auto_export) + return FileNoteStore(data_dir, auto_export=auto_export) + + +__all__ = [ + "BackendInfo", + "DEFAULT_MODE", + "ENV_STORAGE_MODE", + "MODE_MARKER", + "MODE_PLAIN", + "MODE_YAML", + "NoteStore", + "create_store", + "detect_mode", + "write_mode_marker", +] diff --git a/nexanote/storage/export.py b/nexanote/storage/export.py index 5421d0d..b9ac7fa 100644 --- a/nexanote/storage/export.py +++ b/nexanote/storage/export.py @@ -7,20 +7,37 @@ NexaNote storage (notes/<id>.md with frontmatter, drawings/<id>.json, notebooks/<id>.yaml) is never touched. + Two entry points: + * `export_all` / `export_note`: one-shot batch export (manual). + * `AutoExporter`: per-note hook driven by the file store so saves, + creates, title changes and sync pulls keep an Obsidian-friendly + mirror in sync without manual intervention. + FR: Écrit chaque note dans un fichier `<titre>.md` contenant uniquement le corps markdown — sans frontmatter ni métadonnées NexaNote — pour rester compatible Obsidian. Le stockage interne n'est jamais modifié. + + Deux points d'entrée : + * `export_all` / `export_note` : export ponctuel (manuel). + * `AutoExporter` : hook par note, déclenché par le store fichier, pour + garder un miroir Obsidian à jour sans intervention manuelle. """ from __future__ import annotations +import json import logging +import os import re +import threading +from dataclasses import dataclass from pathlib import Path -from typing import Optional +from typing import TYPE_CHECKING, Optional from nexanote.models.note import Note -from nexanote.storage.file_store import FileNoteStore, _atomic_write + +if TYPE_CHECKING: + from nexanote.storage.file_store import FileNoteStore logger = logging.getLogger("nexanote.storage.export") @@ -93,7 +110,7 @@ def _note_body(note: Note) -> str: # --------------------------------------------------------------------------- -# Public API +# Public API — manual / batch export # --------------------------------------------------------------------------- def export_note( @@ -113,6 +130,8 @@ def export_note( FR: Écrit `note` en `<target_dir>/<titre nettoyé>.md`. Contenu = corps markdown uniquement, sans frontmatter. Renvoie le chemin écrit. """ + from nexanote.storage.file_store import _atomic_write + target_dir = Path(target_dir) target_dir.mkdir(parents=True, exist_ok=True) if used_names is None: @@ -128,7 +147,7 @@ def export_note( def export_all( - store: FileNoteStore, + store: "FileNoteStore", target_dir: Path, include_archived: bool = False, ) -> list[Path]: @@ -155,7 +174,280 @@ def export_all( return written +# --------------------------------------------------------------------------- +# Automatic export — driven by the file store +# --------------------------------------------------------------------------- + +# EN: Env vars consumed by `AutoExportConfig.from_env`. The feature is opt-in: +# when the flag is missing or set to a falsy value we never write to disk. +# FR: Variables d'environnement lues par `AutoExportConfig.from_env`. +# Fonctionnalité désactivée par défaut. +ENV_AUTO_EXPORT = "NEXANOTE_AUTO_EXPORT_MARKDOWN" +ENV_EXPORT_DIR = "NEXANOTE_MARKDOWN_EXPORT_DIR" + +# EN: Index sidecar that lets the auto-exporter remember which file backs +# each note. Without it we couldn't rename on title change or clean up +# after a soft delete without scanning every file's body. +# FR: Index annexe qui mémorise le fichier exporté pour chaque note. +INDEX_FILE = ".nexanote_export_index.json" +INDEX_VERSION = 1 + +_TRUE_VALUES = {"1", "true", "yes", "on"} + + +def _env_truthy(value: Optional[str]) -> bool: + return bool(value) and value.strip().lower() in _TRUE_VALUES + + +@dataclass +class AutoExportConfig: + """ + EN: Per-store configuration for the automatic Markdown export. + Disabled by default — the user must explicitly opt in via env var or + constructor argument so internal storage stays the only source of + truth for users who don't care about an Obsidian mirror. + FR: Configuration par store pour l'export Markdown automatique. + Désactivé par défaut — opt-in via env ou argument. + """ + + enabled: bool = False + target_dir: Optional[Path] = None + + @classmethod + def from_env( + cls, + default_dir: Path, + env: Optional[dict] = None, + ) -> "AutoExportConfig": + """ + EN: Build a config from `NEXANOTE_AUTO_EXPORT_MARKDOWN` and + `NEXANOTE_MARKDOWN_EXPORT_DIR`. `default_dir` is used when the + export-dir variable is unset. + FR: Construit une config depuis les variables d'environnement. + """ + source = env if env is not None else os.environ + enabled = _env_truthy(source.get(ENV_AUTO_EXPORT)) + dir_value = source.get(ENV_EXPORT_DIR) + target = Path(dir_value).expanduser() if dir_value else Path(default_dir) + return cls(enabled=enabled, target_dir=target) + + +class AutoExporter: + """ + EN: Mirrors managed notes as clean `<title>.md` files in a target + directory. Owns a small JSON index so it can: + * overwrite the same file when a note is saved repeatedly, + * delete the old file and write a new one when the title changes, + * suffix `(N)` on filename collisions across notes, + * clean up after soft delete / archive / hard delete. + + Failures never propagate — auto-export is an extra, not a guarantee. + + FR: Maintient un miroir des notes en `<titre>.md` propres dans un + dossier cible. Utilise un petit index JSON pour gérer renommages, + suppressions et collisions sans casser l'enregistrement principal. + """ + + def __init__(self, config: AutoExportConfig) -> None: + self.config = config + self._lock = threading.RLock() + + @property + def enabled(self) -> bool: + return bool(self.config.enabled and self.config.target_dir is not None) + + @property + def target_dir(self) -> Path: + # Caller must not invoke this when disabled. + assert self.config.target_dir is not None + return self.config.target_dir + + # ------------------------------------------------------------------ + # Index helpers + # ------------------------------------------------------------------ + + def _index_path(self) -> Path: + return self.target_dir / INDEX_FILE + + def _load_index(self) -> dict[str, str]: + path = self._index_path() + if not path.exists(): + return {} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + logger.warning(f"auto-export index unreadable ({path}): {exc}") + return {} + mapping = payload.get("by_note_id") if isinstance(payload, dict) else None + if not isinstance(mapping, dict): + return {} + # Defensive: drop entries that aren't str→str. + return { + str(k): str(v) + for k, v in mapping.items() + if isinstance(k, str) and isinstance(v, str) + } + + def _save_index(self, by_note_id: dict[str, str]) -> None: + from nexanote.storage.file_store import _atomic_write + + path = self._index_path() + try: + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"version": INDEX_VERSION, "by_note_id": by_note_id} + data = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") + _atomic_write(path, data) + except OSError as exc: + logger.warning(f"auto-export index write failed ({path}): {exc}") + + # ------------------------------------------------------------------ + # Public API used by FileNoteStore + # ------------------------------------------------------------------ + + def export(self, note: Note) -> Optional[Path]: + """ + EN: Mirror `note` as a clean `.md` file. Returns the written path, + or None when auto-export is disabled / the note is excluded / + the write fails. + FR: Reflète `note` en `.md` propre. None si désactivé, exclu, ou + si l'écriture échoue. + """ + if not self.enabled: + return None + + # Match `list_notes` defaults: deleted & archived notes are not + # exported. If we previously exported one, clean it up. + if note.is_deleted or note.is_archived: + self.remove(note.id) + return None + + try: + return self._do_export(note) + except Exception as exc: # noqa: BLE001 — never break save_note + logger.warning(f"auto-export failed for note {note.id}: {exc}") + return None + + def remove(self, note_id: str) -> None: + """ + EN: Drop the exported file (if any) and the index entry for `note_id`. + Safe to call when auto-export is disabled or the note was never + exported. + FR: Supprime le fichier exporté et l'entrée d'index pour `note_id`. + """ + if not self.enabled: + return + + with self._lock: + index = self._load_index() + if note_id not in index: + return + old_name = index.pop(note_id) + old_path = self.target_dir / old_name + try: + old_path.unlink() + except FileNotFoundError: + pass + except OSError as exc: + logger.warning(f"auto-export delete failed ({old_path}): {exc}") + self._save_index(index) + + # ------------------------------------------------------------------ + # Internals + # ------------------------------------------------------------------ + + def _do_export(self, note: Note) -> Optional[Path]: + from nexanote.storage.file_store import _atomic_write + + with self._lock: + target_dir = self.target_dir + target_dir.mkdir(parents=True, exist_ok=True) + + index = self._load_index() + previous_name = index.get(note.id) + + base = sanitize_filename(note.title) + desired = f"{base}.md" + + # Names taken by other notes (and untracked preexisting files). + in_use = self._names_in_use(index, exclude_note_id=note.id) + + previous_path = ( + target_dir / previous_name if previous_name else None + ) + + if previous_name == desired and previous_path and previous_path.exists(): + # Title unchanged → keep writing to the same file. + target_name = previous_name + elif previous_name and previous_path and previous_path.exists(): + # Title changed → pick a fresh name; old file deleted below. + target_name = _unique_name(base, in_use) + else: + # First export for this note (or previous file vanished). + target_name = _unique_name(base, in_use) + + target_path = target_dir / target_name + + try: + _atomic_write(target_path, _note_body(note).encode("utf-8")) + except OSError as exc: + logger.warning( + f"auto-export write failed for {note.title!r}: {exc}" + ) + return None + + # If the title moved us to a different filename, drop the old one. + if ( + previous_name + and previous_name != target_name + and previous_path is not None + and previous_path != target_path + ): + try: + previous_path.unlink() + except FileNotFoundError: + pass + except OSError as exc: + logger.warning( + f"auto-export: removing stale {previous_path}: {exc}" + ) + + index[note.id] = target_name + self._save_index(index) + return target_path + + def _names_in_use( + self, + index: dict[str, str], + exclude_note_id: str, + ) -> set[str]: + """Names (lowercased) we must not overwrite when picking a target.""" + target_dir = self.target_dir + in_use: set[str] = set() + + # Other notes whose mirrored file still exists. + for other_id, other_name in index.items(): + if other_id == exclude_note_id: + continue + if (target_dir / other_name).exists(): + in_use.add(other_name.lower()) + + # Files in the directory that aren't tracked at all (user-managed). + tracked_lower = {n.lower() for n in index.values()} + try: + for path in target_dir.glob("*.md"): + if path.name.lower() not in tracked_lower: + in_use.add(path.name.lower()) + except OSError as exc: + logger.debug(f"auto-export: scanning {target_dir} failed: {exc}") + + return in_use + + __all__ = [ + "AutoExportConfig", + "AutoExporter", + "ENV_AUTO_EXPORT", + "ENV_EXPORT_DIR", "export_all", "export_note", "sanitize_filename", diff --git a/nexanote/storage/file_store.py b/nexanote/storage/file_store.py index c7bdc9e..42dfa2d 100644 --- a/nexanote/storage/file_store.py +++ b/nexanote/storage/file_store.py @@ -34,7 +34,7 @@ import threading from datetime import datetime, timezone from pathlib import Path -from typing import Optional +from typing import TYPE_CHECKING, Optional import yaml @@ -48,6 +48,11 @@ SyncStatus, ) +if TYPE_CHECKING: + # EN: `export` imports this module, so only resolve the type at check time + # to keep runtime imports cycle-free. + from nexanote.storage.export import AutoExportConfig # noqa: F401 + logger = logging.getLogger("nexanote.storage.file") @@ -464,7 +469,11 @@ class FileNoteStore: SCHEMA_VERSION = 2 # Bumped when the on-disk layout changes. - def __init__(self, data_dir: Path) -> None: + def __init__( + self, + data_dir: Path, + auto_export: Optional["AutoExportConfig"] = None, + ) -> None: self.data_dir = Path(data_dir) self.notes_dir = self.data_dir / NOTES_DIR self.drawings_dir = self.data_dir / DRAWINGS_DIR @@ -472,6 +481,14 @@ def __init__(self, data_dir: Path) -> None: for d in (self.notes_dir, self.drawings_dir, self.notebooks_dir): d.mkdir(parents=True, exist_ok=True) + # EN: Lazy import — `export` depends on this module for `_atomic_write`, + # so doing it at module top would create a circular import. + # FR: Import retardé — évite l'import circulaire avec `export`. + from nexanote.storage.export import AutoExportConfig, AutoExporter + if auto_export is None: + auto_export = AutoExportConfig.from_env(self.data_dir / "export") + self.auto_exporter = AutoExporter(auto_export) + # ------------------------------------------------------------------ # Compatibility shims for code paths that used to read db.db_path # ------------------------------------------------------------------ @@ -592,6 +609,14 @@ def save_note(self, note: Note, save_pages: bool = True) -> None: json.dumps(drawings, ensure_ascii=False, indent=2).encode("utf-8"), ) + # EN: Mirror the saved note as clean Markdown if auto-export is on. + # Runs outside the per-file lock so a slow target FS can't block + # concurrent writes to other notes; the exporter uses its own lock + # to keep the index consistent. + # FR: Réplique la note sauvée en Markdown propre si l'export auto est + # activé. Hors du lock fichier pour ne pas bloquer d'autres écritures. + self.auto_exporter.export(note) + def get_note(self, note_id: str, load_pages: bool = True) -> Optional[Note]: return self._read_note(note_id, load_pages=load_pages) @@ -669,6 +694,9 @@ def delete_note_permanent(self, note_id: str) -> None: except FileNotFoundError: pass + # Mirror the purge into the auto-exported directory if enabled. + self.auto_exporter.remove(note_id) + # ------------------------------------------------------------------ # Pages CRUD (kept for API parity — a page write rewrites the note) # ------------------------------------------------------------------ diff --git a/nexanote/storage/migration.py b/nexanote/storage/migration.py index ff0802b..e2dbb74 100644 --- a/nexanote/storage/migration.py +++ b/nexanote/storage/migration.py @@ -170,10 +170,170 @@ def _marker_payload(**fields) -> str: return json.dumps(fields, indent=2) +# --------------------------------------------------------------------------- +# YAML → plain Markdown migration +# --------------------------------------------------------------------------- + +PLAIN_MIGRATION_MARKER = ".nexanote_plain_migrated" +PLAIN_BACKUP_DIR = "_yaml_backup" + + +@dataclass +class PlainMigrationReport: + """Summary of a YAML → plain-Markdown migration run.""" + + ran: bool = False + notebooks: int = 0 + notes: int = 0 + pages: int = 0 + strokes: int = 0 + errors: list[str] = None # type: ignore[assignment] + backup_dir: Optional[Path] = None + skipped_reason: Optional[str] = None + + def __post_init__(self) -> None: + if self.errors is None: + self.errors = [] + + def summary(self) -> str: + if not self.ran: + return f"Plain-MD migration skipped: {self.skipped_reason or 'unknown'}" + return ( + f"Plain-MD migration done — {self.notebooks} notebooks, " + f"{self.notes} notes, {self.pages} pages, {self.strokes} strokes" + + (f" ({len(self.errors)} errors)" if self.errors else "") + ) + + +def migrate_yaml_to_plain( + data_dir: Path, + backup: bool = True, +) -> PlainMigrationReport: + """ + EN: Convert a YAML-frontmatter store at `data_dir` into the plain + Markdown + JSON-sidecar layout in the same directory. The original + YAML notes are moved to ``<data_dir>/notes/_yaml_backup/`` so the + operation is reversible. Idempotent — a marker prevents double runs. + + After migration, ``backend.create_store`` returns a + ``PlainMarkdownNoteStore`` for this directory. + + FR: Convertit un store YAML en stockage Markdown brut + sidecar JSON + dans le même dossier. Les fichiers YAML originaux sont sauvegardés + dans ``notes/_yaml_backup/``. Idempotent. + """ + from nexanote.storage.backend import ( + MODE_PLAIN, + write_mode_marker, + ) + from nexanote.storage.file_store import FileNoteStore + from nexanote.storage.plain_store import PlainMarkdownNoteStore + + data_dir = Path(data_dir) + data_dir.mkdir(parents=True, exist_ok=True) + marker = data_dir / PLAIN_MIGRATION_MARKER + + report = PlainMigrationReport() + + if marker.exists(): + report.skipped_reason = "already migrated to plain-MD (marker present)" + return report + + yaml_store = FileNoteStore(data_dir) + notes_dir = yaml_store.notes_dir + + # Snapshot the YAML-format files BEFORE swapping the backend, so the + # plain store doesn't see them as foreign sidecars. + yaml_md_files = sorted(notes_dir.glob("*.md")) + if not yaml_md_files: + # Nothing to migrate, but still pin the mode + marker so the user's + # next start opens the plain backend. + write_mode_marker(data_dir, MODE_PLAIN) + marker.write_text(_marker_payload(reason="empty store"), encoding="utf-8") + report.skipped_reason = "no YAML notes to convert" + return report + + backup_dir: Optional[Path] = None + if backup: + backup_dir = notes_dir / PLAIN_BACKUP_DIR + backup_dir.mkdir(parents=True, exist_ok=True) + + # Read every note from the YAML store (with pages + strokes) before + # we start writing — avoids reading half-converted files. + notes: list = [] + for meta in yaml_store.list_notes(include_deleted=True, include_archived=True): + full = yaml_store.get_note(meta.id, load_pages=True) + if full is not None: + notes.append((full, yaml_store._note_path(full.id))) + + notebooks = list(yaml_store.list_notebooks(include_archived=True)) + yaml_store.close() + + plain_store = PlainMarkdownNoteStore(data_dir) + + for nb in notebooks: + try: + plain_store.save_notebook(nb) + report.notebooks += 1 + except Exception as exc: # pragma: no cover — defensive + msg = f"notebook {nb.id[:8]}: {exc}" + logger.error(f"plain migration failed: {msg}") + report.errors.append(msg) + + for note, original_path in notes: + try: + # Move the original YAML-format .md aside so the plain writer + # never sees its frontmatter as a stray plain MD file. + if backup_dir is not None and original_path.exists(): + target = backup_dir / original_path.name + try: + original_path.replace(target) + except OSError as exc: + logger.warning( + f"could not back up {original_path} → {target}: {exc}" + ) + else: + try: + original_path.unlink() + except FileNotFoundError: + pass + + plain_store.save_note(note) + report.notes += 1 + report.pages += len(note.pages) + for page in note.pages: + report.strokes += len(page.strokes) + except Exception as exc: # pragma: no cover — defensive + msg = f"note {note.id[:8]}: {exc}" + logger.error(f"plain migration failed: {msg}") + report.errors.append(msg) + + write_mode_marker(data_dir, MODE_PLAIN) + marker.write_text( + _marker_payload( + reason="migrated YAML → plain", + notebooks=report.notebooks, + notes=report.notes, + pages=report.pages, + strokes=report.strokes, + ), + encoding="utf-8", + ) + + report.ran = True + report.backup_dir = backup_dir + logger.info(report.summary()) + return report + + __all__ = [ "MigrationReport", + "PlainMigrationReport", "needs_migration", "run_migration", + "migrate_yaml_to_plain", "LEGACY_DB_NAME", "MIGRATION_MARKER", + "PLAIN_BACKUP_DIR", + "PLAIN_MIGRATION_MARKER", ] diff --git a/nexanote/storage/plain_store.py b/nexanote/storage/plain_store.py new file mode 100644 index 0000000..59a1332 --- /dev/null +++ b/nexanote/storage/plain_store.py @@ -0,0 +1,627 @@ +""" +NexaNote — Plain Markdown storage backend / Backend Markdown brut. + +EN: Alternative to ``FileNoteStore`` (the YAML-frontmatter backend) for users + who want their notes to live as clean ``.md`` files directly editable in + Obsidian and other plain-Markdown tools. The on-disk layout is: + + <data_dir>/ + notebooks/<notebook_id>.yaml # Same as YAML mode (notebooks + # don't show up in the user's + # Markdown vault). + notes/ + <Sanitized Title>.md # Pure Markdown body — no + # frontmatter, no NexaNote tags. + <Sanitized Title>.json # Sidecar metadata: id, tags, + # dates, notebook_id, page meta. + drawings/<note_id>.json # Same as YAML mode (keyed by + # the stable note id). + + Stable ids come from the sidecar JSON. ``.md`` files dropped in by the + user with no sidecar (Obsidian users adding a file by hand) get a + deterministic id derived from the filename so the API can address them + until the user saves an edit through NexaNote (which writes a sidecar + with a real UUID). + + The public method surface mirrors ``FileNoteStore`` so REST routes, + the WebDAV provider and the sync engine all work unchanged. + +FR: Backend alternatif à ``FileNoteStore`` pour stocker les notes comme + fichiers ``.md`` propres + sidecars ``.json``. Compatible API avec + ``FileNoteStore`` (REST, WebDAV, sync). +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from typing import TYPE_CHECKING, Optional + +from nexanote.models.note import ( + InkStroke, + Note, + Notebook, + NoteType, + Page, + SyncStatus, +) +from nexanote.storage.export import sanitize_filename +from nexanote.storage.file_store import ( + DRAWINGS_DIR, + NOTEBOOKS_DIR, + NOTES_DIR, + _LOCKS, + _atomic_write, + _dict_to_stroke, + _fmt_dt, + _join_pages_body, + _merge_metadata, + _now, + _parse_dt, + _safe_id, + _split_pages_body, + _stroke_to_dict, + deserialize_notebook, + plain_md_id_from_stem, + serialize_notebook, + stem_from_plain_md_id, + synthesize_plain_md_note, +) + +if TYPE_CHECKING: + from nexanote.storage.export import AutoExportConfig + +logger = logging.getLogger("nexanote.storage.plain") + +SIDECAR_SUFFIX = ".json" +PLAIN_SCHEMA_VERSION = 1 + + +class PlainMarkdownNoteStore: + """ + EN: Plain-Markdown backend with the same public surface as + ``FileNoteStore``. Drop-in replacement for code that already + consumes the YAML store via duck-typed calls. + FR: Backend Markdown brut, même API publique que ``FileNoteStore``. + """ + + SCHEMA_VERSION = 3 # Distinct from FileNoteStore's layout version. + + def __init__( + self, + data_dir: Path, + auto_export: Optional["AutoExportConfig"] = None, + ) -> None: + self.data_dir = Path(data_dir) + self.notes_dir = self.data_dir / NOTES_DIR + self.drawings_dir = self.data_dir / DRAWINGS_DIR + self.notebooks_dir = self.data_dir / NOTEBOOKS_DIR + for d in (self.notes_dir, self.drawings_dir, self.notebooks_dir): + d.mkdir(parents=True, exist_ok=True) + + # Lazy import — `export` reaches back into file_store helpers. + from nexanote.storage.export import AutoExportConfig, AutoExporter + if auto_export is None: + auto_export = AutoExportConfig.from_env(self.data_dir / "export") + self.auto_exporter = AutoExporter(auto_export) + + # ------------------------------------------------------------------ + # Compat shims expected by callers written against FileNoteStore. + # ------------------------------------------------------------------ + + @property + def db_path(self) -> Path: + return self.data_dir / "nexanote.db" + + def close(self) -> None: + return None + + # ------------------------------------------------------------------ + # Path helpers + # ------------------------------------------------------------------ + + def _md_path(self, stem: str) -> Path: + return self.notes_dir / f"{stem}.md" + + def _sidecar_path(self, stem: str) -> Path: + return self.notes_dir / f"{stem}{SIDECAR_SUFFIX}" + + def _drawing_path(self, note_id: str) -> Path: + return self.drawings_dir / f"{_safe_id(note_id)}.json" + + def _notebook_path(self, notebook_id: str) -> Path: + return self.notebooks_dir / f"{_safe_id(notebook_id)}.yaml" + + # ------------------------------------------------------------------ + # ID ↔ stem resolution + # ------------------------------------------------------------------ + + def _stem_for_id(self, note_id: str) -> Optional[str]: + """ + EN: Find the filename stem (`<Sanitized Title>`) for `note_id`. + Reads sidecars to find the match. Falls back to the derived id + for plain ``.md`` files dropped in by the user. + FR: Trouve le stem du fichier pour `note_id`. Lit les sidecars, + avec repli sur l'id dérivé pour les .md sans sidecar. + """ + # Plain MD imports use the deterministic prefix scheme — recover the + # stem directly when a file with that name exists. + derived_stem = stem_from_plain_md_id(note_id) + if derived_stem is not None: + md = self._md_path(derived_stem) + if md.exists() and not self._sidecar_path(derived_stem).exists(): + return derived_stem + + for sidecar in self.notes_dir.glob(f"*{SIDECAR_SUFFIX}"): + data = self._safe_read_sidecar(sidecar) + if data and data.get("id") == note_id: + return sidecar.stem + return None + + def _safe_read_sidecar(self, path: Path) -> Optional[dict]: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + logger.warning(f"unreadable sidecar {path}: {exc}") + return None + return data if isinstance(data, dict) else None + + def _pick_stem(self, base: str, owner_id: str) -> str: + """ + EN: Pick a free filename stem for a note titled `base`. Reuses the + slot already owned by `owner_id` if one exists; otherwise + suffixes ` (N)` until a free slot is found. + FR: Choisit un stem libre. Réutilise le slot déjà détenu par + `owner_id` si possible, sinon suffixe ` (N)`. + """ + candidate = base + n = 1 + while True: + md = self._md_path(candidate) + sidecar = self._sidecar_path(candidate) + + if not md.exists() and not sidecar.exists(): + return candidate + + if sidecar.exists(): + data = self._safe_read_sidecar(sidecar) + if data and data.get("id") == owner_id: + return candidate + elif md.exists(): + # Plain MD without sidecar — owned by the derived id. + if plain_md_id_from_stem(candidate) == owner_id: + return candidate + + n += 1 + candidate = f"{base} ({n})" + + # ------------------------------------------------------------------ + # Notebooks (delegated to YAML — they are not user-visible files) + # ------------------------------------------------------------------ + + def save_notebook(self, nb: Notebook) -> None: + path = self._notebook_path(nb.id) + with _LOCKS.get(path): + _atomic_write(path, serialize_notebook(nb).encode("utf-8")) + + def get_notebook(self, notebook_id: str) -> Optional[Notebook]: + path = self._notebook_path(notebook_id) + if not path.exists(): + return None + with _LOCKS.get(path): + try: + return deserialize_notebook(path.read_text(encoding="utf-8")) + except OSError as exc: + logger.error(f"read notebook {path} failed: {exc}") + return None + + def list_notebooks(self, include_archived: bool = False) -> list[Notebook]: + out: list[Notebook] = [] + for path in sorted(self.notebooks_dir.glob("*.yaml")): + try: + nb = deserialize_notebook(path.read_text(encoding="utf-8")) + except OSError as exc: + logger.warning(f"skip unreadable notebook {path}: {exc}") + continue + if nb is None: + continue + if not include_archived and nb.is_archived: + continue + out.append(nb) + out.sort(key=lambda nb: nb.name.lower()) + return out + + def delete_notebook(self, notebook_id: str) -> None: + path = self._notebook_path(notebook_id) + with _LOCKS.get(path): + try: + path.unlink() + except FileNotFoundError: + pass + + # ------------------------------------------------------------------ + # Notes + # ------------------------------------------------------------------ + + def save_note(self, note: Note, save_pages: bool = True) -> None: + """ + EN: Persist a Note as `<Title>.md` + `<Title>.json`. Renames the + existing files when the title changes; reuses them otherwise. + With `save_pages=False`, the existing `.md` body and drawings + are kept and only the sidecar metadata is refreshed. + FR: Persiste une note en `<titre>.md` + `<titre>.json`. Renomme + si le titre change, conserve le corps si `save_pages=False`. + """ + base = sanitize_filename(note.title) + current_stem = self._stem_for_id(note.id) + target_stem = self._pick_stem(base, owner_id=note.id) + + target_md = self._md_path(target_stem) + target_sidecar = self._sidecar_path(target_stem) + drawing_path = self._drawing_path(note.id) + + with _LOCKS.get(target_md): + if not save_pages and current_stem: + # Reload the existing pages/body so we don't drop them when + # only metadata is being updated (mirrors FileNoteStore). + existing = self._read_note_from_stem(current_stem, load_pages=True) + if existing is not None and existing.pages: + note = _merge_metadata(note, existing) + + pages_text = {p.page_number: p.typed_content for p in note.pages} + body = _join_pages_body(pages_text) + sidecar = self._build_sidecar(note) + + # Write the new pair atomically before deleting the old one so a + # crash mid-rename can't leave the note unreadable. + _atomic_write(target_md, body.encode("utf-8")) + _atomic_write( + target_sidecar, + json.dumps(sidecar, ensure_ascii=False, indent=2).encode("utf-8"), + ) + + # Remove the previous pair if the title moved us to a new stem. + if current_stem and current_stem != target_stem: + for p in ( + self._md_path(current_stem), + self._sidecar_path(current_stem), + ): + try: + p.unlink() + except FileNotFoundError: + pass + + # Drawings (per-note id, independent of the title-based stem). + drawings = self._build_drawings(note) + if drawings is None: + try: + drawing_path.unlink() + except FileNotFoundError: + pass + else: + _atomic_write( + drawing_path, + json.dumps(drawings, ensure_ascii=False, indent=2).encode("utf-8"), + ) + + # Keep the auto-export mirror in sync (no-op when disabled). + self.auto_exporter.export(note) + + def get_note(self, note_id: str, load_pages: bool = True) -> Optional[Note]: + stem = self._stem_for_id(note_id) + if stem is None: + return None + return self._read_note_from_stem(stem, load_pages=load_pages) + + def _read_note_from_stem( + self, stem: str, load_pages: bool + ) -> Optional[Note]: + md_path = self._md_path(stem) + if not md_path.exists(): + return None + + with _LOCKS.get(md_path): + try: + md_text = md_path.read_text(encoding="utf-8") + except OSError as exc: + logger.error(f"read md {md_path} failed: {exc}") + return None + + sidecar = None + sidecar_path = self._sidecar_path(stem) + if sidecar_path.exists(): + sidecar = self._safe_read_sidecar(sidecar_path) + + if sidecar is not None: + note = self._note_from_sidecar(sidecar, md_text) + else: + note = synthesize_plain_md_note(md_path, md_text) + + if not load_pages: + note.pages = [] + return note + + # Drawings sidecar (separate file, keyed by note id) + drawing_path = self._drawing_path(note.id) + if drawing_path.exists(): + try: + drawings = json.loads(drawing_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + logger.warning(f"unreadable drawings {drawing_path}: {exc}") + drawings = None + if drawings: + self._attach_drawings(note, drawings) + + return note + + def list_notes( + self, + notebook_id: Optional[str] = None, + include_deleted: bool = False, + include_archived: bool = False, + search_title: Optional[str] = None, + ) -> list[Note]: + out: list[Note] = [] + needle = search_title.lower() if search_title else None + for md_path in self.notes_dir.glob("*.md"): + sidecar_path = self._sidecar_path(md_path.stem) + if sidecar_path.exists(): + data = self._safe_read_sidecar(sidecar_path) + if data is None: + continue + # Read the body lazily — only when required to build pages. + note = self._note_from_sidecar(data, md_text="") + else: + try: + text = md_path.read_text(encoding="utf-8") + except OSError as exc: + logger.warning(f"skip unreadable note {md_path}: {exc}") + continue + note = synthesize_plain_md_note(md_path, text) + note.pages = [] # listings stay metadata-only + + if not include_deleted and note.is_deleted: + continue + if not include_archived and note.is_archived: + continue + if notebook_id is not None and note.notebook_id != notebook_id: + continue + if needle and needle not in note.title.lower(): + continue + out.append(note) + + out.sort(key=lambda n: n.updated_at, reverse=True) + return out + + def delete_note_permanent(self, note_id: str) -> None: + stem = self._stem_for_id(note_id) + if stem is not None: + md = self._md_path(stem) + sidecar = self._sidecar_path(stem) + with _LOCKS.get(md): + for p in (md, sidecar): + try: + p.unlink() + except FileNotFoundError: + pass + + drawing = self._drawing_path(note_id) + try: + drawing.unlink() + except FileNotFoundError: + pass + + self.auto_exporter.remove(note_id) + + # ------------------------------------------------------------------ + # Pages CRUD (a page write rewrites the note — same as FileNoteStore) + # ------------------------------------------------------------------ + + def save_page(self, page: Page) -> None: + if not page.note_id: + raise ValueError("Page is missing note_id") + note = self.get_note(page.note_id, load_pages=True) + if note is None: + raise KeyError(f"unknown note {page.note_id}") + replaced = False + for i, existing in enumerate(note.pages): + if existing.page_number == page.page_number: + note.pages[i] = page + replaced = True + break + if not replaced: + note.pages.append(page) + note.pages.sort(key=lambda p: p.page_number) + self.save_note(note, save_pages=True) + + def list_pages(self, note_id: str) -> list[Page]: + note = self.get_note(note_id, load_pages=True) + return list(note.pages) if note else [] + + # ------------------------------------------------------------------ + # Strokes CRUD (scan-based, like FileNoteStore — production callers + # route through save_page instead). + # ------------------------------------------------------------------ + + def save_stroke(self, stroke: InkStroke, page_id: str) -> None: + for md_path in self.notes_dir.glob("*.md"): + note = self._read_note_from_stem(md_path.stem, load_pages=True) + if note is None: + continue + for page in note.pages: + if page.id == page_id: + page.strokes = [s for s in page.strokes if s.id != stroke.id] + page.strokes.append(stroke) + self.save_note(note, save_pages=True) + return + raise KeyError(f"unknown page {page_id}") + + def list_strokes(self, page_id: str) -> list[InkStroke]: + for md_path in self.notes_dir.glob("*.md"): + note = self._read_note_from_stem(md_path.stem, load_pages=True) + if note is None: + continue + for page in note.pages: + if page.id == page_id: + return list(page.strokes) + return [] + + def delete_stroke(self, stroke_id: str) -> None: + for md_path in self.notes_dir.glob("*.md"): + note = self._read_note_from_stem(md_path.stem, load_pages=True) + if note is None: + continue + changed = False + for page in note.pages: + before = len(page.strokes) + page.strokes = [s for s in page.strokes if s.id != stroke_id] + if len(page.strokes) != before: + changed = True + if changed: + self.save_note(note, save_pages=True) + return + + # ------------------------------------------------------------------ + # Stats + # ------------------------------------------------------------------ + + def get_stats(self) -> dict: + notebooks = pages = strokes = 0 + notes = notes_deleted = 0 + + for _ in self.list_notebooks(include_archived=False): + notebooks += 1 + + for md_path in self.notes_dir.glob("*.md"): + note = self._read_note_from_stem(md_path.stem, load_pages=True) + if note is None: + continue + if note.is_deleted: + notes_deleted += 1 + continue + if note.is_archived: + continue + notes += 1 + pages += len(note.pages) + for page in note.pages: + strokes += len(page.strokes) + + return { + "notebooks": notebooks, + "notes": notes, + "notes_deleted": notes_deleted, + "pages": pages, + "strokes": strokes, + } + + # ------------------------------------------------------------------ + # Sidecar serialization + # ------------------------------------------------------------------ + + def _build_sidecar(self, note: Note) -> dict: + return { + "schema": PLAIN_SCHEMA_VERSION, + "id": note.id, + "title": note.title, + "note_type": note.note_type.value, + "notebook_id": note.notebook_id, + "tags": list(note.tags), + "is_pinned": note.is_pinned, + "is_archived": note.is_archived, + "is_deleted": note.is_deleted, + "sync_status": note.sync_status.value, + "created_at": _fmt_dt(note.created_at), + "updated_at": _fmt_dt(note.updated_at), + "pages": [ + { + "page_number": page.page_number, + "page_id": page.id, + "template": page.template, + "width_px": float(page.width_px), + "height_px": float(page.height_px), + "updated_at": _fmt_dt(page.updated_at), + } + for page in note.pages + ], + } + + def _note_from_sidecar(self, sidecar: dict, md_text: str) -> Note: + pages_text = _split_pages_body(md_text) if md_text else {} + pages: list[Page] = [] + for pm in sidecar.get("pages") or []: + num = int(pm.get("page_number", 1)) + page = Page( + id=pm.get("page_id") or f"{sidecar['id']}::p{num}", + note_id=sidecar["id"], + page_number=num, + template=pm.get("template", "blank"), + width_px=float(pm.get("width_px", 1404.0)), + height_px=float(pm.get("height_px", 1872.0)), + typed_content=pages_text.get(num, ""), + created_at=_parse_dt( + pm.get("created_at", sidecar.get("created_at", _fmt_dt(_now()))) + ), + updated_at=_parse_dt( + pm.get("updated_at", sidecar.get("updated_at", _fmt_dt(_now()))) + ), + ) + pages.append(page) + pages.sort(key=lambda p: p.page_number) + + return Note( + id=sidecar["id"], + notebook_id=sidecar.get("notebook_id"), + title=sidecar.get("title", "Sans titre"), + note_type=NoteType(sidecar.get("note_type", "typed")), + tags=list(sidecar.get("tags") or []), + is_pinned=bool(sidecar.get("is_pinned", False)), + is_archived=bool(sidecar.get("is_archived", False)), + is_deleted=bool(sidecar.get("is_deleted", False)), + sync_status=SyncStatus( + sidecar.get("sync_status", SyncStatus.LOCAL_ONLY.value) + ), + created_at=_parse_dt(sidecar.get("created_at", _fmt_dt(_now()))), + updated_at=_parse_dt(sidecar.get("updated_at", _fmt_dt(_now()))), + pages=pages, + ) + + # ------------------------------------------------------------------ + # Drawings serialization + # ------------------------------------------------------------------ + + def _build_drawings(self, note: Note) -> Optional[dict]: + if not any(p.strokes for p in note.pages): + return None + return { + "schema": 1, + "note_id": note.id, + "updated_at": _fmt_dt(note.updated_at), + "pages": [ + { + "page_number": p.page_number, + "page_id": p.id, + "updated_at": _fmt_dt(p.updated_at), + "strokes": [_stroke_to_dict(s) for s in p.strokes], + } + for p in note.pages + if p.strokes + ], + } + + def _attach_drawings(self, note: Note, drawings: dict) -> None: + by_num = { + int(p.get("page_number", 1)): [ + _dict_to_stroke(s) for s in p.get("strokes") or [] + ] + for p in drawings.get("pages") or [] + } + for page in note.pages: + page.strokes = by_num.get(page.page_number, []) + + +__all__ = [ + "PlainMarkdownNoteStore", + "PLAIN_SCHEMA_VERSION", + "SIDECAR_SUFFIX", +] diff --git a/nexanote/sync/server.py b/nexanote/sync/server.py index 25bb00a..66404a1 100644 --- a/nexanote/sync/server.py +++ b/nexanote/sync/server.py @@ -22,7 +22,7 @@ from wsgidav.wsgidav_app import WsgiDAVApp from nexanote.models.note import Note, Notebook, NoteType, SyncStatus -from nexanote.storage import FileNoteStore, run_migration +from nexanote.storage import FileNoteStore, create_store, run_migration from nexanote.sync.webdav_provider import NexaNoteDAVProvider # EN: Slug for the fallback notebook used to host notes that aren't assigned @@ -203,7 +203,7 @@ def run_server( migration_report = run_migration(data_dir) if migration_report.ran: logger.info(migration_report.summary()) - db = FileNoteStore(data_dir) + db = create_store(data_dir) # Ensure required WebDAV directories exist + the fallback notebook is # present, so the very first sync push has valid parent collections. diff --git a/tests/test_auto_export.py b/tests/test_auto_export.py new file mode 100644 index 0000000..0311e1d --- /dev/null +++ b/tests/test_auto_export.py @@ -0,0 +1,582 @@ +""" +NexaNote — Tests for the automatic clean Markdown export. + +EN: Auto-export keeps an Obsidian-friendly mirror of the user's notes + next to the internal storage. These tests cover the four trigger + points called out in the spec: + - note create + - note update / save + - note title change + - sync pull (via the engine's call to ``save_note``) + plus the safety guarantees: feature-off by default, archived/deleted + notes are not exported, duplicate titles are suffixed, and the + internal storage is never touched. + +FR: Tests pour l'export Markdown automatique. Vérifient les déclencheurs + requis (création, sauvegarde, changement de titre, sync pull) ainsi + que les garanties (désactivé par défaut, suppressions et collisions + gérées proprement, stockage interne intact). +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from fastapi.testclient import TestClient + +from nexanote.api.routes import create_app +from nexanote.models.note import ( + InkStroke, + Note, + NoteType, + Page, + Point, + SyncStatus, +) +from nexanote.storage import ( + AutoExportConfig, + AutoExporter, + FileNoteStore, +) +from nexanote.storage.export import ( + ENV_AUTO_EXPORT, + ENV_EXPORT_DIR, + INDEX_FILE, +) + + +# --------------------------------------------------------------------------- +# Fixtures / helpers +# --------------------------------------------------------------------------- + + +@pytest.fixture +def export_dir(tmp_path) -> Path: + return tmp_path / "obsidian-mirror" + + +@pytest.fixture +def store(tmp_path, export_dir) -> FileNoteStore: + """Store with auto-export enabled by an explicit config.""" + config = AutoExportConfig(enabled=True, target_dir=export_dir) + s = FileNoteStore(tmp_path / "store", auto_export=config) + yield s + s.close() + + +@pytest.fixture +def disabled_store(tmp_path) -> FileNoteStore: + """Store with no auto-export configured (default behavior).""" + s = FileNoteStore(tmp_path / "no_export_store") + yield s + s.close() + + +def _typed_note(title: str, body: str = "body") -> Note: + note = Note(title=title, note_type=NoteType.TYPED) + note.add_page().typed_content = body + return note + + +def _read_index(target: Path) -> dict[str, str]: + path = target / INDEX_FILE + if not path.exists(): + return {} + return json.loads(path.read_text(encoding="utf-8"))["by_note_id"] + + +# --------------------------------------------------------------------------- +# Config: env var parsing +# --------------------------------------------------------------------------- + + +class TestConfigFromEnv: + def test_disabled_by_default(self): + config = AutoExportConfig.from_env(default_dir=Path("/tmp/x"), env={}) + assert config.enabled is False + + def test_enabled_when_env_truthy(self): + for value in ("true", "1", "yes", "on", "TRUE", " True "): + config = AutoExportConfig.from_env( + default_dir=Path("/tmp/x"), + env={ENV_AUTO_EXPORT: value}, + ) + assert config.enabled, f"value={value!r} should enable auto-export" + + def test_disabled_when_env_falsy(self): + for value in ("false", "0", "no", "off", "", "garbage"): + config = AutoExportConfig.from_env( + default_dir=Path("/tmp/x"), + env={ENV_AUTO_EXPORT: value}, + ) + assert not config.enabled, f"value={value!r} should leave auto-export off" + + def test_target_dir_from_env(self, tmp_path): + config = AutoExportConfig.from_env( + default_dir=tmp_path / "default", + env={ + ENV_AUTO_EXPORT: "true", + ENV_EXPORT_DIR: str(tmp_path / "custom"), + }, + ) + assert config.target_dir == tmp_path / "custom" + + def test_default_dir_used_when_var_unset(self, tmp_path): + config = AutoExportConfig.from_env( + default_dir=tmp_path / "fallback", + env={ENV_AUTO_EXPORT: "true"}, + ) + assert config.target_dir == tmp_path / "fallback" + + +class TestStoreReadsEnv: + """The store reads env vars when no explicit config is given.""" + + def test_store_disabled_by_default(self, tmp_path, monkeypatch): + monkeypatch.delenv(ENV_AUTO_EXPORT, raising=False) + monkeypatch.delenv(ENV_EXPORT_DIR, raising=False) + store = FileNoteStore(tmp_path / "store") + try: + assert store.auto_exporter.enabled is False + store.save_note(_typed_note("Anything", "body")) + # Default fallback dir is `<data_dir>/export` — but with the + # exporter disabled, nothing should be created there. + export_dir = tmp_path / "store" / "export" + assert not export_dir.exists() or not any(export_dir.glob("*.md")) + finally: + store.close() + + def test_store_enabled_via_env(self, tmp_path, monkeypatch): + target = tmp_path / "env-mirror" + monkeypatch.setenv(ENV_AUTO_EXPORT, "true") + monkeypatch.setenv(ENV_EXPORT_DIR, str(target)) + + store = FileNoteStore(tmp_path / "store") + try: + assert store.auto_exporter.enabled + store.save_note(_typed_note("Hello env", "from env")) + assert (target / "Hello env.md").exists() + finally: + store.close() + + +# --------------------------------------------------------------------------- +# Triggers required by the spec +# --------------------------------------------------------------------------- + + +class TestAutoExportOnCreate: + def test_save_note_creates_clean_md_file(self, store, export_dir): + note = _typed_note("First Note", "# Hello\n\nbody text") + store.save_note(note) + + path = export_dir / "First Note.md" + assert path.exists(), "auto-export should create the mirror file on save" + text = path.read_text(encoding="utf-8") + assert not text.startswith("---"), "exported file must have no frontmatter" + assert "# Hello" in text + assert "body text" in text + + def test_index_records_filename(self, store, export_dir): + note = _typed_note("Indexed", "x") + store.save_note(note) + index = _read_index(export_dir) + assert index.get(note.id) == "Indexed.md" + + def test_internal_storage_untouched(self, store, export_dir): + note = _typed_note("Untouched", "body") + store.save_note(note) + + # Internal file keeps the YAML frontmatter. + internal = store._note_path(note.id) + text = internal.read_text(encoding="utf-8") + assert text.startswith("---\n") + + # Exported file has no frontmatter. + exported = (export_dir / "Untouched.md").read_text(encoding="utf-8") + assert not exported.startswith("---") + + +class TestAutoExportOnUpdate: + def test_save_overwrites_same_file(self, store, export_dir): + note = _typed_note("Doc", "first version") + store.save_note(note) + + # Edit body and save again. + note.pages[0].typed_content = "second version" + store.save_note(note) + + path = export_dir / "Doc.md" + assert path.exists() + text = path.read_text(encoding="utf-8") + assert "second version" in text + assert "first version" not in text + + # Only one file in the mirror — no `Doc (2).md` from re-save. + md_files = sorted(p.name for p in export_dir.glob("*.md")) + assert md_files == ["Doc.md"] + + def test_metadata_only_save_still_exports_body(self, store, export_dir): + """`save_note(save_pages=False)` should reuse the existing pages + and the export must still reflect the saved body.""" + note = _typed_note("Meta", "body content") + store.save_note(note) + + # Reload metadata only and re-save (mirrors the API's update_note path). + meta_only = store.get_note(note.id, load_pages=False) + meta_only.tags = ["updated"] + store.save_note(meta_only, save_pages=False) + + text = (export_dir / "Meta.md").read_text(encoding="utf-8") + assert "body content" in text + + def test_save_page_propagates_to_export(self, store, export_dir): + """Editing a page (the route used by the editor) must keep the + Obsidian mirror current.""" + note = _typed_note("Live", "initial") + store.save_note(note) + + page = store.get_note(note.id, load_pages=True).pages[0] + page.typed_content = "edited via save_page" + store.save_page(page) + + text = (export_dir / "Live.md").read_text(encoding="utf-8") + assert "edited via save_page" in text + + +class TestAutoExportOnTitleChange: + def test_renaming_moves_the_export_file(self, store, export_dir): + note = _typed_note("Old Title", "body") + store.save_note(note) + assert (export_dir / "Old Title.md").exists() + + note.title = "New Title" + store.save_note(note) + + assert (export_dir / "New Title.md").exists(), ( + "after a title change the new file must appear" + ) + assert not (export_dir / "Old Title.md").exists(), ( + "the previous export must be removed to avoid stale duplicates" + ) + + # Index is updated. + assert _read_index(export_dir)[note.id] == "New Title.md" + + def test_invalid_chars_in_renamed_title_sanitized(self, store, export_dir): + note = _typed_note("Clean", "body") + store.save_note(note) + + note.title = "weird/title:?<>" + store.save_note(note) + + names = sorted(p.name for p in export_dir.glob("*.md")) + assert all(ch not in name for name in names for ch in '/:?<>"\\|*') + assert any(name.endswith(".md") for name in names) + + +class TestAutoExportOnSyncPull: + """ + EN: The sync engine's pull path calls ``save_note(remote_note)``. We + exercise the full ``engine.sync()`` with a stubbed WebDAV client so + the test is hermetic but still proves the pull → save → export + wiring end-to-end. + """ + + def _patch_client(self, engine, remote_meta: dict) -> None: + """Stub every network method the engine uses so no socket is opened.""" + engine.client.ping = lambda: True + engine.client.list_notebooks = lambda: [ + {"name": "carnet__01234567", "is_collection": True} + ] + engine.client.list_notes = lambda nb_slug: [ + {"name": "remote-note__89abcdef", "is_collection": True} + ] + engine.client.get_note_meta = lambda nb, note: remote_meta + engine.client.get_ink_page = lambda nb, note, page_num: None + + def test_pull_writes_clean_md(self, store, export_dir): + from nexanote.sync.client import NexaNoteSyncEngine, SyncConfig + + engine = NexaNoteSyncEngine( + store, + SyncConfig(server_url="http://test.invalid/", username="u", password="p"), + ) + + remote_meta = { + "id": "89abcdef-1111-2222-3333-444455556666", + "title": "Pulled From Server", + "type": "typed", + "tags": ["sync"], + "is_pinned": False, + "created_at": "2026-01-01T00:00:00+00:00", + "updated_at": "2026-01-01T00:00:00+00:00", + "pages": [ + {"page_number": 1, "template": "blank", "typed_content": "from sync pull"}, + ], + } + self._patch_client(engine, remote_meta) + + # Skip push — we only care about pull behavior. + engine._push = lambda report: None + + report = engine.sync() + assert report.success(), f"sync failed: {report.errors}" + assert report.notes_pulled == 1 + + path = export_dir / "Pulled From Server.md" + assert path.exists(), "auto-export must run on the pulled note" + assert "from sync pull" in path.read_text(encoding="utf-8") + + def test_pull_update_renames_export(self, store, export_dir): + """A subsequent pull with a newer timestamp + new title must move the + mirror file the same way a local rename does.""" + from nexanote.sync.client import NexaNoteSyncEngine, SyncConfig + + engine = NexaNoteSyncEngine( + store, + SyncConfig(server_url="http://test.invalid/", username="u", password="p"), + ) + + first = { + "id": "89abcdef-1111-2222-3333-444455556666", + "title": "Original", + "type": "typed", + "tags": [], + "is_pinned": False, + "created_at": "2026-01-01T00:00:00+00:00", + "updated_at": "2026-01-01T00:00:00+00:00", + "pages": [ + {"page_number": 1, "template": "blank", "typed_content": "v1"}, + ], + } + self._patch_client(engine, first) + engine._push = lambda report: None + engine.sync() + assert (export_dir / "Original.md").exists() + + # Mark local as SYNCED so the engine treats the remote-newer branch + # as a plain update (not a conflict). + local = store.get_note(first["id"], load_pages=True) + local.sync_status = SyncStatus.SYNCED + store.save_note(local) + + # Server now serves a newer version with a different title. + second = dict(first) + second["title"] = "Renamed Server-side" + second["updated_at"] = "2026-02-01T00:00:00+00:00" + second["pages"] = [ + {"page_number": 1, "template": "blank", "typed_content": "v2"}, + ] + self._patch_client(engine, second) + engine.sync() + + assert (export_dir / "Renamed Server-side.md").exists() + assert not (export_dir / "Original.md").exists(), ( + "stale mirror from before the rename must be cleaned up" + ) + + +# --------------------------------------------------------------------------- +# Skip rules + duplicate handling +# --------------------------------------------------------------------------- + + +class TestSkipRules: + def test_soft_deleted_note_not_exported(self, store, export_dir): + note = _typed_note("Trash", "x") + note.soft_delete() + store.save_note(note, save_pages=False) + assert not list(export_dir.glob("*.md")), ( + "soft-deleted notes must not appear in the Obsidian mirror" + ) + + def test_soft_delete_after_export_removes_file(self, store, export_dir): + note = _typed_note("ToDelete", "body") + store.save_note(note) + assert (export_dir / "ToDelete.md").exists() + + note.soft_delete() + store.save_note(note, save_pages=False) + + assert not (export_dir / "ToDelete.md").exists() + assert _read_index(export_dir).get(note.id) is None + + def test_archived_note_not_exported(self, store, export_dir): + note = _typed_note("Archived", "x") + note.is_archived = True + store.save_note(note, save_pages=False) + assert not list(export_dir.glob("*.md")) + + def test_restore_re_creates_export(self, store, export_dir): + note = _typed_note("Comes Back", "body") + store.save_note(note) + note.soft_delete() + store.save_note(note, save_pages=False) + assert not (export_dir / "Comes Back.md").exists() + + note.restore() + store.save_note(note, save_pages=False) + assert (export_dir / "Comes Back.md").exists() + + def test_hard_delete_removes_export(self, store, export_dir): + note = _typed_note("Goodbye", "body") + store.save_note(note) + assert (export_dir / "Goodbye.md").exists() + + store.delete_note_permanent(note.id) + assert not (export_dir / "Goodbye.md").exists() + + +class TestDuplicateTitles: + def test_two_notes_same_title_get_distinct_files(self, store, export_dir): + a = _typed_note("Same", "body A") + b = _typed_note("Same", "body B") + store.save_note(a) + store.save_note(b) + + names = sorted(p.name for p in export_dir.glob("*.md")) + assert names == ["Same (2).md", "Same.md"] + + index = _read_index(export_dir) + assert sorted(index.values()) == ["Same (2).md", "Same.md"] + assert index[a.id] != index[b.id] + + def test_repeated_save_does_not_grow_suffix(self, store, export_dir): + note = _typed_note("Stable", "body") + for i in range(5): + note.pages[0].typed_content = f"version {i}" + store.save_note(note) + + names = sorted(p.name for p in export_dir.glob("*.md")) + assert names == ["Stable.md"], ( + "a note saved repeatedly must keep occupying the same filename" + ) + + def test_does_not_overwrite_user_file(self, store, export_dir): + export_dir.mkdir(parents=True, exist_ok=True) + kept = export_dir / "Notes.md" + kept.write_text("user-managed content\n", encoding="utf-8") + + note = _typed_note("Notes", "from NexaNote") + store.save_note(note) + + assert kept.read_text(encoding="utf-8") == "user-managed content\n" + assert (export_dir / "Notes (2).md").exists() + assert "from NexaNote" in (export_dir / "Notes (2).md").read_text("utf-8") + + +# --------------------------------------------------------------------------- +# Disabled exporter is a complete no-op +# --------------------------------------------------------------------------- + + +class TestDisabled: + def test_disabled_store_writes_nothing(self, tmp_path, disabled_store): + disabled_store.save_note(_typed_note("Anything", "body")) + # No file in the default `<data_dir>/export` location. + export_dir = tmp_path / "no_export_store" / "export" + if export_dir.exists(): + assert not list(export_dir.glob("*.md")) + + def test_disabled_remove_is_safe(self, disabled_store): + # Should not raise even with no target dir / no index. + disabled_store.auto_exporter.remove("missing-id") + + +# --------------------------------------------------------------------------- +# WebDAV sync compat — ensure auto-export does not break the existing path +# --------------------------------------------------------------------------- + + +class TestWebDAVCompat: + """ + EN: The WebDAV provider also writes notes via ``save_note``. Auto-export + must layer on top without changing the on-disk note format that the + provider serves to clients. + """ + + def test_internal_format_unchanged_with_auto_export(self, tmp_path): + # First, capture the on-disk format produced with auto-export OFF. + plain_store = FileNoteStore(tmp_path / "plain") + try: + note = _typed_note("Compat", "body") + plain_store.save_note(note) + plain_bytes = plain_store._note_path(note.id).read_bytes() + finally: + plain_store.close() + + # Same note, but with auto-export ON — internal bytes must match. + config = AutoExportConfig(enabled=True, target_dir=tmp_path / "mirror") + store = FileNoteStore(tmp_path / "with_export", auto_export=config) + try: + note2 = _typed_note("Compat", "body") + note2.id = note.id # keep the same id so timestamps match + note2.created_at = note.created_at + note2.updated_at = note.updated_at + for orig_page, new_page in zip(note.pages, note2.pages): + new_page.id = orig_page.id + new_page.created_at = orig_page.created_at + new_page.updated_at = orig_page.updated_at + store.save_note(note2) + with_export_bytes = store._note_path(note2.id).read_bytes() + finally: + store.close() + + assert plain_bytes == with_export_bytes, ( + "auto-export must not alter the internal Markdown representation" + ) + + +# --------------------------------------------------------------------------- +# API integration +# --------------------------------------------------------------------------- + + +class TestApiIntegration: + @pytest.fixture + def api_client(self, tmp_path): + target = tmp_path / "api-mirror" + config = AutoExportConfig(enabled=True, target_dir=target) + db = FileNoteStore(tmp_path / "api_store", auto_export=config) + app = create_app(db) + with TestClient(app) as c: + yield c, db, target + db.close() + + def test_create_via_api_exports(self, api_client): + c, db, target = api_client + resp = c.post("/notes", json={"title": "API note", "note_type": "typed"}) + assert resp.status_code == 201 + assert (target / "API note.md").exists() + + def test_update_title_via_api_renames_export(self, api_client): + c, db, target = api_client + created = c.post("/notes", json={"title": "Old", "note_type": "typed"}).json() + c.put(f"/notes/{created['id']}", json={"title": "Brand New"}) + + assert (target / "Brand New.md").exists() + assert not (target / "Old.md").exists() + + def test_text_update_via_api_refreshes_export(self, api_client): + c, db, target = api_client + created = c.post("/notes", json={"title": "Live edit", "note_type": "typed"}).json() + c.put( + f"/notes/{created['id']}/pages/1/text", + json={"typed_content": "fresh body via API"}, + ) + text = (target / "Live edit.md").read_text(encoding="utf-8") + assert "fresh body via API" in text + + def test_delete_via_api_removes_export(self, api_client): + c, db, target = api_client + created = c.post("/notes", json={"title": "Bye", "note_type": "typed"}).json() + assert (target / "Bye.md").exists() + + c.delete(f"/notes/{created['id']}") + assert not (target / "Bye.md").exists() diff --git a/tests/test_plain_store.py b/tests/test_plain_store.py new file mode 100644 index 0000000..172993f --- /dev/null +++ b/tests/test_plain_store.py @@ -0,0 +1,774 @@ +""" +NexaNote — Tests for the plain-Markdown storage backend. + +EN: Covers the new ``PlainMarkdownNoteStore`` and the surrounding wiring + (mode marker, factory, YAML→plain migration). Required scenarios: + - note creation (file pair appears) + - rename (files renamed, id stable) + - sync (engine.save_note path keeps the layout consistent + and conflict resolution works through the same + public API as the YAML backend) + - conflict handling (resolver writes both winner + conflict copy) + +FR: Tests du nouveau backend Markdown brut + sidecar JSON et de l'aiguillage + associé (marqueur, factory, migration YAML→plain). Couvre la création, + le renommage, la sync et la gestion des conflits. +""" + +from __future__ import annotations + +import json +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +from fastapi.testclient import TestClient + +from nexanote.api.routes import create_app +from nexanote.models.note import ( + InkStroke, + Note, + Notebook, + NoteType, + Page, + Point, + SyncStatus, +) +from nexanote.storage import ( + FileNoteStore, + MODE_MARKER, + MODE_PLAIN, + MODE_YAML, + PlainMarkdownNoteStore, + create_store, + detect_mode, + migrate_yaml_to_plain, +) +from nexanote.storage.backend import ENV_STORAGE_MODE +from nexanote.storage.plain_store import SIDECAR_SUFFIX +from nexanote.sync.client import ( + NexaNoteSyncEngine, + SyncConfig, + SyncReport, +) +from nexanote.sync.conflict import ConflictResolver, ConflictStrategy + + +# --------------------------------------------------------------------------- +# Helpers / fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def store(tmp_path) -> PlainMarkdownNoteStore: + s = PlainMarkdownNoteStore(tmp_path / "plain") + yield s + s.close() + + +def _typed_note(title: str, body: str = "body") -> Note: + note = Note(title=title, note_type=NoteType.TYPED) + note.add_page().typed_content = body + return note + + +def _read_sidecar(store: PlainMarkdownNoteStore, stem: str) -> dict: + return json.loads( + (store.notes_dir / f"{stem}{SIDECAR_SUFFIX}").read_text(encoding="utf-8") + ) + + +# --------------------------------------------------------------------------- +# Note creation +# --------------------------------------------------------------------------- + + +class TestCreate: + def test_save_writes_md_and_sidecar(self, store): + note = _typed_note("Hello World", "# Hi\n\nbody text") + store.save_note(note) + + md = store.notes_dir / "Hello World.md" + sidecar = store.notes_dir / "Hello World.json" + assert md.exists() and sidecar.exists() + + body = md.read_text(encoding="utf-8") + assert not body.startswith("---"), "plain backend must not emit frontmatter" + assert "# Hi" in body + assert "body text" in body + + meta = json.loads(sidecar.read_text("utf-8")) + assert meta["id"] == note.id + assert meta["title"] == "Hello World" + assert meta["note_type"] == "typed" + + def test_get_note_round_trip(self, store): + note = _typed_note("Round trip", "content here") + store.save_note(note) + + loaded = store.get_note(note.id, load_pages=True) + assert loaded is not None + assert loaded.id == note.id + assert loaded.title == "Round trip" + assert loaded.pages[0].typed_content == "content here" + + def test_filename_sanitized(self, store): + store.save_note(_typed_note("bad/title:?<>", "x")) + names = [p.name for p in store.notes_dir.glob("*.md")] + assert all(ch not in n for n in names for ch in '/:?<>"\\|*') + + def test_unicode_title_preserved(self, store): + store.save_note(_typed_note("Café ☕", "espresso")) + assert (store.notes_dir / "Café ☕.md").exists() + assert (store.notes_dir / "Café ☕.json").exists() + + def test_blank_title_uses_fallback(self, store): + note = _typed_note(" ", "still has body") + store.save_note(note) + assert (store.notes_dir / "untitled.md").exists() + + def test_two_notes_same_title_get_distinct_files(self, store): + a = _typed_note("Same", "body A") + b = _typed_note("Same", "body B") + store.save_note(a) + store.save_note(b) + + names = sorted(p.name for p in store.notes_dir.glob("*.md")) + assert names == ["Same (2).md", "Same.md"] + + # Each note loads back with its own body. + loaded_a = store.get_note(a.id, load_pages=True) + loaded_b = store.get_note(b.id, load_pages=True) + assert loaded_a.pages[0].typed_content == "body A" + assert loaded_b.pages[0].typed_content == "body B" + + def test_internal_md_has_no_frontmatter(self, store): + store.save_note(_typed_note("Pure", "plain body")) + text = (store.notes_dir / "Pure.md").read_text(encoding="utf-8") + assert text == "plain body\n" + + def test_save_creates_drawings_when_strokes_present(self, store): + note = Note(title="Inked", note_type=NoteType.HANDWRITTEN) + page = note.add_page() + page.add_stroke( + InkStroke( + color="#000000", + points=[Point(0, 0), Point(10, 10)], + ) + ) + store.save_note(note) + assert (store.drawings_dir / f"{note.id}.json").exists() + + loaded = store.get_note(note.id, load_pages=True) + assert loaded.pages[0].strokes + assert loaded.pages[0].strokes[0].color == "#000000" + + +# --------------------------------------------------------------------------- +# Rename +# --------------------------------------------------------------------------- + + +class TestRename: + def test_title_change_renames_files(self, store): + note = _typed_note("Old name", "body") + store.save_note(note) + old_md = store.notes_dir / "Old name.md" + old_sidecar = store.notes_dir / "Old name.json" + assert old_md.exists() and old_sidecar.exists() + + note.title = "Brand new" + store.save_note(note) + + assert (store.notes_dir / "Brand new.md").exists() + assert (store.notes_dir / "Brand new.json").exists() + assert not old_md.exists() + assert not old_sidecar.exists() + + def test_id_stable_across_rename(self, store): + note = _typed_note("Original", "body") + store.save_note(note) + first_id = note.id + + note.title = "Rebadged" + store.save_note(note) + + loaded = store.get_note(first_id, load_pages=True) + assert loaded is not None + assert loaded.id == first_id + assert loaded.title == "Rebadged" + + def test_metadata_only_save_keeps_body(self, store): + note = _typed_note("Edit Me", "important content") + store.save_note(note) + + # Reload metadata only and re-save — pages should not be wiped. + meta_only = store.get_note(note.id, load_pages=False) + meta_only.tags = ["updated"] + store.save_note(meta_only, save_pages=False) + + again = store.get_note(note.id, load_pages=True) + assert again.pages[0].typed_content == "important content" + assert "updated" in again.tags + + def test_rename_into_collision_gets_suffix(self, store): + a = _typed_note("Apple", "A body") + b = _typed_note("Banana", "B body") + store.save_note(a) + store.save_note(b) + + # Rename Banana → Apple: must not stomp on Apple's file. + b.title = "Apple" + store.save_note(b) + + assert (store.notes_dir / "Apple.md").exists() + assert (store.notes_dir / "Apple (2).md").exists() + # The original Banana files are gone. + assert not (store.notes_dir / "Banana.md").exists() + assert not (store.notes_dir / "Banana.json").exists() + + # Each note still resolves by id and keeps its own body. + assert store.get_note(a.id).title == "Apple" + assert store.get_note(b.id).title == "Apple" + bodies = { + store.get_note(a.id, load_pages=True).pages[0].typed_content, + store.get_note(b.id, load_pages=True).pages[0].typed_content, + } + assert bodies == {"A body", "B body"} + + def test_rename_back_reuses_existing_slot(self, store): + note = _typed_note("Notes", "body") + store.save_note(note) + note.title = "Other" + store.save_note(note) + note.title = "Notes" + store.save_note(note) + + # No suffixed duplicate from churning the title. + names = sorted(p.name for p in store.notes_dir.glob("*.md")) + assert names == ["Notes.md"] + + +# --------------------------------------------------------------------------- +# List / soft delete / archive +# --------------------------------------------------------------------------- + + +class TestList: + def test_list_returns_metadata_only(self, store): + store.save_note(_typed_note("First", "body 1")) + store.save_note(_typed_note("Second", "body 2")) + listed = store.list_notes() + assert {n.title for n in listed} == {"First", "Second"} + assert all(n.pages == [] for n in listed) + + def test_list_skips_soft_deleted_by_default(self, store): + keep = _typed_note("Keep", "x") + gone = _typed_note("Gone", "y") + store.save_note(keep) + store.save_note(gone) + gone.soft_delete() + store.save_note(gone, save_pages=False) + + titles = [n.title for n in store.list_notes()] + assert titles == ["Keep"] + + with_deleted = [n.title for n in store.list_notes(include_deleted=True)] + assert sorted(with_deleted) == ["Gone", "Keep"] + + def test_list_skips_archived_by_default(self, store): + active = _typed_note("Active", "x") + archived = _typed_note("Archived", "y") + store.save_note(active) + store.save_note(archived) + archived.is_archived = True + store.save_note(archived, save_pages=False) + + titles = [n.title for n in store.list_notes()] + assert titles == ["Active"] + + def test_filter_by_notebook(self, store): + nb = Notebook(name="Carnet") + store.save_notebook(nb) + in_nb = _typed_note("Inside", "x") + in_nb.notebook_id = nb.id + outside = _typed_note("Outside", "y") + store.save_note(in_nb) + store.save_note(outside) + + filtered = store.list_notes(notebook_id=nb.id) + assert [n.title for n in filtered] == ["Inside"] + + def test_search_by_title(self, store): + store.save_note(_typed_note("Apple", "x")) + store.save_note(_typed_note("Banana", "y")) + results = store.list_notes(search_title="app") + assert [n.title for n in results] == ["Apple"] + + +class TestDelete: + def test_delete_permanent_removes_pair(self, store): + note = _typed_note("Bye", "body") + store.save_note(note) + assert (store.notes_dir / "Bye.md").exists() + + store.delete_note_permanent(note.id) + assert not (store.notes_dir / "Bye.md").exists() + assert not (store.notes_dir / "Bye.json").exists() + assert store.get_note(note.id) is None + + +# --------------------------------------------------------------------------- +# Plain MD without sidecar (Obsidian drop-in) +# --------------------------------------------------------------------------- + + +class TestExternalMdImport: + def test_md_without_sidecar_is_listed(self, store): + (store.notes_dir / "External.md").write_text( + "# Dropped in\n\nbody\n", encoding="utf-8" + ) + listed = store.list_notes() + assert any(n.title == "External" for n in listed) + + def test_external_md_id_resolves(self, store): + (store.notes_dir / "Dropped.md").write_text("hi", encoding="utf-8") + listed = store.list_notes() + external = next(n for n in listed if n.title == "Dropped") + full = store.get_note(external.id, load_pages=True) + assert full is not None + assert "hi" in full.pages[0].typed_content + + +# --------------------------------------------------------------------------- +# Notebook CRUD parity +# --------------------------------------------------------------------------- + + +class TestNotebooks: + def test_save_and_list(self, store): + nb = Notebook(name="Cours", color="#3b82f6") + store.save_notebook(nb) + assert [n.id for n in store.list_notebooks()] == [nb.id] + + def test_delete_notebook(self, store): + nb = Notebook(name="Tmp") + store.save_notebook(nb) + store.delete_notebook(nb.id) + assert store.list_notebooks() == [] + + +# --------------------------------------------------------------------------- +# Backend factory + mode marker +# --------------------------------------------------------------------------- + + +class TestBackendFactory: + def test_default_mode_is_yaml(self, tmp_path, monkeypatch): + monkeypatch.delenv(ENV_STORAGE_MODE, raising=False) + info = detect_mode(tmp_path / "fresh") + assert info.mode == MODE_YAML + assert info.source == "default" + + def test_env_var_picks_plain(self, tmp_path, monkeypatch): + monkeypatch.setenv(ENV_STORAGE_MODE, "plain") + info = detect_mode(tmp_path / "via_env") + assert info.mode == MODE_PLAIN + assert info.source == "env" + + def test_marker_overrides_env(self, tmp_path, monkeypatch): + monkeypatch.setenv(ENV_STORAGE_MODE, "plain") + data_dir = tmp_path / "with_marker" + data_dir.mkdir() + (data_dir / MODE_MARKER).write_text("yaml\n", encoding="utf-8") + info = detect_mode(data_dir) + assert info.mode == MODE_YAML + assert info.source == "marker" + + def test_create_store_yaml(self, tmp_path, monkeypatch): + monkeypatch.delenv(ENV_STORAGE_MODE, raising=False) + store = create_store(tmp_path / "y") + try: + assert isinstance(store, FileNoteStore) + finally: + store.close() + + def test_create_store_plain(self, tmp_path): + store = create_store(tmp_path / "p", mode=MODE_PLAIN) + try: + assert isinstance(store, PlainMarkdownNoteStore) + # Forced mode is also recorded so subsequent opens are stable. + assert (tmp_path / "p" / MODE_MARKER).exists() + finally: + store.close() + + def test_marker_pinned_after_first_open(self, tmp_path, monkeypatch): + # Open once with env=plain, the marker pins it. + monkeypatch.setenv(ENV_STORAGE_MODE, "plain") + s1 = create_store(tmp_path / "stick") + s1.close() + + # Now drop the env var — second open must still see plain. + monkeypatch.delenv(ENV_STORAGE_MODE, raising=False) + s2 = create_store(tmp_path / "stick") + try: + assert isinstance(s2, PlainMarkdownNoteStore) + finally: + s2.close() + + +# --------------------------------------------------------------------------- +# YAML → plain migration +# --------------------------------------------------------------------------- + + +class TestMigration: + def _seed_yaml(self, data_dir: Path) -> dict: + store = FileNoteStore(data_dir) + try: + nb = Notebook(name="Carnet", color="#3b82f6") + store.save_notebook(nb) + note = Note(notebook_id=nb.id, title="Migrated", note_type=NoteType.TYPED) + note.add_page(template="lined").typed_content = "yaml body" + store.save_note(note) + return {"nb": nb, "note": note} + finally: + store.close() + + def test_migrate_converts_notes(self, tmp_path): + data_dir = tmp_path / "migration" + seeded = self._seed_yaml(data_dir) + + report = migrate_yaml_to_plain(data_dir) + assert report.ran + assert report.notes == 1 + assert report.notebooks == 1 + + # Plain backend reads the migrated note correctly. + plain = PlainMarkdownNoteStore(data_dir) + try: + loaded = plain.get_note(seeded["note"].id, load_pages=True) + assert loaded is not None + assert loaded.title == "Migrated" + assert loaded.pages[0].typed_content == "yaml body" + finally: + plain.close() + + def test_migration_creates_plain_files(self, tmp_path): + data_dir = tmp_path / "creates_files" + self._seed_yaml(data_dir) + + migrate_yaml_to_plain(data_dir) + + assert (data_dir / "notes" / "Migrated.md").exists() + assert (data_dir / "notes" / "Migrated.json").exists() + # And the old YAML-format file is preserved as a backup. + backup = data_dir / "notes" / "_yaml_backup" + assert backup.exists() + assert any(backup.glob("*.md")) + + def test_migration_pins_mode_to_plain(self, tmp_path): + data_dir = tmp_path / "mode_pinned" + self._seed_yaml(data_dir) + + migrate_yaml_to_plain(data_dir) + info = detect_mode(data_dir) + assert info.mode == MODE_PLAIN + + def test_migration_idempotent(self, tmp_path): + data_dir = tmp_path / "idem" + self._seed_yaml(data_dir) + first = migrate_yaml_to_plain(data_dir) + second = migrate_yaml_to_plain(data_dir) + assert first.ran + assert not second.ran + assert "marker present" in (second.skipped_reason or "") + + def test_migration_on_empty_store(self, tmp_path): + data_dir = tmp_path / "empty" + # Fresh dir: no .md files. Migration still pins the mode. + report = migrate_yaml_to_plain(data_dir) + assert not report.ran # nothing to convert + assert detect_mode(data_dir).mode == MODE_PLAIN + + def test_migration_preserves_notebook(self, tmp_path): + data_dir = tmp_path / "with_nb" + seeded = self._seed_yaml(data_dir) + migrate_yaml_to_plain(data_dir) + + plain = PlainMarkdownNoteStore(data_dir) + try: + nbs = plain.list_notebooks() + assert any(nb.id == seeded["nb"].id for nb in nbs) + finally: + plain.close() + + +# --------------------------------------------------------------------------- +# Sync compatibility +# --------------------------------------------------------------------------- + + +def _stub_remote_note(note_id: str, title: str, body: str, updated_iso: str) -> dict: + return { + "id": note_id, + "title": title, + "type": "typed", + "tags": [], + "is_pinned": False, + "created_at": "2026-01-01T00:00:00+00:00", + "updated_at": updated_iso, + "pages": [ + {"page_number": 1, "template": "blank", "typed_content": body}, + ], + } + + +def _patch_engine_for_pull(engine, remote_meta: dict) -> None: + engine.client.ping = lambda: True + engine.client.list_notebooks = lambda: [ + {"name": "carnet__01234567", "is_collection": True} + ] + engine.client.list_notes = lambda nb_slug: [ + {"name": "remote-note__89abcdef", "is_collection": True} + ] + engine.client.get_note_meta = lambda nb, note: remote_meta + engine.client.get_ink_page = lambda nb, note, page_num: None + + +class TestSync: + def test_pull_writes_plain_files(self, tmp_path): + store = PlainMarkdownNoteStore(tmp_path / "pull") + engine = NexaNoteSyncEngine( + store, + SyncConfig(server_url="http://test/", username="u", password="p"), + ) + + remote = _stub_remote_note( + "89abcdef-1111-2222-3333-444455556666", + "From Server", + "synced body", + "2026-02-01T00:00:00+00:00", + ) + _patch_engine_for_pull(engine, remote) + engine._push = lambda report: None + + report = engine.sync() + assert report.success(), report.errors + assert report.notes_pulled == 1 + + # Plain layout files were written. + assert (store.notes_dir / "From Server.md").exists() + assert (store.notes_dir / "From Server.json").exists() + body = (store.notes_dir / "From Server.md").read_text("utf-8") + assert "synced body" in body + assert not body.startswith("---") + + store.close() + + def test_pull_rename_updates_filenames(self, tmp_path): + store = PlainMarkdownNoteStore(tmp_path / "pull_rename") + engine = NexaNoteSyncEngine( + store, + SyncConfig(server_url="http://test/", username="u", password="p"), + ) + + first = _stub_remote_note( + "89abcdef-1111-2222-3333-444455556666", + "Original", + "v1", + "2026-01-01T00:00:00+00:00", + ) + _patch_engine_for_pull(engine, first) + engine._push = lambda report: None + engine.sync() + assert (store.notes_dir / "Original.md").exists() + + # Mark synced so the conflict path is skipped. + local = store.get_note(first["id"], load_pages=True) + local.sync_status = SyncStatus.SYNCED + store.save_note(local) + + renamed = dict(first) + renamed["title"] = "Renamed" + renamed["updated_at"] = "2026-03-01T00:00:00+00:00" + renamed["pages"] = [ + {"page_number": 1, "template": "blank", "typed_content": "v2"}, + ] + _patch_engine_for_pull(engine, renamed) + engine.sync() + + assert (store.notes_dir / "Renamed.md").exists() + assert not (store.notes_dir / "Original.md").exists() + store.close() + + def test_push_marks_notes_synced(self, tmp_path): + """Push uses save_note(save_pages=False) — must reuse existing body.""" + store = PlainMarkdownNoteStore(tmp_path / "push") + nb = Notebook(name="Carnet") + store.save_notebook(nb) + note = Note(notebook_id=nb.id, title="Pushable", note_type=NoteType.TYPED) + note.add_page().typed_content = "to push" + store.save_note(note) + + engine = NexaNoteSyncEngine( + store, + SyncConfig(server_url="http://test/", username="u", password="p"), + ) + # Stub all network so the push success path completes. + engine.client.ping = lambda: True + engine.client.list_notebooks = lambda: [] + engine.client.list_notes = lambda nb_slug: [] + engine.client.create_notebook_dir = lambda nb_slug: True + engine.client.create_note_dir = lambda nb_slug, note_slug: True + engine.client.put_note_meta = lambda *a, **k: (True, None) + engine.client.put_ink_page = lambda *a, **k: (True, None) + # Skip pull so we exercise only push. + engine._pull = lambda report: None + + report = engine.sync() + assert report.success(), report.errors + assert report.notes_pushed == 1 + + # The note is marked SYNCED and the body is preserved across the + # save_pages=False call the engine performs. + loaded = store.get_note(note.id, load_pages=True) + assert loaded.sync_status == SyncStatus.SYNCED + assert loaded.pages[0].typed_content == "to push" + store.close() + + +# --------------------------------------------------------------------------- +# Conflict handling +# --------------------------------------------------------------------------- + + +class TestConflict: + def test_keep_both_writes_winner_and_conflict_copy(self, tmp_path): + store = PlainMarkdownNoteStore(tmp_path / "conflict") + + local = Note(title="Doc", note_type=NoteType.TYPED) + local.id = "shared-id" + local.add_page().typed_content = "local body" + local.updated_at = datetime.now(timezone.utc) + timedelta(seconds=10) + store.save_note(local) + + remote = Note(title="Doc", note_type=NoteType.TYPED) + remote.id = "shared-id" + remote.add_page().typed_content = "remote body" + remote.updated_at = datetime.now(timezone.utc) + + resolver = ConflictResolver(strategy=ConflictStrategy.KEEP_BOTH) + result = resolver.resolve(local, remote) + assert result.had_conflict() + store.save_note(result.winner) + if result.conflict_copy: + store.save_note(result.conflict_copy) + + names = sorted(p.name for p in store.notes_dir.glob("*.md")) + assert "Doc.md" in names + # Conflict copy gets a separate file (different title -> different stem). + assert any("conflit" in n.lower() or "doc" in n.lower() for n in names) + store.close() + + def test_conflict_through_sync_engine(self, tmp_path): + """End-to-end: local modified + remote different → resolver runs and + the resulting note(s) land on disk in the plain layout.""" + store = PlainMarkdownNoteStore(tmp_path / "sync_conflict") + + # Pre-existing local note marked MODIFIED so the engine treats the + # remote pull as a potential conflict. + local = Note(title="Conflict", note_type=NoteType.TYPED) + local.id = "89abcdef-1111-2222-3333-444455556666" + local.add_page().typed_content = "local edit" + local.sync_status = SyncStatus.MODIFIED + local.updated_at = datetime.now(timezone.utc) + timedelta(seconds=5) + store.save_note(local) + + engine = NexaNoteSyncEngine( + store, + SyncConfig( + server_url="http://test/", + username="u", + password="p", + conflict_strategy=ConflictStrategy.KEEP_BOTH, + ), + ) + + remote = _stub_remote_note( + local.id, + "Conflict", + "remote edit", + "2026-01-01T00:00:00+00:00", + ) + _patch_engine_for_pull(engine, remote) + engine._push = lambda report: None + + report = engine.sync() + assert report.success(), report.errors + assert report.conflicts_resolved == 1 + + # Both winner and conflict copy are persisted. + all_notes = store.list_notes() + assert len(all_notes) >= 2 + store.close() + + +# --------------------------------------------------------------------------- +# API integration — `create_app` works the same with the plain backend +# --------------------------------------------------------------------------- + + +class TestApiPlain: + @pytest.fixture + def client(self, tmp_path): + db = PlainMarkdownNoteStore(tmp_path / "api_plain") + app = create_app(db) + with TestClient(app) as c: + yield c, db + db.close() + + def test_create_via_api(self, client): + c, db = client + resp = c.post("/notes", json={"title": "Via API", "note_type": "typed"}) + assert resp.status_code == 201 + assert (db.notes_dir / "Via API.md").exists() + assert (db.notes_dir / "Via API.json").exists() + + def test_rename_via_api(self, client): + c, db = client + created = c.post( + "/notes", json={"title": "Old via API", "note_type": "typed"} + ).json() + c.put(f"/notes/{created['id']}", json={"title": "New via API"}) + assert (db.notes_dir / "New via API.md").exists() + assert not (db.notes_dir / "Old via API.md").exists() + + def test_text_update_via_api(self, client): + c, db = client + created = c.post( + "/notes", json={"title": "Editable", "note_type": "typed"} + ).json() + c.put( + f"/notes/{created['id']}/pages/1/text", + json={"typed_content": "edited via API"}, + ) + body = (db.notes_dir / "Editable.md").read_text("utf-8") + assert "edited via API" in body + + def test_delete_via_api_soft_then_purge(self, client): + c, db = client + created = c.post( + "/notes", json={"title": "Trash", "note_type": "typed"} + ).json() + c.delete(f"/notes/{created['id']}") + # Soft delete: file still on disk but excluded from default listing. + assert (db.notes_dir / "Trash.md").exists() + meta = _read_sidecar(db, "Trash") + assert meta["is_deleted"] is True + + listed_titles = [n["title"] for n in c.get("/notes").json()] + assert "Trash" not in listed_titles