"""Write-back tests: surgical edits, atomicity, conflicts (SPEC 12, tests 3/8).""" from __future__ import annotations import difflib from pathlib import Path import pytest from mdcaldav.config import Config from mdcaldav.index import Index from mdcaldav.model import Status from mdcaldav.writer import ConflictError, NotAllowedError, Writer @pytest.fixture def env(vault: Path, tmp_path: Path): cfg = Config() cfg.vault.path = vault cfg.index.db = tmp_path / "index.db" index = Index(cfg) index.rescan() yield cfg, index, Writer(cfg, index), vault index.close() def uid_of(index: Index, title: str, rel: str | None = None) -> str: matches = [ t for t in index.tasks.values() if t.title == title and (rel is None or t.source.rel_path == rel) ] assert len(matches) == 1, f"{title!r} matched {len(matches)}" return matches[0].uid def diff_lines(before: str, after: str) -> list[str]: return [ line for line in difflib.unified_diff( before.splitlines(), after.splitlines(), lineterm="", n=0 ) if line[:1] in "+-" and not line.startswith(("---", "+++")) ] def test_completion_changes_exactly_one_line(env): _, index, writer, vault = env rel = "daily/2026-01-07.md" before = (vault / rel).read_text() writer.apply(uid_of(index, "Water the seedlings", rel), {"status": Status.COMPLETED}) after = (vault / rel).read_text() assert diff_lines(before, after) == [ "-- [ ] Water the seedlings", "+- [x] Water the seedlings", ] def test_completion_toggles_back(env): _, index, writer, vault = env rel = "daily/2026-01-07.md" before = (vault / rel).read_bytes() uid = uid_of(index, "Water the seedlings", rel) writer.apply(uid, {"status": Status.COMPLETED}) writer.apply(uid, {"status": Status.NEEDS_ACTION}) assert (vault / rel).read_bytes() == before # round trip is byte-identical def test_nested_task_completion_leaves_siblings_alone(env): _, index, writer, vault = env rel = "daily/2026-01-05.md" before = (vault / rel).read_text() writer.apply( uid_of(index, "Decide on trellis height", rel), {"status": Status.COMPLETED} ) after = (vault / rel).read_text() assert diff_lines(before, after) == [ "- - [ ] Decide on trellis height", "+ - [x] Decide on trellis height", ] def test_retitle_preserves_indent_bullet_and_priority(env): _, index, writer, vault = env rel = "daily/2026-01-07.md" writer.apply(uid_of(index, "Book the soil test", rel), {"title": "Book the soil test today"}) assert "- [ ] [#A] Book the soil test today" in (vault / rel).read_text() def test_priority_set_clear_and_change(env): _, index, writer, vault = env rel = "daily/2026-01-07.md" uid = uid_of(index, "Water the seedlings", rel) writer.apply(uid, {"priority": 1}) assert "- [ ] [#A] Water the seedlings" in (vault / rel).read_text() writer.apply(uid, {"priority": 9}) assert "- [ ] [#C] Water the seedlings" in (vault / rel).read_text() writer.apply(uid, {"priority": None}) assert "- [ ] Water the seedlings" in (vault / rel).read_text() assert "[#" not in (vault / rel).read_text().split("Water the seedlings")[0].splitlines()[-1] def test_description_replaced_in_place(env): _, index, writer, vault = env rel = "daily/2026-01-05.md" writer.apply( uid_of(index, "Order the cover crop", rel), {"description": "Invoice filed"} ) text = (vault / rel).read_text() assert " - Invoice filed" in text assert "Arrived on the 12th" not in text def test_description_added_where_none_existed(env): _, index, writer, vault = env rel = "daily/2026-01-07.md" writer.apply( uid_of(index, "Call the arborist", rel), {"description": "Ask about the oak"} ) text = (vault / rel).read_text() assert "- [ ] [#B] Call the arborist\n - Ask about the oak\n" in text def test_description_removed(env): _, index, writer, vault = env rel = "daily/2026-01-05.md" writer.apply(uid_of(index, "Order the cover crop", rel), {"description": None}) text = (vault / rel).read_text() assert "Arrived on the 12th" not in text assert "- [x] Order the cover crop\n - [x] Sharpen the pruners" in text def test_untouched_files_are_not_rewritten(env): _, index, writer, vault = env others = { rel: (vault / rel).read_bytes() for rel in ("daily/2026-01-06.md", "projects/irrigation.md", "reference/glossary.md") } writer.apply( uid_of(index, "Water the seedlings", "daily/2026-01-07.md"), {"status": Status.COMPLETED}, ) for rel, content in others.items(): assert (vault / rel).read_bytes() == content def test_crlf_and_missing_final_newline_preserved(env, tmp_path: Path): cfg, index, writer, vault = env rel = "daily/crlf.md" (vault / rel).write_bytes(b"# T\r\n\r\n- [ ] alpha\r\n- [ ] omega") # no final newline index.rescan() writer.apply(uid_of(index, "alpha", rel), {"status": Status.COMPLETED}) data = (vault / rel).read_bytes() assert data == b"# T\r\n\r\n- [x] alpha\r\n- [ ] omega" def test_unicode_title_edit_keeps_bytes_aligned(env): _, index, writer, vault = env rel = "daily/uni.md" (vault / rel).write_text("- [ ] café → naïve\n- [ ] after\n", encoding="utf-8") index.rescan() writer.apply(uid_of(index, "café → naïve", rel), {"status": Status.COMPLETED}) assert (vault / rel).read_text() == "- [x] café → naïve\n- [ ] after\n" def test_conflict_when_task_deleted_underneath(env): """The task the client addressed no longer exists → refuse, don't guess.""" _, index, writer, vault = env rel = "daily/2026-01-07.md" uid = uid_of(index, "Water the seedlings", rel) text = (vault / rel).read_text() (vault / rel).write_text(text.replace("- [ ] Water the seedlings\n", "")) before = (vault / rel).read_bytes() with pytest.raises(ConflictError): writer.apply(uid, {"status": Status.COMPLETED}) assert (vault / rel).read_bytes() == before # not corrupted def test_external_retitle_is_the_same_task(env): """Identity rule 2 treats an in-place retitle as the same task, so the write lands on the renamed line rather than failing. The client's own staleness is caught a layer up by Radicale's If-Match/ETag check.""" _, index, writer, vault = env rel = "daily/2026-01-07.md" uid = uid_of(index, "Water the seedlings", rel) text = (vault / rel).read_text() (vault / rel).write_text(text.replace("Water the seedlings", "Water the beds")) writer.apply(uid, {"status": Status.COMPLETED}) assert "- [x] Water the beds" in (vault / rel).read_text() def test_stale_spans_are_never_used(env): """An edit that shifts line offsets must not corrupt a later write.""" _, index, writer, vault = env rel = "daily/2026-01-07.md" uid = uid_of(index, "Water the seedlings", rel) # Insert lines above the target, invalidating every cached byte offset. text = (vault / rel).read_text() (vault / rel).write_text(text.replace("## TODO\n", "## TODO\n\nSome new prose here.\n\n")) writer.apply(uid, {"status": Status.COMPLETED}) result = (vault / rel).read_text() assert "- [x] Water the seedlings" in result assert "Some new prose here." in result assert "- [ ] [#A] Book the soil test" in result # neighbours intact def test_unrelated_external_edit_is_absorbed(env): """A change elsewhere in the file must not block an unrelated write.""" _, index, writer, vault = env rel = "daily/2026-01-07.md" uid = uid_of(index, "Water the seedlings", rel) text = (vault / rel).read_text() (vault / rel).write_text(text.replace("# Wednesday", "# Weds")) writer.apply(uid, {"status": Status.COMPLETED}) assert "- [x] Water the seedlings" in (vault / rel).read_text() def test_delete_cascades_to_subtree(env): _, index, writer, vault = env rel = "projects/irrigation.md" writer.delete(uid_of(index, "Trench the main line", rel)) text = (vault / rel).read_text() assert "Trench the main line" not in text assert "Rent the trencher" not in text # cascaded assert "frost line" not in text # its description went too assert "Price the manifold" in text # sibling untouched def test_delete_reject_policy(env): cfg, index, writer, vault = env cfg.write.delete_children = "reject" with pytest.raises(NotAllowedError): writer.delete(uid_of(index, "Trench the main line", "projects/irrigation.md")) def test_delete_disabled(env): cfg, index, writer, _ = env cfg.write.allow_delete = False with pytest.raises(NotAllowedError): writer.delete(uid_of(index, "Reseal the base", "daily/2026-01-06.md")) def test_create_lands_in_inbox(env): _, index, writer, vault = env uid = writer.create({"title": "Buy netting", "priority": 1}) text = (vault / "inbox.md").read_text() assert "## Inbox" in text assert "- [ ] [#A] Buy netting" in text assert index.tasks[uid].title == "Buy netting" def test_create_disabled(env): cfg, _, writer, _ = env cfg.write.allow_create = False with pytest.raises(NotAllowedError): writer.create({"title": "nope"}) def test_atomic_write_leaves_no_temp_files(env): _, index, writer, vault = env writer.apply( uid_of(index, "Water the seedlings", "daily/2026-01-07.md"), {"status": Status.COMPLETED}, ) assert not list((vault / "daily").glob(".*tmp*"))