test_todo_plugin.py
python
sha256:832d1ca80cb25129c91d8c8c190ec30af2e638a3e8160431f4704fd881bceaee
feat: add the todo domain plugin -- Build With Muse Episode 06
Sonnet 5
patch
3 hours ago
| 1 | """Tests for muse/plugins/todo/plugin.py — the Episode 06 reference domain. |
| 2 | |
| 3 | Built live in Build With Muse, Episode 06 ("Build a Domain"). Covers the |
| 4 | six required protocol methods plus two real regressions hit while building |
| 5 | it on camera: |
| 6 | |
| 7 | 1. Object IDs must use the canonical `blob_id()` scheme (Episode 03's |
| 8 | `blob <size>\\0<content>` format), not a bare sha256 of raw bytes. |
| 9 | 2. `diff()`'s top-level ops must be keyed by the *file* address (so |
| 10 | `muse checkout` can restore/delete the right object), with task-level |
| 11 | detail nested as `child_ops` -- not task-level addresses at the top, |
| 12 | which `muse checkout` silently fails to act on. |
| 13 | """ |
| 14 | |
| 15 | from __future__ import annotations |
| 16 | |
| 17 | import pathlib |
| 18 | |
| 19 | from muse.core.object_store import read_object, write_object |
| 20 | from muse.core.types import blob_id |
| 21 | from muse.plugins.todo.plugin import TodoPlugin, _TASK_FILE |
| 22 | |
| 23 | |
| 24 | def _repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 25 | (tmp_path / ".muse" / "objects").mkdir(parents=True) |
| 26 | return tmp_path |
| 27 | |
| 28 | |
| 29 | def _commit_tasks(plugin: TodoPlugin, repo_root: pathlib.Path, tasks: list[str]) -> dict: |
| 30 | body = "\n".join(tasks) + ("\n" if tasks else "") |
| 31 | (repo_root / _TASK_FILE).write_text(body) |
| 32 | snap = plugin.snapshot(repo_root) |
| 33 | if tasks: |
| 34 | raw = body.encode("utf-8") |
| 35 | write_object(repo_root, blob_id(raw), raw) |
| 36 | return snap |
| 37 | |
| 38 | |
| 39 | class TestSchema: |
| 40 | def test_top_level_is_content_addressed_set(self) -> None: |
| 41 | schema = TodoPlugin().schema() |
| 42 | assert schema["top_level"]["kind"] == "set" |
| 43 | assert schema["top_level"]["identity"] == "by_content" |
| 44 | |
| 45 | def test_merge_mode_is_three_way(self) -> None: |
| 46 | assert TodoPlugin().schema()["merge_mode"] == "three_way" |
| 47 | |
| 48 | |
| 49 | class TestSnapshot: |
| 50 | def test_empty_repo_has_no_files(self, tmp_path: pathlib.Path) -> None: |
| 51 | root = _repo(tmp_path) |
| 52 | snap = TodoPlugin().snapshot(root) |
| 53 | assert snap["files"] == {} |
| 54 | |
| 55 | def test_object_id_uses_canonical_blob_format(self, tmp_path: pathlib.Path) -> None: |
| 56 | """Regression: object IDs must be blob_id(), not bare sha256(raw). |
| 57 | |
| 58 | A bare sha256 of raw bytes doesn't match what write_object_from_path |
| 59 | (and the rest of the store) expects -- confirmed live, this made |
| 60 | `muse commit` fail outright with a content-integrity error. |
| 61 | """ |
| 62 | root = _repo(tmp_path) |
| 63 | (root / _TASK_FILE).write_text("Buy milk\n") |
| 64 | snap = TodoPlugin().snapshot(root) |
| 65 | assert snap["files"][_TASK_FILE] == blob_id(b"Buy milk\n") |
| 66 | |
| 67 | |
| 68 | class TestDiff: |
| 69 | def test_no_change_when_hashes_match(self, tmp_path: pathlib.Path) -> None: |
| 70 | root = _repo(tmp_path) |
| 71 | snap = _commit_tasks(TodoPlugin(), root, ["Buy milk"]) |
| 72 | delta = TodoPlugin().diff(snap, snap, repo_root=root) |
| 73 | assert delta["ops"] == [] |
| 74 | |
| 75 | def test_top_level_op_is_keyed_by_file_not_task(self, tmp_path: pathlib.Path) -> None: |
| 76 | """Regression: `muse checkout` restores/deletes files by matching |
| 77 | `op["address"]` against the snapshot manifest's file paths. A |
| 78 | task-level address (e.g. "todo.txt::<hash>") never matches, so |
| 79 | checkout silently restored nothing -- confirmed live, switching |
| 80 | branches left stale content from the previously-checked-out branch |
| 81 | sitting in the working tree with no error at all. |
| 82 | """ |
| 83 | plugin = TodoPlugin() |
| 84 | base = _commit_tasks(plugin, root := tmp_path, []) |
| 85 | target = _commit_tasks(plugin, root, ["Buy milk"]) |
| 86 | delta = plugin.diff(base, target, repo_root=root) |
| 87 | assert len(delta["ops"]) == 1 |
| 88 | assert delta["ops"][0]["op"] == "patch" |
| 89 | assert delta["ops"][0]["address"] == _TASK_FILE |
| 90 | |
| 91 | def test_child_ops_carry_task_level_detail(self, tmp_path: pathlib.Path) -> None: |
| 92 | plugin = TodoPlugin() |
| 93 | root = tmp_path |
| 94 | base = _commit_tasks(plugin, root, ["Buy milk"]) |
| 95 | target = _commit_tasks(plugin, root, ["Buy milk", "Fix the bug"]) |
| 96 | delta = plugin.diff(base, target, repo_root=root) |
| 97 | child_ops = delta["ops"][0]["child_ops"] |
| 98 | assert len(child_ops) == 1 |
| 99 | assert child_ops[0]["op"] == "insert" |
| 100 | assert "Fix the bug" in child_ops[0]["content_summary"] |
| 101 | |
| 102 | def test_added_and_removed_tasks_both_reported(self, tmp_path: pathlib.Path) -> None: |
| 103 | plugin = TodoPlugin() |
| 104 | root = tmp_path |
| 105 | base = _commit_tasks(plugin, root, ["Buy milk"]) |
| 106 | target = _commit_tasks(plugin, root, ["Fix the bug"]) |
| 107 | delta = plugin.diff(base, target, repo_root=root) |
| 108 | assert delta["summary"] == "1 task(s) added, 1 task(s) removed" |
| 109 | |
| 110 | |
| 111 | class TestMerge: |
| 112 | def test_two_independent_additions_never_conflict(self, tmp_path: pathlib.Path) -> None: |
| 113 | plugin = TodoPlugin() |
| 114 | root = tmp_path |
| 115 | base = _commit_tasks(plugin, root, ["Write episode 06 script"]) |
| 116 | left = _commit_tasks(plugin, root, ["Write episode 06 script", "Buy milk"]) |
| 117 | right = _commit_tasks(plugin, root, ["Write episode 06 script", "Fix the bug"]) |
| 118 | |
| 119 | result = plugin.merge(base, left, right, repo_root=root) |
| 120 | |
| 121 | assert result.conflicts == [] |
| 122 | merged_raw = read_object(root, result.merged["files"][_TASK_FILE]) |
| 123 | merged_tasks = set(merged_raw.decode("utf-8").splitlines()) |
| 124 | assert merged_tasks == {"Write episode 06 script", "Buy milk", "Fix the bug"} |
| 125 | |
| 126 | def test_task_removed_on_one_side_stays_removed(self, tmp_path: pathlib.Path) -> None: |
| 127 | plugin = TodoPlugin() |
| 128 | root = tmp_path |
| 129 | base = _commit_tasks(plugin, root, ["Buy milk", "Fix the bug"]) |
| 130 | left = _commit_tasks(plugin, root, ["Fix the bug"]) # removed "Buy milk" |
| 131 | right = _commit_tasks(plugin, root, ["Buy milk", "Fix the bug"]) # unchanged |
| 132 | |
| 133 | result = plugin.merge(base, left, right, repo_root=root) |
| 134 | |
| 135 | merged_raw = read_object(root, result.merged["files"][_TASK_FILE]) |
| 136 | merged_tasks = set(merged_raw.decode("utf-8").splitlines()) |
| 137 | assert merged_tasks == {"Fix the bug"} |
| 138 | |
| 139 | |
| 140 | class TestDrift: |
| 141 | def test_uncommitted_change_is_detected(self, tmp_path: pathlib.Path) -> None: |
| 142 | """Regression: drift must see live, uncommitted file content even |
| 143 | though snapshot() never writes the live file's blob to the object |
| 144 | store -- only `commit` does that. Reading back via read_object alone |
| 145 | returned None for anything not yet committed, so drift saw "0 added, |
| 146 | N removed" for any dirty working tree, regardless of what actually |
| 147 | changed. |
| 148 | """ |
| 149 | plugin = TodoPlugin() |
| 150 | root = tmp_path |
| 151 | committed = _commit_tasks(plugin, root, ["Buy milk"]) |
| 152 | (root / _TASK_FILE).write_text("Buy milk\nShip it\n") # uncommitted |
| 153 | |
| 154 | report = plugin.drift(committed, root) |
| 155 | |
| 156 | assert report.has_drift is True |
| 157 | assert report.summary == "1 task(s) added, 0 task(s) removed" |
| 158 | |
| 159 | def test_clean_tree_has_no_drift(self, tmp_path: pathlib.Path) -> None: |
| 160 | plugin = TodoPlugin() |
| 161 | root = tmp_path |
| 162 | committed = _commit_tasks(plugin, root, ["Buy milk"]) |
| 163 | report = plugin.drift(committed, root) |
| 164 | assert report.has_drift is False |
File History
1 commit
sha256:832d1ca80cb25129c91d8c8c190ec30af2e638a3e8160431f4704fd881bceaee
feat: add the todo domain plugin -- Build With Muse Episode 06
Sonnet 5
patch
3 hours ago