From 72eb1b4a0cb34086b86ed7cab71b68e5adfef061 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 3 May 2026 07:30:28 +0000 Subject: [PATCH] fix: create WebDAV parent directories during sync push MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Push was failing on the new file-based backend with 409 Conflicts on note metadata/page uploads and a 500 Internal Server Error on the welcome note. Root causes: - The WebDAV provider had no MKCOL support on the root collection, so notebook directories never persisted server-side. Subsequent PUTs to nested paths returned 409 (parent missing). - NotebookCollection.create_collection minted a fresh note id, so the slug returned to the caller no longer matched the slug the client requested — every follow-up PUT 409'd against the lost slug. - NoteCollection had no create_empty_resource, so PUT to a not-yet- existing page_N.ink (e.g. a new page during sync) bubbled up as a permission error. - NoteMetaFile.get_etag emitted a quoted token, which WsgiDAV's checked_etag rejects — every PUT to an existing note (the welcome note in particular) crashed with 500. - Writer crashes (DB errors, malformed JSON) propagated as bare 500 responses instead of typed DAVErrors. Fix: - RootCollection.create_collection materialises a notebook with an id whose first 8 hex chars match the slug, so MKCOL → PUT round-trips cleanly. NotebookCollection does the same for notes. - NoteCollection.create_empty_resource provisions new pages on PUT. - _NoteMetaWriter migrates the placeholder id to the client's id when the PUT body declares a different one. - get_etag now returns a quote-free token; WsgiDAV adds the quotes. - Writers convert exceptions into DAVError with a useful context_info, so callers get 400/500 with reasons instead of opaque "500 internal". - The server bootstraps the "Uncategorized" fallback notebook so the very first push from a fresh client has a valid parent collection. - The sync client MKCOLs missing parents on a 409 and retries the PUT once. Failure messages now include 5xx body snippets. Tests: - tests/test_webdav_sync_push.py covers slug parsing, MKCOL idempotency, PUT into missing parents (live WsgiDAV server), end-to-end sync push, server bootstrap, and writer error reporting. https://claude.ai/code/session_011hfnfHGLVJR1aUerZ5sZD6 --- nexanote/sync/client.py | 145 ++++++-- nexanote/sync/server.py | 54 ++- nexanote/sync/webdav_provider.py | 556 ++++++++++++++++++++++++------- tests/test_webdav_sync_push.py | 510 ++++++++++++++++++++++++++++ 4 files changed, 1122 insertions(+), 143 deletions(-) create mode 100644 tests/test_webdav_sync_push.py diff --git a/nexanote/sync/client.py b/nexanote/sync/client.py index 5b2dc6b..103769b 100644 --- a/nexanote/sync/client.py +++ b/nexanote/sync/client.py @@ -204,50 +204,139 @@ def get_ink_page(self, notebook_slug: str, note_slug: str, page_num: int) -> Opt def put_note_meta( self, notebook_slug: str, note_slug: str, data: dict ) -> tuple[bool, Optional[str]]: - """PUT /{notebook}/{note}/note.json — returns (ok, reason_if_failed).""" + """ + EN: PUT /{notebook}/{note}/note.json. On 409 (parent missing) we + transparently MKCOL the parents and retry once — keeps push + reliable when the server hasn't seen this notebook/note yet. + FR: PUT note.json. Si le serveur renvoie 409 (parent absent), on + crée les dossiers parents en MKCOL et on retente une fois. + Returns (ok, reason_if_failed). + """ url = self._url(notebook_slug, note_slug, "note.json") - try: - resp = self.session.put( - url, - json=data, - headers={"Content-Type": "application/json"}, - timeout=self.config.timeout_seconds, - ) - if resp.status_code in (200, 201, 204): - return True, None - reason = f"WebDAV upload failed: {resp.status_code} {resp.reason or ''}".strip() - logger.error(f"PUT note.json échoué ({url}): {reason}") - return False, reason - except requests.RequestException as e: - reason = _sanitize_request_error(e) - logger.error(f"PUT note.json échoué ({url}): {reason}") - return False, reason + return self._put_json( + url, + data, + label="note.json", + mkcol_paths=(notebook_slug, f"{notebook_slug}/{note_slug}"), + ) def put_ink_page( self, notebook_slug: str, note_slug: str, page_num: int, data: dict ) -> tuple[bool, Optional[str]]: - """PUT /{notebook}/{note}/page_N.ink — returns (ok, reason_if_failed).""" + """ + EN: PUT /{notebook}/{note}/page_N.ink with the same MKCOL-on-409 + recovery as ``put_note_meta``. + FR: PUT page_N.ink avec la même récupération MKCOL sur 409. + """ url = self._url(notebook_slug, note_slug, f"page_{page_num}.ink") - try: - resp = self.session.put( + return self._put_json( + url, + data, + label=f"page_{page_num}.ink", + mkcol_paths=(notebook_slug, f"{notebook_slug}/{note_slug}"), + page_num=page_num, + ) + + def _put_json( + self, + url: str, + data: dict, + label: str, + mkcol_paths: tuple[str, ...], + page_num: Optional[int] = None, + ) -> tuple[bool, Optional[str]]: + """ + EN: Internal helper that performs PUT with a single MKCOL-and-retry + cycle when the server returns 409 (parent collection missing). + Treats every non-2xx response uniformly so callers get a stable + (ok, reason) shape. + FR: Helper interne pour PUT avec récupération MKCOL+retry sur 409. + """ + prefix = f"{label}: " if page_num is None else f"page {page_num}: " + + def _do_put() -> requests.Response: + return self.session.put( url, json=data, headers={"Content-Type": "application/json"}, timeout=self.config.timeout_seconds, ) + + try: + resp = _do_put() if resp.status_code in (200, 201, 204): return True, None - reason = ( - f"WebDAV upload failed (page {page_num}): " - f"{resp.status_code} {resp.reason or ''}" - ).strip() - logger.error(f"PUT page.ink échoué ({url}): {reason}") + + if resp.status_code == 409 and self._mkcol_chain(mkcol_paths): + logger.info( + "PUT %s returned 409 — MKCOL'd parents, retrying once", + label, + ) + resp = _do_put() + if resp.status_code in (200, 201, 204): + return True, None + + reason = self._format_http_failure(resp, prefix) + logger.error(f"PUT {label} échoué ({url}): {reason}") return False, reason - except requests.RequestException as e: - reason = f"page {page_num}: {_sanitize_request_error(e)}" - logger.error(f"PUT page.ink échoué ({url}): {reason}") + except requests.RequestException as exc: + reason = f"{prefix}{_sanitize_request_error(exc)}" + logger.error(f"PUT {label} échoué ({url}): {reason}") return False, reason + @staticmethod + def _format_http_failure(resp: requests.Response, prefix: str) -> str: + """ + EN: Turn a non-2xx response into a short user-safe message. We + include the body for 5xx so backend errors aren't reduced to + an opaque "500" — but cap the length so leaked stack traces + don't bloat the UI. + FR: Formate une réponse non-2xx en message court. Pour les 5xx, + on inclut le corps tronqué pour exposer le motif réel. + """ + base = f"{prefix}WebDAV upload failed: {resp.status_code} {resp.reason or ''}".strip() + if 500 <= resp.status_code < 600: + try: + body = resp.text.strip() + except Exception: + body = "" + if body: + snippet = body[:200].replace("\n", " ") + if len(body) > 200: + snippet += "…" + return f"{base} — {snippet}" + return base + + def _mkcol_chain(self, paths: tuple[str, ...]) -> bool: + """ + EN: MKCOL each path in order. Returns True if every step yielded a + success-equivalent status (created or already-exists). Used to + heal a 409 caused by a missing parent collection. + FR: MKCOL chaque chemin dans l'ordre. Retourne True si chaque étape + a réussi (créé ou déjà présent). Sert à corriger un 409 dû à + un parent manquant. + """ + for path in paths: + url = urljoin(self.base_url, "/".join(quote(p, safe="") for p in path.split("/") if p)) + try: + resp = self.session.request( + "MKCOL", + url, + timeout=self.config.timeout_seconds, + ) + except requests.RequestException as exc: + logger.warning(f"MKCOL chain failed at {path}: {exc}") + return False + if not self._is_mkcol_success(resp.status_code): + logger.warning( + "MKCOL chain rejected at %s: %s %s", + path, + resp.status_code, + resp.reason, + ) + return False + return True + def create_notebook_dir(self, notebook_slug: str) -> bool: """MKCOL /{notebook} — creates a notebook folder on the remote server.""" url = self._url(notebook_slug) diff --git a/nexanote/sync/server.py b/nexanote/sync/server.py index 3864b5d..57c9368 100644 --- a/nexanote/sync/server.py +++ b/nexanote/sync/server.py @@ -20,9 +20,18 @@ from cheroot import wsgi from wsgidav.wsgidav_app import WsgiDAVApp +from nexanote.models.note import Notebook from nexanote.storage import FileNoteStore, run_migration from nexanote.sync.webdav_provider import NexaNoteDAVProvider +# EN: Slug for the fallback notebook used to host notes that aren't assigned +# to any user-created notebook. Mirrors the constant in sync.client so +# push targets always have a valid parent collection on the server. +# FR: Slug du carnet de repli pour les notes sans carnet attribué. Identique +# au constant côté client : garantit un parent valide pour les PUT. +DEFAULT_NOTEBOOK_NAME = "Uncategorized" +DEFAULT_NOTEBOOK_ID_PREFIX = "00000000" + logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", @@ -36,6 +45,39 @@ def _hash_password(password: str) -> str: return hashlib.sha256(password.encode()).hexdigest() +def ensure_storage_layout(db: FileNoteStore) -> None: + """ + EN: Make sure the on-disk directories WebDAV expects (notes/, drawings/, + notebooks/) exist. ``FileNoteStore.__init__`` already creates them, + but call this defensively before serving so a wiped data dir during + runtime is recreated rather than throwing on the first request. + FR: S'assure que les dossiers attendus par WebDAV existent. Idempotent. + """ + for d in (db.notes_dir, db.drawings_dir, db.notebooks_dir): + d.mkdir(parents=True, exist_ok=True) + + +def ensure_default_notebook(db: FileNoteStore) -> Notebook: + """ + EN: Ensure the fallback "Uncategorized" notebook exists. Notes pushed + without a notebook id land here, so its parent collection must be + present on the server before the client's first PUT. + FR: Garantit que le carnet "Uncategorized" existe. Les notes sans + carnet y sont rangées : son dossier doit déjà être créé côté serveur. + """ + for nb in db.list_notebooks(): + if nb.id.startswith(DEFAULT_NOTEBOOK_ID_PREFIX): + return nb + notebook = Notebook( + id=f"{DEFAULT_NOTEBOOK_ID_PREFIX}-0000-0000-0000-000000000000", + name=DEFAULT_NOTEBOOK_NAME, + color="#6b7280", + ) + db.save_notebook(notebook) + logger.info("Default notebook created: %s", DEFAULT_NOTEBOOK_NAME) + return notebook + + def build_app( db: FileNoteStore, username: str = "nexanote", @@ -114,8 +156,16 @@ def run_server( logger.info(migration_report.summary()) db = FileNoteStore(data_dir) - # Créer un carnet de démonstration si le store est vide - notebooks = db.list_notebooks() + # Ensure required WebDAV directories exist + the fallback notebook is + # present, so the very first sync push has valid parent collections. + ensure_storage_layout(db) + ensure_default_notebook(db) + + # Créer un carnet de démonstration si seul le carnet de repli existe. + notebooks = [ + nb for nb in db.list_notebooks() + if not nb.id.startswith(DEFAULT_NOTEBOOK_ID_PREFIX) + ] if not notebooks: from nexanote.models.note import Note, Notebook, NoteType nb = Notebook(name="Mon premier carnet", color="#6366f1") diff --git a/nexanote/sync/webdav_provider.py b/nexanote/sync/webdav_provider.py index a181c59..ca2f3bc 100644 --- a/nexanote/sync/webdav_provider.py +++ b/nexanote/sync/webdav_provider.py @@ -19,11 +19,24 @@ import io import json import logging +import re +import uuid from datetime import datetime, timezone from typing import Optional from wsgidav.dav_provider import DAVCollection, DAVNonCollection, DAVProvider -from wsgidav.dav_error import DAVError, HTTP_FORBIDDEN, HTTP_NOT_FOUND +from wsgidav import dav_error as _dav_error +from wsgidav.dav_error import DAVError + +# EN: Some WsgiDAV builds expose HTTP_* both at module level and via the +# `dav_error` namespace. Pull them via getattr to stay forward-compatible. +# FR: Certaines versions exposent les codes HTTP aux deux endroits. On y +# accède via getattr pour rester compatible. +HTTP_FORBIDDEN = getattr(_dav_error, "HTTP_FORBIDDEN", 403) +HTTP_NOT_FOUND = getattr(_dav_error, "HTTP_NOT_FOUND", 404) +HTTP_CONFLICT = getattr(_dav_error, "HTTP_CONFLICT", 409) +HTTP_INTERNAL_ERROR = getattr(_dav_error, "HTTP_INTERNAL_ERROR", 500) +HTTP_BAD_REQUEST = getattr(_dav_error, "HTTP_BAD_REQUEST", 400) from nexanote.models.note import InkStroke, Note, Notebook, NoteType, Page, Point from nexanote.storage.file_store import FileNoteStore @@ -31,19 +44,109 @@ logger = logging.getLogger("nexanote.webdav") +# --------------------------------------------------------------------------- +# Slug helpers +# --------------------------------------------------------------------------- + +# EN: Slugs follow `__`. The id-prefix is the first +# 8 hex chars of a UUID — used to disambiguate notes/notebooks that +# share a title. +# FR: Les slugs suivent `__`. Le préfixe ID = 8 hex +# du UUID, pour différencier des notes au même titre. + +_SLUG_SEPARATOR = "__" +_HEX_RE = re.compile(r"^[0-9a-f]{1,16}$") + +# EN: The fallback "Uncategorized" notebook is exposed at a bare-name slug so +# clients can target it without knowing its synthetic id. The literal id +# prefix (all zeros) is also accepted for legacy callers. +# FR: Le carnet "Uncategorized" est exposé sous un slug court ("uncategorized") +# pour que les clients n'aient pas à connaître l'ID interne. +_DEFAULT_NB_ID_PREFIX = "00000000" +_DEFAULT_NB_SLUG = "uncategorized" + + def _slugify(name: str) -> str: """Transforme un nom en slug URL-safe.""" - import re name = name.strip().lower() name = re.sub(r"[^\w\s-]", "", name) name = re.sub(r"[\s_-]+", "-", name) return name or "sans-titre" +def _parse_slug(slug: str) -> tuple[str, Optional[str]]: + """ + EN: Parse a slug like ``my-note__abcd1234`` into a human-readable title + and an optional 8-char hex id-prefix. If no separator is present, + the prefix is None. + FR: Décompose un slug en (titre lisible, préfixe d'ID). + """ + if _SLUG_SEPARATOR in slug: + title_slug, _, id_prefix = slug.rpartition(_SLUG_SEPARATOR) + if id_prefix and _HEX_RE.match(id_prefix): + title = title_slug.replace("-", " ").strip().title() or "Sans titre" + return title, id_prefix + title = slug.replace("-", " ").strip().title() or "Sans titre" + return title, None + + +def _notebook_slug(nb: Notebook) -> str: + """ + EN: Canonical slug for a notebook. The synthetic "Uncategorized" notebook + gets a bare slug so it lines up with the well-known fallback name + used by sync clients. + FR: Slug canonique d'un carnet. "Uncategorized" est exposé sous le slug + court connu des clients de synchro. + """ + if nb.id.startswith(_DEFAULT_NB_ID_PREFIX): + return _DEFAULT_NB_SLUG + return _slugify(nb.name) + _SLUG_SEPARATOR + nb.id[:8] + + +def _id_with_prefix(prefix: Optional[str]) -> str: + """ + EN: Mint a UUID-formatted id whose first 8 hex chars match ``prefix`` + (when given). Lets the WebDAV slug round-trip cleanly through + MKCOL → PUT → PROPFIND without mismatches. + FR: Génère un UUID dont les 8 premiers caractères correspondent à + ``prefix`` — pour que le slug WebDAV reste stable. + """ + fresh = uuid.uuid4().hex + if prefix and _HEX_RE.match(prefix) and len(prefix) >= 1: + prefix_hex = prefix.lower()[:8].ljust(8, "0") + fresh = prefix_hex + fresh[8:] + return f"{fresh[0:8]}-{fresh[8:12]}-{fresh[12:16]}-{fresh[16:20]}-{fresh[20:32]}" + + def _epoch(dt: datetime) -> float: return dt.replace(tzinfo=timezone.utc).timestamp() +def _safe_etag(dt: datetime) -> str: + """ + EN: Return an ETag token for ``dt`` that satisfies WsgiDAV's rules: + no surrounding quotes, no embedded quotes, not a weak tag. WsgiDAV + wraps it in quotes when emitting the header. + FR: Token ETag conforme à WsgiDAV (sans guillemets, ni W/). + """ + return dt.replace(tzinfo=timezone.utc).isoformat().replace('"', "") + + +def _safe_dav_error(exc: BaseException, default_msg: str) -> DAVError: + """ + EN: Wrap any non-DAVError exception into a 500 with a short, useful + message — so writers don't crash WsgiDAV with a bare traceback. + FR: Convertit une exception en DAVError 500 avec un message lisible. + """ + if isinstance(exc, DAVError): + return exc + return DAVError( + HTTP_INTERNAL_ERROR, + context_info=f"{default_msg}: {type(exc).__name__}: {exc}", + src_exception=exc, + ) + + # --------------------------------------------------------------------------- # Ressources DAV # --------------------------------------------------------------------------- @@ -52,6 +155,7 @@ class RootCollection(DAVCollection): """ Racine DAV → liste tous les carnets. URL : / + Supporte aussi MKCOL pour créer un nouveau carnet. """ def __init__(self, path: str, environ: dict, db: FileNoteStore) -> None: @@ -61,20 +165,58 @@ def __init__(self, path: str, environ: dict, db: FileNoteStore) -> None: def get_member_names(self) -> list[str]: notebooks = self.db.list_notebooks() # On expose chaque carnet comme un dossier slug - return [_slugify(nb.name) + "__" + nb.id[:8] for nb in notebooks] + return [_notebook_slug(nb) for nb in notebooks] def get_member(self, name: str) -> Optional[DAVCollection]: - notebooks = self.db.list_notebooks() - for nb in notebooks: - slug = _slugify(nb.name) + "__" + nb.id[:8] - if slug == name: - return NotebookCollection( - self.path.rstrip("/") + "/" + name, - self.environ, - self.db, - nb, - ) - return None + notebook = _find_notebook_by_slug(self.db, name) + if notebook is None: + return None + return NotebookCollection( + self.path.rstrip("/") + "/" + name, + self.environ, + self.db, + notebook, + ) + + def create_collection(self, name: str) -> "NotebookCollection": + """ + EN: MKCOL on the root creates a fresh notebook. The slug is parsed + so the on-disk notebook keeps the same id-prefix the client + advertised — this way the slug is stable across MKCOL → PUT. + FR: MKCOL à la racine crée un nouveau carnet. Le slug est analysé + pour préserver le préfixe d'ID — slug stable entre les requêtes. + """ + if not name or "/" in name: + raise DAVError(HTTP_BAD_REQUEST, context_info="invalid notebook name") + + existing = _find_notebook_by_slug(self.db, name) + if existing is not None: + # Idempotent — return the existing notebook so retries don't fail. + return NotebookCollection( + self.path.rstrip("/") + "/" + name, + self.environ, + self.db, + existing, + ) + + title, id_prefix = _parse_slug(name) + try: + notebook = Notebook( + id=_id_with_prefix(id_prefix), + name=title, + ) + self.db.save_notebook(notebook) + except Exception as exc: + logger.exception("MKCOL notebook failed: %s", name) + raise _safe_dav_error(exc, "could not create notebook") from exc + + logger.info("Carnet créé via WebDAV MKCOL : %s", title) + return NotebookCollection( + self.path.rstrip("/") + "/" + name, + self.environ, + self.db, + notebook, + ) class NotebookCollection(DAVCollection): @@ -105,36 +247,58 @@ def get_last_modified(self) -> float: def get_member_names(self) -> list[str]: notes = self.db.list_notes(notebook_id=self.notebook.id) - return [_slugify(n.title) + "__" + n.id[:8] for n in notes] + return [_slugify(n.title) + _SLUG_SEPARATOR + n.id[:8] for n in notes] def get_member(self, name: str) -> Optional[DAVCollection]: - notes = self.db.list_notes(notebook_id=self.notebook.id) - for note in notes: - slug = _slugify(note.title) + "__" + note.id[:8] - if slug == name: - full_note = self.db.get_note(note.id, load_pages=True) - return NoteCollection( - self.path.rstrip("/") + "/" + name, - self.environ, - self.db, - full_note, - ) - return None - - def create_collection(self, name: str) -> "NotebookCollection": - """Créer une nouvelle note via MKCOL.""" - from nexanote.models.note import Note - note = Note( - notebook_id=self.notebook.id, - title=name.replace("-", " ").title(), + note = _find_note_by_slug(self.db, self.notebook.id, name) + if note is None: + return None + full_note = self.db.get_note(note.id, load_pages=True) + return NoteCollection( + self.path.rstrip("/") + "/" + name, + self.environ, + self.db, + full_note, ) - note.add_page() - self.db.save_note(note) - logger.info(f"Note créée via WebDAV : {note.title}") - slug = _slugify(note.title) + "__" + note.id[:8] + + def create_collection(self, name: str) -> "NoteCollection": + """ + EN: MKCOL on a notebook creates a fresh note. The note's id-prefix + is taken from the slug so the path stays valid for the follow- + up PUT note.json/page_N.ink — no slug churn between requests. + FR: MKCOL sur un carnet crée une note. Le préfixe d'ID est repris + du slug pour que le PUT suivant trouve la bonne ressource. + """ + if not name or "/" in name: + raise DAVError(HTTP_BAD_REQUEST, context_info="invalid note name") + + existing = _find_note_by_slug(self.db, self.notebook.id, name) + if existing is not None: + full_note = self.db.get_note(existing.id, load_pages=True) + return NoteCollection( + self.path.rstrip("/") + "/" + name, + self.environ, + self.db, + full_note, + ) + + title, id_prefix = _parse_slug(name) + try: + note = Note( + id=_id_with_prefix(id_prefix), + notebook_id=self.notebook.id, + title=title, + ) + note.add_page() + self.db.save_note(note) + except Exception as exc: + logger.exception("MKCOL note failed: %s", name) + raise _safe_dav_error(exc, "could not create note") from exc + + logger.info("Note créée via WebDAV MKCOL : %s", note.title) full_note = self.db.get_note(note.id, load_pages=True) return NoteCollection( - self.path.rstrip("/") + "/" + slug, + self.path.rstrip("/") + "/" + name, self.environ, self.db, full_note, @@ -181,12 +345,8 @@ def get_member(self, name: str) -> Optional[DAVNonCollection]: self.db, self.note, ) - # page_N.ink - if name.startswith("page_") and name.endswith(".ink"): - try: - page_num = int(name[5:-4]) - except ValueError: - return None + page_num = _parse_page_filename(name) + if page_num is not None: page = self.note.get_page(page_num) if page: return InkFile( @@ -198,6 +358,108 @@ def get_member(self, name: str) -> Optional[DAVNonCollection]: ) return None + def create_empty_resource(self, name: str) -> DAVNonCollection: + """ + EN: PUT to a not-yet-existing file inside a note (typically a new + page_N.ink). We allocate the page on the fly so the writer can + stream content into it. note.json always returns the existing + placeholder (a note always has metadata). + FR: PUT sur un fichier inexistant (ex: page_N.ink d'une nouvelle + page). On crée la page à la volée pour que l'écriture aboutisse. + """ + if name == "note.json": + # note.json always exists conceptually — return the writable view. + return NoteMetaFile( + self.path.rstrip("/") + "/" + name, + self.environ, + self.db, + self.note, + ) + + page_num = _parse_page_filename(name) + if page_num is None: + raise DAVError( + HTTP_FORBIDDEN, + context_info=f"unsupported file name: {name}", + ) + + # Create a new (empty) page on the note so the upcoming write target + # is well-defined. The actual content is filled by `_InkWriter`. + try: + page = self.note.get_page(page_num) + if page is None: + page = Page(note_id=self.note.id, page_number=page_num) + self.note.pages.append(page) + self.note.pages.sort(key=lambda p: p.page_number) + self.note.touch() + self.db.save_note(self.note) + except Exception as exc: + logger.exception("create_empty_resource failed: %s", name) + raise _safe_dav_error(exc, "could not create page") from exc + + return InkFile( + self.path.rstrip("/") + "/" + name, + self.environ, + self.db, + page, + self.note, + ) + + +def _parse_page_filename(name: str) -> Optional[int]: + """Return the page number for ``page_.ink`` or None.""" + if not name.startswith("page_") or not name.endswith(".ink"): + return None + try: + return int(name[len("page_"): -len(".ink")]) + except ValueError: + return None + + +def _find_notebook_by_slug(db: FileNoteStore, slug: str) -> Optional[Notebook]: + """ + EN: Resolve a path component to a notebook. Tries the canonical slug, + then id-prefix (handles renames), then slugified name (handles + slugs without an id, like the well-known ``uncategorized``). + FR: Résout un composant de chemin vers un carnet. + """ + notebooks = db.list_notebooks() + for nb in notebooks: + if _notebook_slug(nb) == slug: + return nb + _, id_prefix = _parse_slug(slug) + if id_prefix: + for nb in notebooks: + if nb.id.startswith(id_prefix): + return nb + if _SLUG_SEPARATOR not in slug: + for nb in notebooks: + if _slugify(nb.name) == slug: + return nb + return None + + +def _find_note_by_slug(db: FileNoteStore, notebook_id: str, slug: str) -> Optional[Note]: + """Same lookup strategy as ``_find_notebook_by_slug`` for notes.""" + notes = db.list_notes(notebook_id=notebook_id) + for n in notes: + if (_slugify(n.title) + _SLUG_SEPARATOR + n.id[:8]) == slug: + return n + _, id_prefix = _parse_slug(slug) + if id_prefix: + for n in notes: + if n.id.startswith(id_prefix): + return n + if _SLUG_SEPARATOR not in slug: + for n in notes: + if _slugify(n.title) == slug: + return n + return None + + +# --------------------------------------------------------------------------- +# File-like resources (note.json, page_N.ink) +# --------------------------------------------------------------------------- class NoteMetaFile(DAVNonCollection): """ @@ -209,7 +471,11 @@ def support_etag(self) -> bool: return True def get_etag(self) -> Optional[str]: - return f'"{self.note.updated_at.isoformat()}"' + # WsgiDAV's `checked_etag` rejects values containing quotes — return + # the raw token; the framework adds the surrounding quotes for the + # ETag header itself. Returning a pre-quoted value used to crash + # WsgiDAV with a 500 on PUT to existing notes (e.g. the welcome note). + return _safe_etag(self.note.updated_at) def __init__( self, @@ -271,26 +537,83 @@ def write(self, data: bytes) -> int: return self._buf.write(data) def close(self) -> None: - if not self.closed: + if self.closed: + super().close() + return + + try: self._buf.seek(0) - try: - payload = json.loads(self._buf.read().decode("utf-8")) - self.note.title = payload.get("title", self.note.title) - self.note.tags = payload.get("tags", self.note.tags) - self.note.is_pinned = payload.get("is_pinned", self.note.is_pinned) - # Mettre à jour le contenu texte des pages - for page_data in payload.get("pages", []): - page = self.note.get_page(page_data["page_number"]) - if page: - page.typed_content = page_data.get( - "typed_content", page.typed_content - ) - self.note.touch() - self.db.save_note(self.note) - logger.info(f"Note mise à jour via WebDAV PUT : {self.note.title}") - except (json.JSONDecodeError, KeyError) as e: - logger.error(f"Erreur parsing note.json : {e}") - super().close() + raw = self._buf.read() + finally: + super().close() + + try: + payload = json.loads(raw.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + logger.error("note.json: invalid JSON body: %s", exc) + raise DAVError( + HTTP_BAD_REQUEST, + context_info=f"invalid note.json body: {exc}", + ) from exc + + try: + # If MKCOL minted a placeholder id that the PUT body now wants to + # replace (e.g. client and server both generated UUIDs sharing + # only the slug's 8-char prefix), drop the placeholder so the + # client's full id wins. Without this the on-disk file id would + # diverge from the client's id and break future updates. + payload_id = payload.get("id") + if ( + isinstance(payload_id, str) + and payload_id + and payload_id != self.note.id + ): + old_id = self.note.id + self.note.id = payload_id + for page in self.note.pages: + page.note_id = payload_id + try: + self.db.delete_note_permanent(old_id) + except Exception: + logger.warning( + "could not delete placeholder note %s — ignoring", old_id + ) + + self.note.title = payload.get("title", self.note.title) + self.note.tags = payload.get("tags", self.note.tags) or [] + self.note.is_pinned = bool( + payload.get("is_pinned", self.note.is_pinned) + ) + note_type = payload.get("type") + if note_type: + try: + self.note.note_type = NoteType(note_type) + except ValueError: + logger.warning("ignoring unknown note type: %s", note_type) + for page_data in payload.get("pages") or []: + page_number = page_data.get("page_number") + if page_number is None: + continue + page = self.note.get_page(page_number) + if page is None: + page = Page( + note_id=self.note.id, + page_number=int(page_number), + template=page_data.get("template", "blank"), + ) + self.note.pages.append(page) + page.typed_content = page_data.get( + "typed_content", page.typed_content + ) + self.note.pages.sort(key=lambda p: p.page_number) + self.note.touch() + self.db.save_note(self.note) + logger.info("Note mise à jour via WebDAV PUT : %s", self.note.title) + except DAVError: + raise + except Exception as exc: + logger.exception("note.json write failed") + raise _safe_dav_error(exc, "saving note failed") from exc class InkFile(DAVNonCollection): @@ -304,7 +627,7 @@ def support_etag(self) -> bool: return True def get_etag(self) -> Optional[str]: - return f'"{self.page.updated_at.isoformat()}"' + return _safe_etag(self.page.updated_at) def __init__( self, @@ -379,42 +702,61 @@ def write(self, data: bytes) -> int: return self._buf.write(data) def close(self) -> None: - if not self.closed: + if self.closed: + super().close() + return + + try: self._buf.seek(0) - try: - payload = json.loads(self._buf.read().decode("utf-8")) - new_strokes = [] - for s_data in payload.get("strokes", []): - points = [ - Point( - x=p["x"], - y=p["y"], - pressure=p.get("pressure", 0.5), - timestamp_ms=p.get("ts", 0), - ) - for p in s_data.get("points", []) - ] - stroke = InkStroke( - id=s_data["id"], - color=s_data.get("color", "#000000"), - width=s_data.get("width", 2.0), - tool=s_data.get("tool", "pen"), - points=points, + raw = self._buf.read() + finally: + super().close() + + try: + payload = json.loads(raw.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + logger.error("page_N.ink: invalid JSON body: %s", exc) + raise DAVError( + HTTP_BAD_REQUEST, + context_info=f"invalid ink page body: {exc}", + ) from exc + + try: + new_strokes: list[InkStroke] = [] + for s_data in payload.get("strokes") or []: + points = [ + Point( + x=p["x"], + y=p["y"], + pressure=p.get("pressure", 0.5), + timestamp_ms=p.get("ts", 0), ) - new_strokes.append(stroke) - - self.page.strokes = new_strokes - self.page.touch() - self.db.save_page(self.page) - self.note.touch() - self.db.save_note(self.note, save_pages=False) - logger.info( - f"Page {self.page.page_number} mise à jour : " - f"{len(new_strokes)} strokes" + for p in s_data.get("points", []) + ] + stroke = InkStroke( + id=s_data["id"], + color=s_data.get("color", "#000000"), + width=s_data.get("width", 2.0), + tool=s_data.get("tool", "pen"), + points=points, ) - except (json.JSONDecodeError, KeyError) as e: - logger.error(f"Erreur parsing .ink : {e}") - super().close() + new_strokes.append(stroke) + + self.page.strokes = new_strokes + self.page.touch() + self.db.save_page(self.page) + self.note.touch() + self.db.save_note(self.note, save_pages=False) + logger.info( + "Page %d mise à jour : %d strokes", + self.page.page_number, + len(new_strokes), + ) + except DAVError: + raise + except Exception as exc: + logger.exception("page ink write failed") + raise _safe_dav_error(exc, "saving page failed") from exc # --------------------------------------------------------------------------- @@ -447,13 +789,7 @@ def get_resource_inst( return RootCollection("/", environ, self.db) # Niveau carnet - notebooks = self.db.list_notebooks() - target_nb = None - for nb in notebooks: - if (_slugify(nb.name) + "__" + nb.id[:8]) == parts[0]: - target_nb = nb - break - + target_nb = _find_notebook_by_slug(self.db, parts[0]) if target_nb is None: return None @@ -463,13 +799,7 @@ def get_resource_inst( return NotebookCollection(nb_path, environ, self.db, target_nb) # Niveau note - notes = self.db.list_notes(notebook_id=target_nb.id) - target_note = None - for note in notes: - if (_slugify(note.title) + "__" + note.id[:8]) == parts[1]: - target_note = note - break - + target_note = _find_note_by_slug(self.db, target_nb.id, parts[1]) if target_note is None: return None diff --git a/tests/test_webdav_sync_push.py b/tests/test_webdav_sync_push.py new file mode 100644 index 0000000..cf91428 --- /dev/null +++ b/tests/test_webdav_sync_push.py @@ -0,0 +1,510 @@ +""" +NexaNote — End-to-end tests for WebDAV sync push with file-based storage. + +EN: These tests spin up a real WsgiDAV WSGI app backed by ``FileNoteStore`` + and exercise the same client code path used during sync. They cover + the bug where PUT to a nested path (note metadata or page ink) returned + 409 because the WebDAV provider didn't auto-create parent collections, + and where a writer crash bubbled up as 500. +FR: Tests bout-en-bout pour le push WebDAV avec le stockage fichier. + Reproduit le bug 409 (parents manquants) + 500 (writer non-géré) et + vérifie que les correctifs tiennent. +""" + +import sys +import threading +import time +from pathlib import Path + +import pytest +import requests +from requests.auth import HTTPBasicAuth + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from cheroot import wsgi as cheroot_wsgi + +from nexanote.models.note import Note, Notebook, NoteType +from nexanote.storage import FileNoteStore +from nexanote.sync.client import ( + DEFAULT_NOTEBOOK_SLUG, + NexaNoteSyncEngine, + SyncConfig, + WebDAVClient, +) +from nexanote.sync.server import ( + DEFAULT_NOTEBOOK_ID_PREFIX, + build_app, + ensure_default_notebook, + ensure_storage_layout, +) +from nexanote.sync.webdav_provider import ( + NexaNoteDAVProvider, + NoteCollection, + NotebookCollection, + _id_with_prefix, + _parse_slug, + _slugify, +) + + +# --------------------------------------------------------------------------- +# Live server fixture — starts a real WebDAV server on a random port +# --------------------------------------------------------------------------- + + +def _free_port() -> int: + import socket + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture +def live_server(tmp_path): + """Run a real WsgiDAV server on a background thread.""" + db = FileNoteStore(tmp_path / "server_store") + ensure_storage_layout(db) + ensure_default_notebook(db) + + app = build_app(db, username="user", password="pass", verbose=False) + port = _free_port() + server = cheroot_wsgi.Server( + bind_addr=("127.0.0.1", port), + wsgi_app=app, + numthreads=4, + request_queue_size=8, + ) + + thread = threading.Thread(target=server.start, daemon=True) + thread.start() + + # Wait for the server to start accepting connections. + base_url = f"http://127.0.0.1:{port}/" + deadline = time.time() + 5 + while time.time() < deadline: + try: + requests.options(base_url, timeout=1) + break + except requests.RequestException: + time.sleep(0.05) + else: + server.stop() + raise RuntimeError("test WebDAV server failed to start") + + yield { + "url": base_url, + "username": "user", + "password": "pass", + "db": db, + } + + server.stop() + db.close() + + +def _make_client(live_server) -> WebDAVClient: + return WebDAVClient( + SyncConfig( + server_url=live_server["url"], + username=live_server["username"], + password=live_server["password"], + timeout_seconds=5, + ) + ) + + +# --------------------------------------------------------------------------- +# Slug helpers +# --------------------------------------------------------------------------- + + +class TestSlugHelpers: + def test_parse_slug_extracts_id_prefix(self): + title, prefix = _parse_slug("ma-note__abcd1234") + assert prefix == "abcd1234" + assert title.lower() == "ma note" + + def test_parse_slug_no_prefix(self): + title, prefix = _parse_slug("ma-note") + assert prefix is None + assert title.lower() == "ma note" + + def test_parse_slug_rejects_non_hex_prefix(self): + # Slugs from the client are always lowercase hex; reject anything else. + title, prefix = _parse_slug("ma-note__ZZZZ1234") + assert prefix is None # non-hex → not a valid id prefix + + def test_id_with_prefix_preserves_first_8_chars(self): + new_id = _id_with_prefix("abcd1234") + assert new_id[:8] == "abcd1234" + assert new_id.count("-") == 4 # canonical UUID layout + + def test_id_with_prefix_handles_none(self): + new_id = _id_with_prefix(None) + assert len(new_id) == 36 + assert new_id.count("-") == 4 + + +# --------------------------------------------------------------------------- +# Provider unit tests — MKCOL & PUT into missing parents +# --------------------------------------------------------------------------- + + +def _make_environ(provider): + return { + "wsgidav.provider": provider, + "wsgidav.config": {}, + "REQUEST_METHOD": "GET", + "SERVER_NAME": "localhost", + "SERVER_PORT": "8765", + "wsgi.url_scheme": "http", + } + + +class TestProviderCreateCollection: + def test_root_mkcol_creates_notebook_with_id_prefix(self, tmp_path): + db = FileNoteStore(tmp_path / "store") + provider = NexaNoteDAVProvider(db) + environ = _make_environ(provider) + root = provider.get_resource_inst("/", environ) + + slug = "new-notebook__deadbeef" + nb_collection = root.create_collection(slug) + assert isinstance(nb_collection, NotebookCollection) + assert nb_collection.notebook.id.startswith("deadbeef") + + # Subsequent path lookup must find the notebook at the same slug. + again = provider.get_resource_inst(f"/{slug}", environ) + assert again is not None + assert isinstance(again, NotebookCollection) + assert again.notebook.id == nb_collection.notebook.id + + def test_root_mkcol_is_idempotent(self, tmp_path): + db = FileNoteStore(tmp_path / "store") + provider = NexaNoteDAVProvider(db) + environ = _make_environ(provider) + root = provider.get_resource_inst("/", environ) + + slug = "same-name__abcd1234" + a = root.create_collection(slug) + b = root.create_collection(slug) + assert a.notebook.id == b.notebook.id + + def test_notebook_mkcol_creates_note_at_same_slug(self, tmp_path): + db = FileNoteStore(tmp_path / "store") + provider = NexaNoteDAVProvider(db) + environ = _make_environ(provider) + + nb = Notebook(name="Notebook") + db.save_notebook(nb) + nb_slug = _slugify(nb.name) + "__" + nb.id[:8] + + nb_col = provider.get_resource_inst(f"/{nb_slug}", environ) + assert isinstance(nb_col, NotebookCollection) + + note_slug = "fresh-note__cafebabe" + note_col = nb_col.create_collection(note_slug) + assert isinstance(note_col, NoteCollection) + assert note_col.note.id.startswith("cafebabe") + + # Path stays valid for the next request — no slug churn. + resolved = provider.get_resource_inst(f"/{nb_slug}/{note_slug}", environ) + assert resolved is not None + assert isinstance(resolved, NoteCollection) + assert resolved.note.id == note_col.note.id + + +class TestProviderCreateEmptyResource: + """PUT into a missing file inside a note must succeed (auto-create).""" + + def test_create_empty_resource_for_new_page(self, tmp_path): + db = FileNoteStore(tmp_path / "store") + provider = NexaNoteDAVProvider(db) + environ = _make_environ(provider) + + nb = Notebook(name="Notebook") + db.save_notebook(nb) + note = Note(notebook_id=nb.id, title="Note") + note.add_page() + db.save_note(note) + full_note = db.get_note(note.id, load_pages=True) + + note_col = NoteCollection("/", environ, db, full_note) + + # page_2.ink doesn't exist yet — provider must allocate it. + ink_file = note_col.create_empty_resource("page_2.ink") + assert ink_file is not None + + reloaded = db.get_note(note.id, load_pages=True) + assert any(p.page_number == 2 for p in reloaded.pages), ( + "create_empty_resource must materialize the new page on disk" + ) + + def test_create_empty_resource_rejects_unknown_filename(self, tmp_path): + from wsgidav.dav_error import DAVError + + db = FileNoteStore(tmp_path / "store") + provider = NexaNoteDAVProvider(db) + environ = _make_environ(provider) + + nb = Notebook(name="Notebook") + db.save_notebook(nb) + note = Note(notebook_id=nb.id, title="Note") + note.add_page() + db.save_note(note) + full_note = db.get_note(note.id, load_pages=True) + + note_col = NoteCollection("/", environ, db, full_note) + + with pytest.raises(DAVError): + note_col.create_empty_resource("rogue.bin") + + +# --------------------------------------------------------------------------- +# Live server: PUT to missing parent must succeed (or be self-healing) +# --------------------------------------------------------------------------- + + +class TestLivePutIntoMissingParent: + """ + EN: Reproduce the bug where the client did MKCOL → PUT and got 409 + because the parent never persisted. The server must now MKCOL + nested paths transparently and the client retries on 409. + """ + + def test_mkcol_root_then_propfind_lists_notebook(self, live_server): + url = live_server["url"] + auth = HTTPBasicAuth(live_server["username"], live_server["password"]) + slug = "fresh-nb__a1b2c3d4" + + resp = requests.request("MKCOL", f"{url}{slug}", auth=auth, timeout=5) + assert resp.status_code in (201, 405), resp.text + + # PROPFIND root and verify the new notebook shows up. + resp = requests.request( + "PROPFIND", + url, + auth=auth, + headers={"Depth": "1"}, + timeout=5, + ) + assert resp.status_code == 207 + assert slug in resp.text + + def test_put_note_json_after_mkcol_chain(self, live_server): + """ + Full happy path: client MKCOL notebook + note, PUTs note.json, + server stores it and a follow-up GET returns the same payload. + """ + url = live_server["url"] + auth = HTTPBasicAuth(live_server["username"], live_server["password"]) + + nb_slug = "carnet__01234567" + note_slug = "ma-note__89abcdef" + + # MKCOL chain + for path in (nb_slug, f"{nb_slug}/{note_slug}"): + resp = requests.request("MKCOL", f"{url}{path}", auth=auth, timeout=5) + assert resp.status_code in (201, 405), f"MKCOL {path}: {resp.status_code}" + + # PUT note.json + payload = { + "id": "89abcdef-1234-5678-9abc-def012345678", + "title": "Ma note synchronisée", + "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": "lined", "typed_content": "Hello"}, + ], + } + resp = requests.put( + f"{url}{nb_slug}/{note_slug}/note.json", + json=payload, + auth=auth, + timeout=5, + ) + assert resp.status_code in (200, 201, 204), ( + f"PUT note.json failed: {resp.status_code} {resp.text}" + ) + + # GET back and ensure the payload round-tripped via the file store. + resp = requests.get( + f"{url}{nb_slug}/{note_slug}/note.json", + auth=auth, + timeout=5, + ) + assert resp.status_code == 200 + roundtripped = resp.json() + assert roundtripped["title"] == "Ma note synchronisée" + assert roundtripped["pages"][0]["typed_content"] == "Hello" + + def test_put_into_missing_parent_returns_409_not_500(self, live_server): + """ + EN: Direct PUT (without MKCOL) to a nested path with no parent must + return 409, not 500 — and the body must hint that the parent + collection is missing rather than leaking a stack trace. + """ + url = live_server["url"] + auth = HTTPBasicAuth(live_server["username"], live_server["password"]) + + resp = requests.put( + f"{url}does-not-exist__deadbeef/note__cafebabe/note.json", + json={"id": "x", "pages": []}, + auth=auth, + timeout=5, + ) + # 409 (parent missing) is the WebDAV-correct answer — not 500. + assert resp.status_code == 409, f"got {resp.status_code}: {resp.text}" + + +class TestSyncClientPushHappyPath: + """End-to-end sync push using the high-level engine.""" + + def _make_engine(self, live_server, client_dir): + local_db = FileNoteStore(client_dir) + engine = NexaNoteSyncEngine( + local_db, + SyncConfig( + server_url=live_server["url"], + username=live_server["username"], + password=live_server["password"], + timeout_seconds=5, + ), + ) + return local_db, engine + + def test_push_persists_notes_on_server(self, live_server, tmp_path): + local_db, engine = self._make_engine(live_server, tmp_path / "client") + + nb = Notebook(name="Mon carnet", color="#3b82f6") + local_db.save_notebook(nb) + note = Note(notebook_id=nb.id, title="Hello sync", note_type=NoteType.TYPED) + page = note.add_page(template="lined") + page.typed_content = "Synced via WebDAV" + local_db.save_note(note) + + report = engine.sync() + assert report.success(), f"sync failed: {report.errors}" + assert report.notes_pushed == 1, report.summary() + + # Verify server side has the note. + server_note = live_server["db"].get_note(note.id, load_pages=True) + assert server_note is not None + assert server_note.title == "Hello sync" + assert server_note.pages[0].typed_content.strip().endswith("Synced via WebDAV") + + def test_push_recovers_from_409_via_mkcol_retry(self, live_server, tmp_path): + """ + EN: Bypass the engine's pre-MKCOL and call ``put_note_meta`` directly + against a path with no parent. The client must MKCOL the parents + on the 409 and retry — the second attempt succeeds. + """ + local_db = FileNoteStore(tmp_path / "client_only_dav") + client = _make_client(live_server) + + nb_slug = "lazy__deadbeef" + note_slug = "lazy-note__01020304" + payload = { + "id": "01020304-aaaa-bbbb-cccc-ddddeeeeffff", + "title": "Lazy note", + "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": "lazy"}, + ], + } + ok, reason = client.put_note_meta(nb_slug, note_slug, payload) + assert ok, f"expected MKCOL-on-409 retry to succeed: {reason}" + + # Server now has the note — the parent collections were auto-created. + members = client.list_notebooks() + assert any(m["name"] == nb_slug for m in members) + + def test_push_uncategorized_notebook_works(self, live_server, tmp_path): + """A note with no notebook_id falls back to the "uncategorized" slug.""" + local_db, engine = self._make_engine(live_server, tmp_path / "client_uncat") + + note = Note(title="Loose note", note_type=NoteType.TYPED) + note.add_page().typed_content = "no notebook" + local_db.save_note(note) + + report = engine.sync() + assert report.success(), f"sync failed: {report.errors}" + # The fallback slug should now exist on the server. + client = _make_client(live_server) + names = {m["name"] for m in client.list_notebooks()} + assert DEFAULT_NOTEBOOK_SLUG in names + + +# --------------------------------------------------------------------------- +# Server bootstrap +# --------------------------------------------------------------------------- + + +class TestServerBootstrap: + def test_ensure_storage_layout_creates_dirs(self, tmp_path): + db = FileNoteStore(tmp_path / "boot") + # Wipe to simulate a clean slate after the constructor created them. + for d in (db.notes_dir, db.drawings_dir, db.notebooks_dir): + for child in d.iterdir(): + child.unlink() + d.rmdir() + ensure_storage_layout(db) + assert db.notes_dir.is_dir() + assert db.drawings_dir.is_dir() + assert db.notebooks_dir.is_dir() + + def test_ensure_default_notebook_idempotent(self, tmp_path): + db = FileNoteStore(tmp_path / "boot2") + a = ensure_default_notebook(db) + b = ensure_default_notebook(db) + assert a.id == b.id + assert a.id.startswith(DEFAULT_NOTEBOOK_ID_PREFIX) + + +# --------------------------------------------------------------------------- +# 500-mitigation: writer errors must surface as DAVError, not Internal Error +# --------------------------------------------------------------------------- + + +class TestWriterErrorReporting: + def test_invalid_note_json_raises_dav_bad_request(self, tmp_path): + from wsgidav.dav_error import DAVError + + db = FileNoteStore(tmp_path / "boot3") + nb = Notebook(name="Notebook") + db.save_notebook(nb) + note = Note(notebook_id=nb.id, title="Note") + note.add_page() + db.save_note(note) + + from nexanote.sync.webdav_provider import _NoteMetaWriter + + writer = _NoteMetaWriter(db, note) + writer.write(b"not-json-at-all") + with pytest.raises(DAVError): + writer.close() + + def test_invalid_ink_payload_raises_dav_bad_request(self, tmp_path): + from wsgidav.dav_error import DAVError + + db = FileNoteStore(tmp_path / "boot4") + nb = Notebook(name="Notebook") + db.save_notebook(nb) + note = Note(notebook_id=nb.id, title="Note") + page = note.add_page() + db.save_note(note) + + from nexanote.sync.webdav_provider import _InkWriter + + writer = _InkWriter(db, page, note) + writer.write(b"{ this is not json") + with pytest.raises(DAVError): + writer.close()