gabriel / muse public
test_core_patch_record.py python
405 lines 13.8 KB
Raw
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
1 """Unit tests for ``muse.core.patch_record`` — content-addressed Muse patch objects.
2
3 Test tiers
4 ----------
5 - Unit: PatchRecord dataclass, compute_patch_id, serialize/deserialize round-trip
6 - Data integrity: patch_id is stable, deterministic, and changes with content
7 - Security: patch_id forgery, tampered fields detected on re-verify
8 - Edge: empty diff, initial commit (no parent), binary objects skipped gracefully
9 """
10 from __future__ import annotations
11
12 import hashlib
13 import json
14 import pathlib
15
16 import pytest
17
18 from muse.core.patch_record import (
19 PatchRecord,
20 compute_patch_id,
21 deserialize_patch,
22 serialize_patch,
23 )
24 from muse.core.ids import hash_snapshot as compute_snapshot_id
25 from muse.core.commits import (
26 CommitRecord,
27 write_commit,
28 )
29 from muse.core.snapshots import (
30 SnapshotRecord,
31 write_snapshot,
32 )
33 from muse.core.object_store import write_object
34
35 import datetime
36 from muse.core.types import long_id, blob_id
37 from muse.core.paths import muse_dir
38
39
40 # ---------------------------------------------------------------------------
41 # Helpers
42 # ---------------------------------------------------------------------------
43
44
45 def _init_repo(path: pathlib.Path) -> pathlib.Path:
46 dot_muse = muse_dir(path)
47 for sub in ("commits", "snapshots", "objects", "refs/heads"):
48 (dot_muse / sub).mkdir(parents=True, exist_ok=True)
49 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
50 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test", "domain": "code"}))
51 return path
52
53
54 def _make_object(repo: pathlib.Path, content: bytes) -> str:
55 """Write bytes to object store; return sha256:<hex> prefixed ID."""
56 oid = blob_id(content)
57 write_object(repo, oid, content)
58 return oid
59
60
61 def _ts() -> datetime.datetime:
62 return datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
63
64
65 # ---------------------------------------------------------------------------
66 # compute_patch_id
67 # ---------------------------------------------------------------------------
68
69
70 class TestComputePatchId:
71 def test_returns_sha256_prefixed_string(self, tmp_path: pathlib.Path) -> None:
72 repo = _init_repo(tmp_path)
73 rec = PatchRecord(
74 patch_id="",
75 from_snapshot_id=long_id("a" * 64),
76 to_snapshot_id=long_id("b" * 64),
77 from_commit_id=long_id("c" * 64),
78 to_commit_id=long_id("d" * 64),
79 domain="code",
80 format_version="1.0",
81 created_at="2026-01-01T00:00:00+00:00",
82 agent_id="",
83 model_id="",
84 signer_public_key="",
85 signature="",
86 intent="",
87 sem_ver_bump="patch",
88 breaking_changes=[],
89 summary="test",
90 ops=[],
91 files_added=[],
92 files_modified=[],
93 files_deleted=[],
94 files_renamed={},
95 required_objects=[],
96 from_manifest={},
97 to_manifest={},
98 applicability={
99 "requires_snapshot": long_id("a" * 64),
100 "independent_dimensions": [],
101 "conflict_free": True,
102 },
103 blobs={},
104 )
105 pid = compute_patch_id(rec)
106 assert pid.startswith("sha256:")
107 assert len(pid) == 71 # sha256: (7) + 64 hex
108
109 def test_deterministic_across_calls(self, tmp_path: pathlib.Path) -> None:
110 repo = _init_repo(tmp_path)
111 rec = PatchRecord(
112 patch_id="",
113 from_snapshot_id=long_id("a" * 64),
114 to_snapshot_id=long_id("b" * 64),
115 from_commit_id=long_id("c" * 64),
116 to_commit_id=long_id("d" * 64),
117 domain="code",
118 format_version="1.0",
119 created_at="2026-01-01T00:00:00+00:00",
120 agent_id="test-agent",
121 model_id="claude-sonnet-4-6",
122 signer_public_key="",
123 signature="",
124 intent="test intent",
125 sem_ver_bump="minor",
126 breaking_changes=[],
127 summary="2 modified files",
128 ops=[],
129 files_added=["new.py"],
130 files_modified=[],
131 files_deleted=[],
132 files_renamed={},
133 required_objects=[],
134 from_manifest={},
135 to_manifest={},
136 applicability={
137 "requires_snapshot": long_id("a" * 64),
138 "independent_dimensions": ["symbols"],
139 "conflict_free": True,
140 },
141 blobs={},
142 )
143 pid1 = compute_patch_id(rec)
144 pid2 = compute_patch_id(rec)
145 assert pid1 == pid2
146
147 def test_changes_with_different_content(self, tmp_path: pathlib.Path) -> None:
148 base = dict(
149 patch_id="",
150 from_snapshot_id=long_id("a" * 64),
151 to_snapshot_id=long_id("b" * 64),
152 from_commit_id=long_id("c" * 64),
153 to_commit_id=long_id("d" * 64),
154 domain="code",
155 format_version="1.0",
156 created_at="2026-01-01T00:00:00+00:00",
157 agent_id="",
158 model_id="",
159 signer_public_key="",
160 signature="",
161 intent="",
162 sem_ver_bump="patch",
163 breaking_changes=[],
164 summary="v1",
165 ops=[],
166 files_added=[],
167 files_modified=[],
168 files_deleted=[],
169 files_renamed={},
170 required_objects=[],
171 from_manifest={},
172 to_manifest={},
173 applicability={"requires_snapshot": long_id("a" * 64), "independent_dimensions": [], "conflict_free": True},
174 )
175 r1 = PatchRecord(**base)
176 r2 = PatchRecord(**{**base, "summary": "v2"})
177 assert compute_patch_id(r1) != compute_patch_id(r2)
178
179 def test_patch_id_excludes_signature_field(self, tmp_path: pathlib.Path) -> None:
180 """Signature must not influence patch_id (it signs the id, not the other way)."""
181 base = dict(
182 patch_id="",
183 from_snapshot_id=long_id("a" * 64),
184 to_snapshot_id=long_id("b" * 64),
185 from_commit_id=long_id("c" * 64),
186 to_commit_id=long_id("d" * 64),
187 domain="code",
188 format_version="1.0",
189 created_at="2026-01-01T00:00:00+00:00",
190 agent_id="",
191 model_id="",
192 signer_public_key="",
193 signature="",
194 intent="",
195 sem_ver_bump="patch",
196 breaking_changes=[],
197 summary="test",
198 ops=[],
199 files_added=[],
200 files_modified=[],
201 files_deleted=[],
202 files_renamed={},
203 required_objects=[],
204 from_manifest={},
205 to_manifest={},
206 applicability={"requires_snapshot": long_id("a" * 64), "independent_dimensions": [], "conflict_free": True},
207 )
208 r_no_sig = PatchRecord(**base)
209 r_with_sig = PatchRecord(**{**base, "signature": "abc123", "signer_public_key": "pubkey"})
210 assert compute_patch_id(r_no_sig) == compute_patch_id(r_with_sig)
211
212
213 # ---------------------------------------------------------------------------
214 # Serialization round-trip
215 # ---------------------------------------------------------------------------
216
217
218 class TestSerializeDeserialize:
219 def _make_record(self) -> PatchRecord:
220 rec = PatchRecord(
221 patch_id="",
222 from_snapshot_id=long_id("a" * 64),
223 to_snapshot_id=long_id("b" * 64),
224 from_commit_id=long_id("c" * 64),
225 to_commit_id=long_id("d" * 64),
226 domain="code",
227 format_version="1.0",
228 created_at="2026-01-01T00:00:00+00:00",
229 agent_id="claude-code",
230 model_id="claude-sonnet-4-6",
231 signer_public_key="",
232 signature="",
233 intent="improve merge logic",
234 sem_ver_bump="minor",
235 breaking_changes=[],
236 summary="1 modified file",
237 ops=[{"op": "insert", "address": "main.py", "position": 0, "content_id": long_id("e" * 64), "content_summary": "new file", "action_label": "inserted"}],
238 files_added=["main.py"],
239 files_modified=[],
240 files_deleted=[],
241 files_renamed={},
242 required_objects=[long_id("e" * 64)],
243 from_manifest={},
244 to_manifest={"main.py": long_id("e" * 64)},
245 applicability={
246 "requires_snapshot": long_id("a" * 64),
247 "independent_dimensions": ["symbols", "imports"],
248 "conflict_free": True,
249 },
250 blobs={},
251 )
252 rec.patch_id = compute_patch_id(rec)
253 return rec
254
255 def test_serialize_returns_bytes(self) -> None:
256 rec = self._make_record()
257 data = serialize_patch(rec)
258 assert isinstance(data, bytes)
259
260 def test_deserialize_round_trip(self) -> None:
261 rec = self._make_record()
262 data = serialize_patch(rec)
263 rec2 = deserialize_patch(data)
264 assert rec2.patch_id == rec.patch_id
265 assert rec2.domain == rec.domain
266 assert rec2.summary == rec.summary
267 assert rec2.ops == rec.ops
268 assert rec2.files_added == rec.files_added
269 assert rec2.from_manifest == rec.from_manifest
270 assert rec2.to_manifest == rec.to_manifest
271
272 def test_serialized_is_valid_json(self) -> None:
273 rec = self._make_record()
274 data = serialize_patch(rec)
275 parsed = json.loads(data)
276 assert "patch_id" in parsed
277 assert "domain" in parsed
278
279 def test_patch_id_preserved_through_round_trip(self) -> None:
280 rec = self._make_record()
281 data = serialize_patch(rec)
282 rec2 = deserialize_patch(data)
283 assert rec2.patch_id == rec.patch_id
284
285 def test_deserialize_rejects_garbage(self) -> None:
286 with pytest.raises(Exception):
287 deserialize_patch(b"not valid json at all !!!!")
288
289 def test_deserialize_rejects_missing_patch_id(self) -> None:
290 data = json.dumps({"domain": "code"}).encode()
291 with pytest.raises(Exception):
292 deserialize_patch(data)
293
294
295 # ---------------------------------------------------------------------------
296 # PatchRecord dataclass
297 # ---------------------------------------------------------------------------
298
299
300 class TestPatchRecord:
301 def test_has_required_fields(self) -> None:
302 rec = PatchRecord(
303 patch_id=long_id("a" * 64),
304 from_snapshot_id=long_id("b" * 64),
305 to_snapshot_id=long_id("c" * 64),
306 from_commit_id=long_id("d" * 64),
307 to_commit_id=long_id("e" * 64),
308 domain="code",
309 format_version="1.0",
310 created_at="2026-01-01T00:00:00+00:00",
311 agent_id="",
312 model_id="",
313 signer_public_key="",
314 signature="",
315 intent="",
316 sem_ver_bump="patch",
317 breaking_changes=[],
318 summary="",
319 ops=[],
320 files_added=[],
321 files_modified=[],
322 files_deleted=[],
323 files_renamed={},
324 required_objects=[],
325 from_manifest={},
326 to_manifest={},
327 applicability={"requires_snapshot": long_id("b" * 64), "independent_dimensions": [], "conflict_free": True},
328 )
329 assert rec.domain == "code"
330 assert rec.format_version == "1.0"
331 assert rec.sem_ver_bump == "patch"
332
333 def test_ops_with_action_label(self) -> None:
334 """Each op can carry an action_label — Cohen-transform extension."""
335 op = {
336 "op": "insert",
337 "address": "foo.py",
338 "position": 0,
339 "content_id": long_id("a" * 64),
340 "content_summary": "new function",
341 "action_label": "inserted",
342 }
343 rec = PatchRecord(
344 patch_id="",
345 from_snapshot_id=long_id("a" * 64),
346 to_snapshot_id=long_id("b" * 64),
347 from_commit_id=long_id("c" * 64),
348 to_commit_id=long_id("d" * 64),
349 domain="code",
350 format_version="1.0",
351 created_at="2026-01-01T00:00:00+00:00",
352 agent_id="",
353 model_id="",
354 signer_public_key="",
355 signature="",
356 intent="",
357 sem_ver_bump="patch",
358 breaking_changes=[],
359 summary="",
360 ops=[op],
361 files_added=[],
362 files_modified=[],
363 files_deleted=[],
364 files_renamed={},
365 required_objects=[],
366 from_manifest={},
367 to_manifest={},
368 applicability={"requires_snapshot": long_id("a" * 64), "independent_dimensions": [], "conflict_free": True},
369 )
370 assert rec.ops[0]["action_label"] == "inserted"
371
372 def test_applicability_has_requires_snapshot(self) -> None:
373 rec = PatchRecord(
374 patch_id="",
375 from_snapshot_id=long_id("a" * 64),
376 to_snapshot_id=long_id("b" * 64),
377 from_commit_id=long_id("c" * 64),
378 to_commit_id=long_id("d" * 64),
379 domain="code",
380 format_version="1.0",
381 created_at="2026-01-01T00:00:00+00:00",
382 agent_id="",
383 model_id="",
384 signer_public_key="",
385 signature="",
386 intent="",
387 sem_ver_bump="patch",
388 breaking_changes=[],
389 summary="",
390 ops=[],
391 files_added=[],
392 files_modified=[],
393 files_deleted=[],
394 files_renamed={},
395 required_objects=[],
396 from_manifest={},
397 to_manifest={},
398 applicability={
399 "requires_snapshot": long_id("a" * 64),
400 "independent_dimensions": ["symbols"],
401 "conflict_free": False,
402 },
403 )
404 assert rec.applicability["requires_snapshot"] == long_id("a" * 64)
405 assert rec.applicability["conflict_free"] is False
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 22 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 28 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 29 days ago