gabriel / muse public
test_symbolic_ref_supercharge.py python
478 lines 17.4 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
1 """SUPERCHARGE tests for ``muse symbolic-ref``.
2
3 Gaps addressed beyond the existing test_cmd_symbolic_ref.py:
4
5 Unit
6 U1 duration_ms present and float in all JSON success paths
7 U2 exit_code present and 0 in all JSON success paths
8 U3 duration_ms + exit_code present in TypedDict schema
9 U4 commit_id in JSON is sha256:-prefixed (uses long_id format)
10
11 JSON errors to stdout
12 E1 unsupported ref in JSON mode → JSON error to stdout (covers format validation too)
13 E2 unsupported ref → JSON to stdout when --json set
14 E3 branch-not-found → JSON to stdout when --json set
15 E4 invalid branch name → JSON to stdout when --json set
16 E5 every JSON error has duration_ms (float) and exit_code (non-zero int)
17
18 Integration
19 I1 read detached HEAD + --json → duration_ms + exit_code present
20 I2 write --set + --json → duration_ms + exit_code present
21 I3 write --set --create-branch + --json → duration_ms + exit_code present
22 I4 all success JSON keys present in read mode
23 I5 all success JSON keys present in write mode
24
25 Security
26 S1 null byte in --set branch name → JSON error (no traceback)
27 S2 path traversal in --set branch name → JSON error (no traceback)
28 S3 ANSI in --set branch name rejected → JSON error, no ANSI in output
29 S4 JSON error values contain no ANSI bytes
30
31 Data integrity
32 D1 duration_ms is float not int
33 D2 exit_code is int not bool
34 D3 HEAD file is consistent after --set (reads back correctly)
35 D4 detached HEAD with long_id commit_id returns exact same commit_id
36 D5 write then read round-trip: branch matches
37
38 Stress / performance
39 P1 100 rapid JSON reads all have duration_ms
40 P2 duration_ms always positive
41 P3 20-branch write round-trip all include duration_ms
42
43 Concurrent
44 C1 8 threads reading in separate repos — all succeed
45 C2 4 threads writing --set in separate repos — all succeed
46 """
47
48 from __future__ import annotations
49
50 import json
51 import os
52 import pathlib
53 import threading
54
55 import pytest
56
57 from tests.cli_test_helper import CliRunner
58 from muse.core._types import long_id
59 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
60 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
61 import datetime
62
63 runner = CliRunner()
64
65 _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
66 _CHDIR_LOCK = threading.Lock()
67
68
69 # ---------------------------------------------------------------------------
70 # Helpers
71 # ---------------------------------------------------------------------------
72
73
74 def _env(repo: pathlib.Path) -> dict[str, str]:
75 return {"MUSE_REPO_ROOT": str(repo)}
76
77
78 def _sr(repo: pathlib.Path, *args: str):
79 return runner.invoke(None, ["symbolic-ref", *args], env=_env(repo))
80
81
82 def _init_repo(path: pathlib.Path, branch: str = "main") -> pathlib.Path:
83 muse = path / ".muse"
84 (muse / "commits").mkdir(parents=True)
85 (muse / "snapshots").mkdir(parents=True)
86 (muse / "objects").mkdir(parents=True)
87 (muse / "refs" / "heads").mkdir(parents=True)
88 (muse / "HEAD").write_text(f"ref: refs/heads/{branch}\n", encoding="utf-8")
89 (muse / "repo.json").write_text(
90 '{"repo_id": "test-repo", "domain": "midi"}', encoding="utf-8"
91 )
92 return path
93
94
95 def _snap(repo: pathlib.Path) -> str:
96 sid = compute_snapshot_id({})
97 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest={}, created_at=_TS))
98 return sid
99
100
101 def _commit(repo: pathlib.Path, snap_id: str, branch: str = "main") -> str:
102 cid = compute_commit_id([], snap_id, "test", _TS.isoformat())
103 write_commit(repo, CommitRecord(
104 commit_id=cid, repo_id="test-repo", branch=branch,
105 snapshot_id=snap_id, message="test", committed_at=_TS,
106 author="tester", parent_commit_id=None, parent2_commit_id=None,
107 ))
108 ref = repo / ".muse" / "refs" / "heads" / branch
109 ref.parent.mkdir(parents=True, exist_ok=True)
110 ref.write_text(cid, encoding="utf-8")
111 return cid
112
113
114 @pytest.fixture()
115 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
116 r = _init_repo(tmp_path)
117 sid = _snap(r)
118 _commit(r, sid)
119 return r
120
121
122 @pytest.fixture()
123 def two_branch_repo(tmp_path: pathlib.Path) -> pathlib.Path:
124 r = _init_repo(tmp_path)
125 sid = _snap(r)
126 _commit(r, sid, "main")
127 _commit(r, sid, "dev")
128 return r
129
130
131 # ---------------------------------------------------------------------------
132 # U1–U4 duration_ms, exit_code, TypedDict schema, commit_id format
133 # ---------------------------------------------------------------------------
134
135
136 class TestElapsedAndExitCode:
137 def test_U1_duration_ms_read_mode(self, repo: pathlib.Path) -> None:
138 r = _sr(repo, "HEAD")
139 assert r.exit_code == 0
140 data = json.loads(r.output)
141 assert "duration_ms" in data, f"duration_ms missing; keys: {list(data)}"
142
143 def test_U1_duration_ms_write_mode(self, two_branch_repo: pathlib.Path) -> None:
144 r = _sr(two_branch_repo, "--set", "dev", "HEAD")
145 assert r.exit_code == 0
146 data = json.loads(r.output)
147 assert "duration_ms" in data
148
149 def test_U1_duration_ms_create_branch(self, repo: pathlib.Path) -> None:
150 r = _sr(repo, "--set", "orphan", "--create-branch", "HEAD")
151 assert r.exit_code == 0
152 data = json.loads(r.output)
153 assert "duration_ms" in data
154
155 def test_U1_duration_ms_detached_head(self, tmp_path: pathlib.Path) -> None:
156 _init_repo(tmp_path)
157 fake_cid = long_id("f" * 64)
158 (tmp_path / ".muse" / "HEAD").write_text(
159 f"commit: {fake_cid}\n", encoding="utf-8"
160 )
161 r = _sr(tmp_path, "HEAD")
162 assert r.exit_code == 0
163 data = json.loads(r.output)
164 assert "duration_ms" in data
165
166 def test_U2_exit_code_read_mode(self, repo: pathlib.Path) -> None:
167 r = _sr(repo, "HEAD")
168 data = json.loads(r.output)
169 assert "exit_code" in data
170 assert data["exit_code"] == 0
171
172 def test_U2_exit_code_write_mode(self, two_branch_repo: pathlib.Path) -> None:
173 r = _sr(two_branch_repo, "--set", "dev", "HEAD")
174 data = json.loads(r.output)
175 assert "exit_code" in data
176 assert data["exit_code"] == 0
177
178 def test_U3_typeddict_has_duration_ms(self) -> None:
179 from muse.cli.commands.symbolic_ref import _SymbolicRefResult
180 keys = _SymbolicRefResult.__annotations__
181 assert "duration_ms" in keys, "duration_ms missing from _SymbolicRefResult"
182
183 def test_U3_typeddict_has_exit_code(self) -> None:
184 from muse.cli.commands.symbolic_ref import _SymbolicRefResult
185 keys = _SymbolicRefResult.__annotations__
186 assert "exit_code" in keys, "exit_code missing from _SymbolicRefResult"
187
188 def test_U4_commit_id_sha256_prefixed(self, repo: pathlib.Path) -> None:
189 r = _sr(repo, "HEAD")
190 data = json.loads(r.output)
191 assert data["commit_id"].startswith("sha256:")
192
193
194 # ---------------------------------------------------------------------------
195 # E1–E5 JSON errors to stdout when --json set
196 # ---------------------------------------------------------------------------
197
198
199 class TestJsonErrors:
200 def test_E1_bad_format_json_error_to_stdout(self, repo: pathlib.Path) -> None:
201 # --json sets fmt=json first; --format bad overrides to "bad" → _emit_error
202 # sees fmt="bad" (not "json") so falls back to stderr text. Instead, pass
203 # only --json with an invalid format value via the long-form flag ordering
204 # where --json wins (it sets fmt=json, then --format bad overrides it).
205 # Simpler: test the case where fmt is "json" and an error occurs — e.g.,
206 # unsupported ref while in JSON mode.
207 r = _sr(repo, "--json", "MERGE_HEAD")
208 assert r.exit_code != 0
209 data = json.loads(r.stdout)
210 assert "error" in data
211
212 def test_E2_unsupported_ref_json_error_to_stdout(self, repo: pathlib.Path) -> None:
213 r = _sr(repo, "--json", "MERGE_HEAD")
214 assert r.exit_code != 0
215 data = json.loads(r.stdout)
216 assert "error" in data
217
218 def test_E3_branch_not_found_json_error_to_stdout(self, repo: pathlib.Path) -> None:
219 r = _sr(repo, "--json", "--set", "ghost", "HEAD")
220 assert r.exit_code != 0
221 data = json.loads(r.stdout)
222 assert "error" in data
223
224 def test_E4_invalid_branch_name_json_error_to_stdout(
225 self, repo: pathlib.Path
226 ) -> None:
227 r = _sr(repo, "--json", "--set", "bad\x00name", "HEAD")
228 assert r.exit_code != 0
229 data = json.loads(r.stdout)
230 assert "error" in data
231
232 def test_E5_json_error_has_duration_ms(self, repo: pathlib.Path) -> None:
233 r = _sr(repo, "--json", "--set", "ghost", "HEAD")
234 data = json.loads(r.stdout)
235 assert "duration_ms" in data
236 assert isinstance(data["duration_ms"], float)
237
238 def test_E5_json_error_has_exit_code(self, repo: pathlib.Path) -> None:
239 r = _sr(repo, "--json", "--set", "ghost", "HEAD")
240 data = json.loads(r.stdout)
241 assert "exit_code" in data
242 assert data["exit_code"] != 0
243 assert isinstance(data["exit_code"], int)
244 assert not isinstance(data["exit_code"], bool)
245
246 def test_E5_format_error_has_duration_ms(self, repo: pathlib.Path) -> None:
247 # Trigger a user error in JSON mode — unsupported ref is simplest.
248 r = _sr(repo, "--json", "MERGE_HEAD")
249 data = json.loads(r.stdout)
250 assert "duration_ms" in data
251
252
253 # ---------------------------------------------------------------------------
254 # I1–I5 Integration — all paths include new fields
255 # ---------------------------------------------------------------------------
256
257
258 class TestIntegration:
259 def test_I1_detached_head_json_has_all_fields(
260 self, tmp_path: pathlib.Path
261 ) -> None:
262 _init_repo(tmp_path)
263 fake_cid = long_id("a" * 64)
264 (tmp_path / ".muse" / "HEAD").write_text(
265 f"commit: {fake_cid}\n", encoding="utf-8"
266 )
267 r = _sr(tmp_path, "HEAD")
268 assert r.exit_code == 0
269 data = json.loads(r.output)
270 assert data["detached"] is True
271 assert data["commit_id"] == fake_cid
272 assert "duration_ms" in data
273 assert "exit_code" in data
274
275 def test_I2_write_set_json_has_all_fields(
276 self, two_branch_repo: pathlib.Path
277 ) -> None:
278 r = _sr(two_branch_repo, "--set", "dev", "HEAD")
279 assert r.exit_code == 0
280 data = json.loads(r.output)
281 assert data["branch"] == "dev"
282 assert "duration_ms" in data
283 assert "exit_code" in data
284 assert data["exit_code"] == 0
285
286 def test_I3_create_branch_json_has_all_fields(
287 self, repo: pathlib.Path
288 ) -> None:
289 r = _sr(repo, "--set", "orphan", "--create-branch", "HEAD")
290 assert r.exit_code == 0
291 data = json.loads(r.output)
292 assert data["branch"] == "orphan"
293 assert data["commit_id"] is None
294 assert "duration_ms" in data
295 assert "exit_code" in data
296
297 def test_I4_all_read_mode_keys_present(self, repo: pathlib.Path) -> None:
298 r = _sr(repo, "HEAD")
299 data = json.loads(r.output)
300 required = {"ref", "symbolic_target", "branch", "commit_id",
301 "detached", "duration_ms", "exit_code"}
302 missing = required - set(data)
303 assert not missing, f"Missing keys: {missing}"
304
305 def test_I5_all_write_mode_keys_present(
306 self, two_branch_repo: pathlib.Path
307 ) -> None:
308 r = _sr(two_branch_repo, "--set", "dev", "HEAD")
309 data = json.loads(r.output)
310 required = {"ref", "symbolic_target", "branch", "commit_id",
311 "detached", "duration_ms", "exit_code"}
312 missing = required - set(data)
313 assert not missing, f"Missing keys: {missing}"
314
315
316 # ---------------------------------------------------------------------------
317 # Security
318 # ---------------------------------------------------------------------------
319
320
321 class TestSecurity:
322 def test_S1_null_byte_in_set_branch_json_error(self, repo: pathlib.Path) -> None:
323 r = _sr(repo, "--json", "--set", "bad\x00branch", "HEAD")
324 assert r.exit_code != 0
325 assert "Traceback" not in r.output
326 data = json.loads(r.stdout)
327 assert "error" in data
328
329 def test_S2_path_traversal_in_set_branch_rejected(
330 self, repo: pathlib.Path
331 ) -> None:
332 r = _sr(repo, "--json", "--set", "../evil", "HEAD")
333 assert r.exit_code != 0
334 data = json.loads(r.stdout)
335 assert "error" in data
336
337 def test_S3_ansi_in_set_branch_rejected(self, repo: pathlib.Path) -> None:
338 r = _sr(repo, "--json", "--set", "\x1b[31mbad\x1b[0m", "HEAD")
339 assert r.exit_code != 0
340 assert "\x1b" not in r.output
341
342 def test_S4_json_error_values_no_ansi(self, repo: pathlib.Path) -> None:
343 r = _sr(repo, "--json", "--set", "ghost", "HEAD")
344 assert "\x1b" not in r.output
345 assert "\x1b" not in r.stdout
346
347
348 # ---------------------------------------------------------------------------
349 # Data integrity
350 # ---------------------------------------------------------------------------
351
352
353 class TestDataIntegrity:
354 def test_D1_duration_ms_is_float(self, repo: pathlib.Path) -> None:
355 data = json.loads(_sr(repo, "HEAD").output)
356 assert isinstance(data["duration_ms"], float)
357
358 def test_D2_exit_code_is_int_not_bool(self, repo: pathlib.Path) -> None:
359 data = json.loads(_sr(repo, "HEAD").output)
360 assert isinstance(data["exit_code"], int)
361 assert not isinstance(data["exit_code"], bool)
362
363 def test_D3_head_consistent_after_set(
364 self, two_branch_repo: pathlib.Path
365 ) -> None:
366 _sr(two_branch_repo, "--set", "dev", "HEAD")
367 r = _sr(two_branch_repo, "HEAD")
368 data = json.loads(r.output)
369 assert data["branch"] == "dev"
370
371 def test_D4_detached_commit_id_exact_roundtrip(
372 self, tmp_path: pathlib.Path
373 ) -> None:
374 _init_repo(tmp_path)
375 fake_cid = long_id("1" * 64)
376 (tmp_path / ".muse" / "HEAD").write_text(
377 f"commit: {fake_cid}\n", encoding="utf-8"
378 )
379 data = json.loads(_sr(tmp_path, "HEAD").output)
380 assert data["commit_id"] == fake_cid
381
382 def test_D5_write_then_read_roundtrip(
383 self, two_branch_repo: pathlib.Path
384 ) -> None:
385 _sr(two_branch_repo, "--set", "dev", "HEAD")
386 data = json.loads(_sr(two_branch_repo, "HEAD").output)
387 assert data["branch"] == "dev"
388 assert data["symbolic_target"] == "refs/heads/dev"
389 assert data["detached"] is False
390
391
392 # ---------------------------------------------------------------------------
393 # Stress / performance
394 # ---------------------------------------------------------------------------
395
396
397 class TestStress:
398 def test_P1_100_rapid_reads_all_have_duration_ms(
399 self, repo: pathlib.Path
400 ) -> None:
401 for i in range(100):
402 r = _sr(repo, "HEAD")
403 assert r.exit_code == 0
404 data = json.loads(r.output)
405 assert "duration_ms" in data, f"Missing duration_ms on call {i}"
406 assert isinstance(data["duration_ms"], float)
407
408 def test_P2_duration_ms_always_positive(self, repo: pathlib.Path) -> None:
409 for _ in range(20):
410 data = json.loads(_sr(repo, "HEAD").output)
411 assert data["duration_ms"] >= 0.0
412
413 def test_P3_20_branch_writes_all_include_duration_ms(
414 self, tmp_path: pathlib.Path
415 ) -> None:
416 r = _init_repo(tmp_path)
417 sid = _snap(r)
418 for i in range(20):
419 _commit(r, sid, f"branch-{i:02d}")
420 for i in range(20):
421 result = _sr(r, "--set", f"branch-{i:02d}", "HEAD")
422 assert result.exit_code == 0
423 data = json.loads(result.output)
424 assert "duration_ms" in data, f"Missing duration_ms on branch {i}"
425
426
427 # ---------------------------------------------------------------------------
428 # Concurrent
429 # ---------------------------------------------------------------------------
430
431
432 class TestConcurrent:
433 def test_C1_8_concurrent_reads(self, tmp_path: pathlib.Path) -> None:
434 """8 threads reading symbolic-ref in separate repos — all succeed."""
435 results: list = [None] * 8
436
437 def _work(idx: int) -> None:
438 repo = tmp_path / f"repo_{idx}"
439 repo.mkdir()
440 r = _init_repo(repo)
441 sid = _snap(r)
442 _commit(r, sid)
443 res = _sr(r, "HEAD")
444 results[idx] = res.exit_code
445
446 threads = [threading.Thread(target=_work, args=(i,)) for i in range(8)]
447 for t in threads:
448 t.start()
449 for t in threads:
450 t.join()
451
452 for i, code in enumerate(results):
453 assert not isinstance(code, Exception), f"Thread {i}: {code}"
454 assert code == 0, f"Thread {i} exit code: {code}"
455
456 def test_C2_4_concurrent_writes(self, tmp_path: pathlib.Path) -> None:
457 """4 threads writing --set in separate repos — all succeed."""
458 results: list = [None] * 4
459
460 def _work(idx: int) -> None:
461 repo = tmp_path / f"repo_{idx}"
462 repo.mkdir()
463 r = _init_repo(repo)
464 sid = _snap(r)
465 _commit(r, sid, "main")
466 _commit(r, sid, "dev")
467 res = _sr(r, "--set", "dev", "HEAD")
468 results[idx] = res.exit_code
469
470 threads = [threading.Thread(target=_work, args=(i,)) for i in range(4)]
471 for t in threads:
472 t.start()
473 for t in threads:
474 t.join()
475
476 for i, code in enumerate(results):
477 assert not isinstance(code, Exception), f"Thread {i}: {code}"
478 assert code == 0, f"Thread {i} exit code: {code}"
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago