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