test_op_log.py
python
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2
fix: remove commit_exists filter from have anchors — server…
Sonnet 4.6
patch
21 days ago
| 1 | """Tests for muse.core.op_log — OpEntry, OpLogCheckpoint, OpLog.""" |
| 2 | |
| 3 | import pathlib |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | from muse.core.op_log import ( |
| 8 | OpEntry, |
| 9 | OpLog, |
| 10 | list_sessions, |
| 11 | make_op_entry, |
| 12 | ) |
| 13 | from muse.domain import InsertOp |
| 14 | |
| 15 | |
| 16 | # --------------------------------------------------------------------------- |
| 17 | # make_op_entry factory |
| 18 | # --------------------------------------------------------------------------- |
| 19 | |
| 20 | |
| 21 | class TestMakeOpEntry: |
| 22 | def test_all_required_fields_present(self) -> None: |
| 23 | op = InsertOp( |
| 24 | op="insert", |
| 25 | address="note:0", |
| 26 | position=0, |
| 27 | content_id="abc123", |
| 28 | content_summary="C4", |
| 29 | ) |
| 30 | entry = make_op_entry( |
| 31 | actor_id="agent-x", |
| 32 | domain="midi", |
| 33 | domain_op=op, |
| 34 | lamport_ts=1, |
| 35 | ) |
| 36 | assert entry["actor_id"] == "agent-x" |
| 37 | assert entry["domain"] == "midi" |
| 38 | assert entry["lamport_ts"] == 1 |
| 39 | assert entry["parent_op_ids"] == [] |
| 40 | assert entry["intent_id"] == "" |
| 41 | assert entry["reservation_id"] == "" |
| 42 | assert entry["op_id"].startswith("sha256:") # content-addressed |
| 43 | assert len(entry["op_id"]) == 71 |
| 44 | |
| 45 | def test_parent_op_ids_are_copied(self) -> None: |
| 46 | op = InsertOp(op="insert", address="note:0", position=0, content_id="x", content_summary="") |
| 47 | parent_ids = ["aaa", "bbb"] |
| 48 | entry = make_op_entry("a", "midi", op, 1, parent_op_ids=parent_ids) |
| 49 | assert entry["parent_op_ids"] == ["aaa", "bbb"] |
| 50 | # Mutating the original should not affect the entry. |
| 51 | parent_ids.append("ccc") |
| 52 | assert entry["parent_op_ids"] == ["aaa", "bbb"] |
| 53 | |
| 54 | def test_op_ids_are_unique(self) -> None: |
| 55 | op = InsertOp(op="insert", address="note:0", position=0, content_id="x", content_summary="") |
| 56 | ids = {make_op_entry("a", "midi", op, i)["op_id"] for i in range(20)} |
| 57 | assert len(ids) == 20 |
| 58 | |
| 59 | |
| 60 | # --------------------------------------------------------------------------- |
| 61 | # OpLog.append and read_all |
| 62 | # --------------------------------------------------------------------------- |
| 63 | |
| 64 | |
| 65 | class TestOpLogAppendRead: |
| 66 | def test_append_and_read_all_roundtrip(self, tmp_path: pathlib.Path) -> None: |
| 67 | log = OpLog(tmp_path, "session-1") |
| 68 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c1", content_summary="C4") |
| 69 | e1 = make_op_entry("agent-a", "midi", op, 1) |
| 70 | e2 = make_op_entry("agent-a", "midi", op, 2) |
| 71 | log.append(e1) |
| 72 | log.append(e2) |
| 73 | entries = log.read_all() |
| 74 | assert len(entries) == 2 |
| 75 | assert entries[0]["op_id"] == e1["op_id"] |
| 76 | assert entries[1]["op_id"] == e2["op_id"] |
| 77 | |
| 78 | def test_empty_log_returns_empty_list(self, tmp_path: pathlib.Path) -> None: |
| 79 | log = OpLog(tmp_path, "empty-session") |
| 80 | assert log.read_all() == [] |
| 81 | |
| 82 | def test_append_creates_directory(self, tmp_path: pathlib.Path) -> None: |
| 83 | log = OpLog(tmp_path, "new-session") |
| 84 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c1", content_summary="") |
| 85 | log.append(make_op_entry("a", "midi", op, 1)) |
| 86 | assert (tmp_path / ".muse" / "op_log" / "new-session").is_dir() |
| 87 | |
| 88 | |
| 89 | # --------------------------------------------------------------------------- |
| 90 | # Lamport timestamp counter |
| 91 | # --------------------------------------------------------------------------- |
| 92 | |
| 93 | |
| 94 | class TestLamportTs: |
| 95 | def test_lamport_is_monotonic(self, tmp_path: pathlib.Path) -> None: |
| 96 | log = OpLog(tmp_path, "ts-session") |
| 97 | ts_values = [log.next_lamport_ts() for _ in range(10)] |
| 98 | assert ts_values == sorted(ts_values) |
| 99 | assert len(set(ts_values)) == 10 |
| 100 | |
| 101 | def test_lamport_continues_after_reopen(self, tmp_path: pathlib.Path) -> None: |
| 102 | log1 = OpLog(tmp_path, "reopen-session") |
| 103 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="") |
| 104 | for i in range(5): |
| 105 | ts = log1.next_lamport_ts() |
| 106 | log1.append(make_op_entry("a", "midi", op, ts)) |
| 107 | |
| 108 | # Reopen the same session. |
| 109 | log2 = OpLog(tmp_path, "reopen-session") |
| 110 | new_ts = log2.next_lamport_ts() |
| 111 | assert new_ts > 5 |
| 112 | |
| 113 | |
| 114 | # --------------------------------------------------------------------------- |
| 115 | # Checkpoint |
| 116 | # --------------------------------------------------------------------------- |
| 117 | |
| 118 | |
| 119 | class TestCheckpoint: |
| 120 | def test_checkpoint_written_and_readable(self, tmp_path: pathlib.Path) -> None: |
| 121 | log = OpLog(tmp_path, "ckpt-session") |
| 122 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="") |
| 123 | for i in range(3): |
| 124 | log.append(make_op_entry("a", "midi", op, i + 1)) |
| 125 | |
| 126 | ckpt = log.checkpoint("snap-abc") |
| 127 | assert ckpt["snapshot_id"] == "snap-abc" |
| 128 | assert ckpt["op_count"] == 3 |
| 129 | assert ckpt["lamport_ts"] == 3 |
| 130 | |
| 131 | recovered = log.read_checkpoint() |
| 132 | assert recovered is not None |
| 133 | assert recovered["snapshot_id"] == "snap-abc" |
| 134 | |
| 135 | def test_no_checkpoint_returns_none(self, tmp_path: pathlib.Path) -> None: |
| 136 | log = OpLog(tmp_path, "no-ckpt-session") |
| 137 | assert log.read_checkpoint() is None |
| 138 | |
| 139 | def test_replay_since_checkpoint_returns_newer_only(self, tmp_path: pathlib.Path) -> None: |
| 140 | log = OpLog(tmp_path, "replay-session") |
| 141 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="") |
| 142 | |
| 143 | for i in range(3): |
| 144 | log.append(make_op_entry("a", "midi", op, i + 1)) |
| 145 | log.checkpoint("snap-1") |
| 146 | |
| 147 | # Add more entries after checkpoint. |
| 148 | for i in range(3, 6): |
| 149 | log.append(make_op_entry("a", "midi", op, i + 1)) |
| 150 | |
| 151 | entries = log.replay_since_checkpoint() |
| 152 | assert len(entries) == 3 |
| 153 | assert all(e["lamport_ts"] > 3 for e in entries) |
| 154 | |
| 155 | |
| 156 | # --------------------------------------------------------------------------- |
| 157 | # to_structured_delta |
| 158 | # --------------------------------------------------------------------------- |
| 159 | |
| 160 | |
| 161 | class TestToStructuredDelta: |
| 162 | def test_produces_correct_domain_ops_filtered_by_domain(self, tmp_path: pathlib.Path) -> None: |
| 163 | log = OpLog(tmp_path, "delta-session") |
| 164 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="C4") |
| 165 | |
| 166 | for i in range(4): |
| 167 | log.append(make_op_entry("a", "midi", op, i + 1)) |
| 168 | # Add one code op that should be filtered out. |
| 169 | code_op = InsertOp(op="insert", address="sym:0", position=0, content_id="d", content_summary="f()") |
| 170 | log.append(make_op_entry("a", "code", code_op, 5)) |
| 171 | |
| 172 | delta = log.to_structured_delta("midi") |
| 173 | assert delta["domain"] == "midi_notes_tracked" or delta["domain"] == "midi" |
| 174 | # Only the 4 music ops should be included. |
| 175 | assert len(delta["ops"]) == 4 |
| 176 | |
| 177 | def test_summary_mentions_insert(self, tmp_path: pathlib.Path) -> None: |
| 178 | log = OpLog(tmp_path, "summary-session") |
| 179 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="C4") |
| 180 | log.append(make_op_entry("a", "midi", op, 1)) |
| 181 | delta = log.to_structured_delta("midi") |
| 182 | assert "insert" in delta["summary"] |
| 183 | |
| 184 | |
| 185 | # --------------------------------------------------------------------------- |
| 186 | # Session listing |
| 187 | # --------------------------------------------------------------------------- |
| 188 | |
| 189 | |
| 190 | class TestListSessions: |
| 191 | def test_lists_all_sessions(self, tmp_path: pathlib.Path) -> None: |
| 192 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="") |
| 193 | for sid in ["alpha", "beta", "gamma"]: |
| 194 | log = OpLog(tmp_path, sid) |
| 195 | log.append(make_op_entry("a", "midi", op, 1)) |
| 196 | |
| 197 | sessions = list_sessions(tmp_path) |
| 198 | assert "alpha" in sessions |
| 199 | assert "beta" in sessions |
| 200 | assert "gamma" in sessions |
| 201 | |
| 202 | def test_empty_repo_returns_empty_list(self, tmp_path: pathlib.Path) -> None: |
| 203 | assert list_sessions(tmp_path) == [] |
| 204 | |
| 205 | |
| 206 | # --------------------------------------------------------------------------- |
| 207 | # Content-addressed op_id |
| 208 | # --------------------------------------------------------------------------- |
| 209 | |
| 210 | |
| 211 | class TestOpIdContentAddressed: |
| 212 | """op_id must be sha256: of canonical op content, not a UUID.""" |
| 213 | |
| 214 | def test_op_id_is_sha256_prefixed(self, tmp_path: pathlib.Path) -> None: |
| 215 | from muse.core.types import long_id |
| 216 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="C4") |
| 217 | entry = make_op_entry("actor-1", "midi", op, lamport_ts=1) |
| 218 | assert entry["op_id"].startswith("sha256:"), f"Expected sha256: prefix, got {entry['op_id']!r}" |
| 219 | assert len(entry["op_id"]) == 71 |
| 220 | |
| 221 | def test_op_id_is_sha256_not_uuid4(self, tmp_path: pathlib.Path) -> None: |
| 222 | import re |
| 223 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="C4") |
| 224 | entry = make_op_entry("actor-1", "midi", op, lamport_ts=1) |
| 225 | uuid4_re = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$") |
| 226 | assert not uuid4_re.match(entry["op_id"]) |
| 227 | |
| 228 | def test_op_id_deterministic(self, tmp_path: pathlib.Path) -> None: |
| 229 | """Same op content → same op_id.""" |
| 230 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="C4") |
| 231 | e1 = make_op_entry("actor-1", "midi", op, lamport_ts=5, parent_op_ids=["x"]) |
| 232 | e2 = make_op_entry("actor-1", "midi", op, lamport_ts=5, parent_op_ids=["x"]) |
| 233 | assert e1["op_id"] == e2["op_id"] |
| 234 | |
| 235 | def test_op_id_differs_by_actor(self, tmp_path: pathlib.Path) -> None: |
| 236 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="C4") |
| 237 | e1 = make_op_entry("actor-1", "midi", op, lamport_ts=1) |
| 238 | e2 = make_op_entry("actor-2", "midi", op, lamport_ts=1) |
| 239 | assert e1["op_id"] != e2["op_id"] |
| 240 | |
| 241 | def test_op_id_differs_by_lamport_ts(self, tmp_path: pathlib.Path) -> None: |
| 242 | op = InsertOp(op="insert", address="note:0", position=0, content_id="c", content_summary="C4") |
| 243 | e1 = make_op_entry("actor-1", "midi", op, lamport_ts=1) |
| 244 | e2 = make_op_entry("actor-1", "midi", op, lamport_ts=2) |
| 245 | assert e1["op_id"] != e2["op_id"] |
File History
4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2
fix: remove commit_exists filter from have anchors — server…
Sonnet 4.6
patch
21 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e
fix: rename objects→blobs in push client and all stale test…
Sonnet 4.6
patch
23 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a
fix: repair four test failures from post-migration audit
Sonnet 4.6
patch
29 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf
fix: unified object store migration — idempotent writes, JS…
Sonnet 4.6
minor
⚠
29 days ago