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