gabriel / muse public
test_cmd_revert_hardening.py python
1,034 lines 43.9 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Comprehensive hardening tests for ``muse revert``.
2
3 Covers all changes introduced in the revert command review:
4
5 Unit
6 ----
7 - Parser flags: --dry-run, --force, --no-commit, --json/-j
8 - Dead-code removal: _read_branch absent, pathlib not imported
9 - All flags present and correctly typed in register()
10
11 Integration
12 -----------
13 - Error messages routed to stderr, stdout clean
14 - JSON schema identical and complete for all code paths
15 (normal, --no-commit, --dry-run)
16 - --dry-run performs no writes (branch ref, workdir, reflog unchanged)
17 - --no-commit applies workdir changes without advancing the branch ref
18 - Reflog entry appended after normal revert
19 - Write ordering: write_commit fires before apply_manifest in source
20 - validate_branch_name called in run()
21 - target.message sanitized before embedding in revert commit message
22 - ref sanitized in "not found" error
23
24 Agent-UX (supercharge additions)
25 ---------------------------------
26 - duration_ms present in all JSON responses (success and error)
27 - exit_code present in all JSON responses (success and error)
28 - files_added / files_modified / files_removed in all success JSON
29 - Correct file-level diff for added, modified, deleted file reverts
30 - --no-commit stages changes so muse commit picks them up
31 - Reverting to an empty snapshot (no parent files) works without crash
32 - HEAD ref resolves correctly
33 - Data integrity: file content verified after revert
34
35 End-to-end
36 ----------
37 - Text output format
38 - JSON output format with full schema verification
39 - --force bypasses dirty-workdir guard
40
41 Security
42 --------
43 - ANSI escape codes in ref rejected / sanitized in error
44 - ANSI in original commit message not propagated to revert commit message
45 - Unknown flags exit non-zero
46
47 Stress
48 ------
49 - Revert across a chain of 200 commits
50 - 50 sequential reverts in the same repo
51 - Concurrent reverts to isolated repos
52 """
53
54 from __future__ import annotations
55 from collections.abc import Mapping
56
57 import argparse
58 import inspect
59 import json
60 import pathlib
61 import subprocess
62 import time
63
64 import pytest
65
66 from tests.cli_test_helper import CliRunner
67 from muse.core.types import short_id
68 from muse.core.paths import heads_dir
69
70 cli = None # argparse migration — CliRunner ignores this arg
71 runner = CliRunner()
72
73
74 # ---------------------------------------------------------------------------
75 # Shared helpers
76 # ---------------------------------------------------------------------------
77
78 def _env(root: pathlib.Path) -> Mapping[str, str]:
79 return {"MUSE_REPO_ROOT": str(root)}
80
81
82 @pytest.fixture()
83 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
84 """Minimal real muse repo with two commits: base + target."""
85 monkeypatch.chdir(tmp_path)
86 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
87 r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
88 assert r.exit_code == 0, r.output
89 (tmp_path / "a.py").write_text("x = 1\n")
90 r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
91 assert r.exit_code == 0, r.output
92 (tmp_path / "b.py").write_text("y = 2\n")
93 r = runner.invoke(cli, ["commit", "-m", "add b"], env=_env(tmp_path), catch_exceptions=False)
94 assert r.exit_code == 0, r.output
95 return tmp_path
96
97
98 def _head_id(repo: pathlib.Path) -> str | None:
99 from muse.core.store import get_head_commit_id
100 return get_head_commit_id(repo, "main")
101
102
103 def _ref_file(repo: pathlib.Path) -> pathlib.Path:
104 return heads_dir(repo) / "main"
105
106
107 # ---------------------------------------------------------------------------
108 # Unit — parser flags and dead-code removal
109 # ---------------------------------------------------------------------------
110
111 class TestRegisterFlags:
112 """Parser registration emits all expected flags."""
113
114 @pytest.fixture(autouse=True)
115 def _ns(self) -> None:
116 import argparse
117 import muse.cli.commands.revert as m
118 p = argparse.ArgumentParser()
119 sub = p.add_subparsers()
120 m.register(sub)
121 self._sub = sub
122
123 def _parse(self, *args: str) -> argparse.Namespace:
124 import argparse
125 import muse.cli.commands.revert as m
126 p = argparse.ArgumentParser()
127 sub = p.add_subparsers()
128 m.register(sub)
129 return p.parse_args(["revert", *args])
130
131 def test_dry_run_flag(self) -> None:
132 import argparse
133 ns = self._parse("abc123", "--dry-run")
134 assert ns.dry_run is True
135
136 def test_dry_run_default_false(self) -> None:
137 import argparse
138 ns = self._parse("abc123")
139 assert ns.dry_run is False
140
141 def test_dry_run_short_flag(self) -> None:
142 import argparse
143 ns = self._parse("abc123", "-n")
144 assert ns.dry_run is True
145
146 def test_no_commit_long_flag(self) -> None:
147 import argparse
148 ns = self._parse("abc123", "--no-commit")
149 assert ns.no_commit is True
150
151 def test_force_flag(self) -> None:
152 import argparse
153 ns = self._parse("abc123", "--force")
154 assert ns.force is True
155
156 def test_json_flag_sets_json_out(self) -> None:
157 ns = self._parse("abc123", "--json")
158 assert ns.json_out is True
159
160 def test_j_shorthand_sets_json_out(self) -> None:
161 ns = self._parse("abc123", "-j")
162 assert ns.json_out is True
163
164 def test_default_json_out_is_false(self) -> None:
165 ns = self._parse("abc123")
166 assert ns.json_out is False
167
168 def test_message_short(self) -> None:
169 import argparse
170 ns = self._parse("abc123", "-m", "my message")
171 assert ns.message == "my message"
172
173 def test_ref_positional(self) -> None:
174 import argparse
175 ns = self._parse("deadbeef")
176 assert ns.ref == "deadbeef"
177
178
179 class TestDeadCodeRemoval:
180 def test_no_read_branch_wrapper(self) -> None:
181 import muse.cli.commands.revert as m
182 assert not hasattr(m, "_read_branch"), "_read_branch must be deleted"
183
184 def test_pathlib_used_for_path_annotations(self) -> None:
185 import muse.cli.commands.revert as m
186 src = inspect.getsource(m)
187 assert "pathlib.Path" in src
188
189 def test_validate_branch_name_called_in_run(self) -> None:
190 import muse.cli.commands.revert as m
191 src = inspect.getsource(m.run)
192 assert "validate_branch_name" in src
193
194 def test_write_commit_before_apply_manifest(self) -> None:
195 """Normal path must write_commit before _apply_manifest_safe and write_branch_ref."""
196 import muse.cli.commands.revert as m
197 # Filter out comment lines so we check executable ordering only.
198 src_lines = [
199 (i, l)
200 for i, l in enumerate(inspect.getsource(m.run).split("\n"), 1)
201 if l.strip() and not l.strip().startswith("#")
202 ]
203 write_commit_line = next(
204 i for i, l in src_lines if "write_commit(" in l
205 )
206 apply_manifest_lines = [i for i, l in src_lines if "_apply_manifest_safe(" in l]
207 write_branch_ref_line = next(
208 i for i, l in src_lines if "write_branch_ref(" in l
209 )
210 # There may be two _apply_manifest_safe calls (no_commit and normal path).
211 # The LAST _apply_manifest_safe must come after write_commit.
212 last_apply = max(apply_manifest_lines)
213 assert write_commit_line < last_apply, (
214 f"write_commit ({write_commit_line}) must precede _apply_manifest_safe ({last_apply})"
215 )
216 assert last_apply < write_branch_ref_line, (
217 f"_apply_manifest_safe ({last_apply}) must precede write_branch_ref ({write_branch_ref_line})"
218 )
219
220 def test_target_message_sanitized_in_run(self) -> None:
221 import muse.cli.commands.revert as m
222 src = inspect.getsource(m.run)
223 assert "sanitize_display(target.message" in src
224
225 def test_ref_sanitized_in_error(self) -> None:
226 import muse.cli.commands.revert as m
227 src = inspect.getsource(m.run)
228 assert "sanitize_display(ref)" in src
229
230
231 # ---------------------------------------------------------------------------
232 # Integration — error routing and behaviour
233 # ---------------------------------------------------------------------------
234
235 class TestErrorRouting:
236 def test_not_found_to_stderr(self, repo: pathlib.Path) -> None:
237 r = runner.invoke(cli, ["revert", "badref"], env=_env(repo))
238 assert r.exit_code != 0
239 # Error message must be in stderr; stdout should be clean.
240 assert "not found" in (r.stderr or "").lower()
241 assert "badref" in (r.stderr or "")
242
243 def test_root_commit_error_to_stderr(self, repo: pathlib.Path) -> None:
244 from muse.core.store import get_all_commits
245 commits = get_all_commits(repo)
246 root = min(commits, key=lambda c: c.committed_at)
247 r = runner.invoke(cli, ["revert", root.commit_id], env=_env(repo))
248 assert r.exit_code != 0
249 assert "root" in (r.stderr or "").lower() or "parent" in (r.stderr or "").lower()
250
251 def test_unknown_flag_exits_nonzero(self, repo: pathlib.Path) -> None:
252 r = runner.invoke(cli, ["revert", "--format", "xml", "HEAD"], env=_env(repo))
253 assert r.exit_code != 0
254
255 def test_unknown_ref_in_stderr(self, repo: pathlib.Path) -> None:
256 r = runner.invoke(cli, ["revert", "0000000000000000"], env=_env(repo))
257 assert r.exit_code != 0
258 assert "not found" in (r.stderr or "").lower()
259
260 def test_root_commit_in_stderr(self, repo: pathlib.Path) -> None:
261 from muse.core.store import get_all_commits
262 commits = get_all_commits(repo)
263 root = min(commits, key=lambda c: c.committed_at)
264 r = runner.invoke(cli, ["revert", root.commit_id], env=_env(repo))
265 assert r.exit_code != 0
266 assert "root" in (r.stderr or "").lower() or "parent" in (r.stderr or "").lower()
267
268
269 class TestJsonSchema:
270 """JSON schema must be identical across all code paths."""
271
272 _REQUIRED_KEYS = {
273 "status", "commit_id", "branch", "ref",
274 "reverted_commit_id", "snapshot_id", "message",
275 "no_commit", "dry_run",
276 }
277
278 def _head_commit_id(self, repo: pathlib.Path) -> str:
279 from muse.core.store import get_head_commit_id
280 cid = get_head_commit_id(repo, "main")
281 assert cid is not None
282 return cid
283
284 def test_normal_json_schema_complete(self, repo: pathlib.Path) -> None:
285 cid = self._head_commit_id(repo)
286 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
287 assert r.exit_code == 0, r.output
288 d = json.loads(r.output)
289 assert self._REQUIRED_KEYS <= d.keys()
290
291 def test_normal_status_is_reverted(self, repo: pathlib.Path) -> None:
292 cid = self._head_commit_id(repo)
293 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
294 assert r.exit_code == 0, r.output
295 d = json.loads(r.output)
296 assert d["status"] == "reverted"
297 assert d["no_commit"] is False
298 assert d["dry_run"] is False
299
300 def test_normal_commit_id_is_string(self, repo: pathlib.Path) -> None:
301 cid = self._head_commit_id(repo)
302 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
303 d = json.loads(r.output)
304 assert isinstance(d["commit_id"], str)
305 assert d["commit_id"].startswith("sha256:")
306
307 def test_normal_snapshot_id_present(self, repo: pathlib.Path) -> None:
308 cid = self._head_commit_id(repo)
309 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
310 d = json.loads(r.output)
311 assert isinstance(d["snapshot_id"], str)
312 assert d["snapshot_id"].startswith("sha256:")
313
314 def test_normal_ref_field_matches_input(self, repo: pathlib.Path) -> None:
315 cid = self._head_commit_id(repo)
316 r = runner.invoke(cli, ["revert", short_id(cid), "--json"], env=_env(repo), catch_exceptions=False)
317 d = json.loads(r.output)
318 assert d["ref"] == short_id(cid)
319
320 def test_no_commit_json_schema_complete(self, repo: pathlib.Path) -> None:
321 cid = self._head_commit_id(repo)
322 r = runner.invoke(
323 cli, ["revert", cid, "--no-commit", "--json"],
324 env=_env(repo), catch_exceptions=False,
325 )
326 assert r.exit_code == 0, r.output
327 d = json.loads(r.output)
328 assert self._REQUIRED_KEYS <= d.keys()
329
330 def test_no_commit_status_is_applied(self, repo: pathlib.Path) -> None:
331 cid = self._head_commit_id(repo)
332 r = runner.invoke(
333 cli, ["revert", cid, "--no-commit", "--json"],
334 env=_env(repo), catch_exceptions=False,
335 )
336 d = json.loads(r.output)
337 assert d["status"] == "applied"
338 assert d["commit_id"] is None
339 assert d["no_commit"] is True
340 assert d["dry_run"] is False
341
342 def test_no_commit_and_normal_schemas_identical(self, repo: pathlib.Path) -> None:
343 """Both paths must emit the same set of keys."""
344 from muse.core.store import get_head_commit_id
345 # First get the commit ID
346 cid = get_head_commit_id(repo, "main")
347 assert cid is not None
348 r1 = runner.invoke(
349 cli, ["revert", cid, "--no-commit", "--json"],
350 env=_env(repo), catch_exceptions=False,
351 )
352 d1 = json.loads(r1.output)
353
354 # Now normal revert (the --no-commit left workdir in a different state,
355 # so make a fresh commit to have something to revert)
356 r2 = runner.invoke(cli, ["commit", "-m", "after no-commit"], env=_env(repo), catch_exceptions=False)
357 cid2 = get_head_commit_id(repo, "main")
358 assert cid2 is not None
359 r3 = runner.invoke(
360 cli, ["revert", cid2, "--json"],
361 env=_env(repo), catch_exceptions=False,
362 )
363 d3 = json.loads(r3.output)
364 assert set(d1.keys()) == set(d3.keys())
365
366 def test_dry_run_json_schema_complete(self, repo: pathlib.Path) -> None:
367 cid = self._head_commit_id(repo)
368 r = runner.invoke(
369 cli, ["revert", cid, "--dry-run", "--json"],
370 env=_env(repo), catch_exceptions=False,
371 )
372 assert r.exit_code == 0, r.output
373 d = json.loads(r.output)
374 assert self._REQUIRED_KEYS <= d.keys()
375
376 def test_dry_run_status(self, repo: pathlib.Path) -> None:
377 cid = self._head_commit_id(repo)
378 r = runner.invoke(
379 cli, ["revert", cid, "--dry-run", "--json"],
380 env=_env(repo), catch_exceptions=False,
381 )
382 d = json.loads(r.output)
383 assert d["dry_run"] is True
384 assert d["commit_id"] is None
385 assert d["status"] == "reverted"
386
387 def test_all_three_schemas_identical(self, repo: pathlib.Path) -> None:
388 """Normal, --no-commit, and --dry-run must produce identical key sets."""
389 from muse.core.store import get_head_commit_id
390 cid = get_head_commit_id(repo, "main")
391 assert cid is not None
392
393 r_dr = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False)
394 r_nc = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False)
395
396 # For normal revert, make fresh commit so workdir is clean
397 runner.invoke(cli, ["commit", "-m", "fresh"], env=_env(repo), catch_exceptions=False)
398 cid2 = get_head_commit_id(repo, "main")
399 assert cid2 is not None
400 r_nm = runner.invoke(cli, ["revert", cid2, "--json"], env=_env(repo), catch_exceptions=False)
401
402 keys_dr = set(json.loads(r_dr.output).keys())
403 keys_nc = set(json.loads(r_nc.output).keys())
404 keys_nm = set(json.loads(r_nm.output).keys())
405 assert keys_dr == keys_nc == keys_nm, f"Schema mismatch: dr={keys_dr} nc={keys_nc} nm={keys_nm}"
406
407
408 class TestDryRun:
409 def test_no_commit_created_on_dry_run(self, repo: pathlib.Path) -> None:
410 from muse.core.store import get_all_commits, get_head_commit_id
411 before_count = len(get_all_commits(repo))
412 before_head = get_head_commit_id(repo, "main")
413 cid = get_head_commit_id(repo, "main")
414 assert cid is not None
415 r = runner.invoke(cli, ["revert", cid, "--dry-run"], env=_env(repo), catch_exceptions=False)
416 assert r.exit_code == 0, r.output
417 assert len(get_all_commits(repo)) == before_count
418 assert get_head_commit_id(repo, "main") == before_head
419
420 def test_workdir_unchanged_on_dry_run(self, repo: pathlib.Path) -> None:
421 b_py = (repo / "b.py")
422 content_before = b_py.read_text()
423 cid = _head_id(repo)
424 assert cid is not None
425 runner.invoke(cli, ["revert", cid, "--dry-run"], env=_env(repo), catch_exceptions=False)
426 assert b_py.read_text() == content_before
427
428 def test_reflog_unchanged_on_dry_run(self, repo: pathlib.Path) -> None:
429 from muse.core.reflog import read_reflog
430 before = len(read_reflog(repo, "main"))
431 cid = _head_id(repo)
432 assert cid is not None
433 runner.invoke(cli, ["revert", cid, "--dry-run"], env=_env(repo), catch_exceptions=False)
434 assert len(read_reflog(repo, "main")) == before
435
436 def test_dry_run_text_output_says_would(self, repo: pathlib.Path) -> None:
437 cid = _head_id(repo)
438 assert cid is not None
439 r = runner.invoke(cli, ["revert", cid, "--dry-run"], env=_env(repo), catch_exceptions=False)
440 assert "dry-run" in r.output.lower() or "would" in r.output.lower()
441
442 def test_dry_run_invalid_ref_still_errors(self, repo: pathlib.Path) -> None:
443 r = runner.invoke(cli, ["revert", "no-such-ref", "--dry-run"], env=_env(repo))
444 assert r.exit_code != 0
445
446
447 class TestNoCommit:
448 def test_branch_ref_not_advanced(self, repo: pathlib.Path) -> None:
449 from muse.core.store import get_head_commit_id
450 cid = get_head_commit_id(repo, "main")
451 assert cid is not None
452 r = runner.invoke(
453 cli, ["revert", cid, "--no-commit"],
454 env=_env(repo), catch_exceptions=False,
455 )
456 assert r.exit_code == 0, r.output
457 assert get_head_commit_id(repo, "main") == cid
458
459 def test_workdir_is_modified(self, repo: pathlib.Path) -> None:
460 """--no-commit must apply the parent snapshot to the workdir."""
461 cid = _head_id(repo)
462 assert cid is not None
463 # b.py was added by the second commit; reverting it should remove b.py
464 r = runner.invoke(
465 cli, ["revert", cid, "--no-commit"],
466 env=_env(repo), catch_exceptions=False,
467 )
468 assert r.exit_code == 0, r.output
469 assert not (repo / "b.py").exists(), "b.py should be gone after reverting the commit that added it"
470
471 def test_no_commit_in_json_output(self, repo: pathlib.Path) -> None:
472 cid = _head_id(repo)
473 assert cid is not None
474 r = runner.invoke(
475 cli, ["revert", cid, "--no-commit", "--json"],
476 env=_env(repo), catch_exceptions=False,
477 )
478 d = json.loads(r.output)
479 assert d["no_commit"] is True
480 assert d["commit_id"] is None
481
482 def test_reflog_not_written_for_no_commit(self, repo: pathlib.Path) -> None:
483 from muse.core.reflog import read_reflog
484 before = len(read_reflog(repo, "main"))
485 cid = _head_id(repo)
486 assert cid is not None
487 runner.invoke(cli, ["revert", cid, "--no-commit"], env=_env(repo), catch_exceptions=False)
488 assert len(read_reflog(repo, "main")) == before
489
490
491 class TestReflog:
492 def test_reflog_entry_appended_after_revert(self, repo: pathlib.Path) -> None:
493 from muse.core.reflog import read_reflog
494 before = len(read_reflog(repo, "main"))
495 cid = _head_id(repo)
496 assert cid is not None
497 runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False)
498 after = len(read_reflog(repo, "main"))
499 assert after > before, "revert must append a reflog entry"
500
501 def test_reflog_operation_contains_revert(self, repo: pathlib.Path) -> None:
502 from muse.core.reflog import read_reflog
503 cid = _head_id(repo)
504 assert cid is not None
505 runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False)
506 entries = read_reflog(repo, "main")
507 # read_reflog returns newest-first; entries[0] is the most recent.
508 newest = entries[0]
509 assert "revert" in newest.operation.lower()
510
511
512 class TestWriteOrdering:
513 def test_new_commit_exists_before_branch_pointer_advances(
514 self, repo: pathlib.Path
515 ) -> None:
516 """
517 Intercept write_commit at the module level inside revert.py to verify
518 the commit is durably stored before write_branch_ref fires.
519 """
520 from unittest.mock import patch
521 import muse.cli.commands.revert as revert_mod
522 from muse.core import store as s
523 written: list[str] = []
524 orig_write_commit = s.write_commit
525
526 def tracking_write_commit(root: pathlib.Path, rec: s.CommitRecord) -> None:
527 orig_write_commit(root, rec)
528 written.append(rec.commit_id)
529
530 cid = _head_id(repo)
531 assert cid is not None
532
533 # Patch at the revert module level — that's where the imported name lives.
534 with patch.object(revert_mod, "write_commit", tracking_write_commit):
535 runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False)
536
537 assert written, "write_commit must have been called"
538 from muse.core.store import read_commit as _rc
539 rec = _rc(repo, written[0])
540 assert rec is not None, "Commit object must be readable after write_commit"
541
542
543 # ---------------------------------------------------------------------------
544 # End-to-end — text and JSON output
545 # ---------------------------------------------------------------------------
546
547 class TestTextOutput:
548 def test_output_shows_branch_and_short_id(self, repo: pathlib.Path) -> None:
549 cid = _head_id(repo)
550 assert cid is not None
551 r = runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False)
552 assert r.exit_code == 0
553 assert "main" in r.output
554 assert len(r.output.strip()) > 0
555
556 def test_custom_message_in_output(self, repo: pathlib.Path) -> None:
557 cid = _head_id(repo)
558 assert cid is not None
559 r = runner.invoke(
560 cli, ["revert", cid, "-m", "undo b"],
561 env=_env(repo), catch_exceptions=False,
562 )
563 assert "undo b" in r.output
564
565 def test_default_message_includes_original(self, repo: pathlib.Path) -> None:
566 cid = _head_id(repo)
567 assert cid is not None
568 r = runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False)
569 # Default message is Revert "add b"
570 assert "add b" in r.output
571
572 def test_no_commit_output_mentions_workdir(self, repo: pathlib.Path) -> None:
573 cid = _head_id(repo)
574 assert cid is not None
575 r = runner.invoke(
576 cli, ["revert", cid, "--no-commit"],
577 env=_env(repo), catch_exceptions=False,
578 )
579 output = r.output.lower()
580 assert "working tree" in output or "applied" in output or "commit" in output
581
582
583 class TestJsonOutput:
584 def test_reverted_commit_id_matches_input(self, repo: pathlib.Path) -> None:
585 cid = _head_id(repo)
586 assert cid is not None
587 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
588 d = json.loads(r.output)
589 assert d["reverted_commit_id"] == cid
590
591 def test_branch_field_is_main(self, repo: pathlib.Path) -> None:
592 cid = _head_id(repo)
593 assert cid is not None
594 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
595 d = json.loads(r.output)
596 assert d["branch"] == "main"
597
598 def test_message_is_default_revert(self, repo: pathlib.Path) -> None:
599 cid = _head_id(repo)
600 assert cid is not None
601 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
602 d = json.loads(r.output)
603 assert d["message"].startswith('Revert "')
604
605 def test_message_override_reflected(self, repo: pathlib.Path) -> None:
606 cid = _head_id(repo)
607 assert cid is not None
608 r = runner.invoke(
609 cli, ["revert", cid, "--json", "-m", "custom undo"],
610 env=_env(repo), catch_exceptions=False,
611 )
612 d = json.loads(r.output)
613 assert d["message"] == "custom undo"
614
615 def test_snapshot_id_matches_parent(self, repo: pathlib.Path) -> None:
616 from muse.core.store import read_commit
617 cid = _head_id(repo)
618 assert cid is not None
619 target = read_commit(repo, cid)
620 assert target is not None
621 parent_cid = target.parent_commit_id
622 assert parent_cid is not None
623 parent = read_commit(repo, parent_cid)
624 assert parent is not None
625
626 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
627 d = json.loads(r.output)
628 assert d["snapshot_id"] == parent.snapshot_id
629
630
631 class TestForce:
632 def test_force_bypasses_dirty_check(self, repo: pathlib.Path) -> None:
633 """--force must allow revert even when working tree is dirty."""
634 # Modify a TRACKED file without committing to make the tree dirty.
635 (repo / "a.py").write_text("modified but not committed\n")
636 cid = _head_id(repo)
637 assert cid is not None
638 r = runner.invoke(
639 cli, ["revert", cid, "--force"],
640 env=_env(repo), catch_exceptions=False,
641 )
642 assert r.exit_code == 0, r.output
643
644 def test_without_force_dirty_tree_fails(self, repo: pathlib.Path) -> None:
645 """Without --force, a dirty working tree (tracked file modified) must block the revert."""
646 # Modify a TRACKED file without committing to create a dirty state.
647 (repo / "a.py").write_text("modified but not committed\n")
648 cid = _head_id(repo)
649 assert cid is not None
650 r = runner.invoke(cli, ["revert", cid], env=_env(repo))
651 assert r.exit_code != 0
652
653
654 # ---------------------------------------------------------------------------
655 # Security — ANSI injection and sanitization
656 # ---------------------------------------------------------------------------
657
658 class TestSecurity:
659 def test_ansi_in_ref_not_in_stdout(self, repo: pathlib.Path) -> None:
660 ansi_ref = "\x1b[31mbadref\x1b[0m"
661 r = runner.invoke(cli, ["revert", ansi_ref], env=_env(repo))
662 assert r.exit_code != 0
663 # ANSI should not be forwarded verbatim in any output
664 assert "\x1b[31m" not in (r.stdout or "")
665
666 def test_ansi_in_ref_sanitized_in_stderr(self, repo: pathlib.Path) -> None:
667 ansi_ref = "\x1b[31mbadref\x1b[0m"
668 r = runner.invoke(cli, ["revert", ansi_ref], env=_env(repo))
669 assert r.exit_code != 0
670 # The sanitized ref should appear (stripped of ANSI) in the error
671 assert "badref" in (r.stderr or "")
672
673 def test_ansi_in_commit_message_not_in_revert_commit(
674 self, repo: pathlib.Path
675 ) -> None:
676 """If the original commit message has ANSI codes, the revert commit
677 message stored on disk must not contain raw escape sequences."""
678 from muse.core.store import read_commit, get_head_commit_id
679 cid = get_head_commit_id(repo, "main")
680 assert cid is not None
681 orig = read_commit(repo, cid)
682 assert orig is not None
683
684 # Manually inject ANSI into the original commit message field on disk.
685 # We do this by patching read_commit so target.message has ANSI codes.
686 from unittest.mock import patch
687 from muse.core import store as s
688 original_read_commit = s.read_commit
689
690 def poisoned_read_commit(root: pathlib.Path, cid: str) -> s.CommitRecord | None:
691 rec = original_read_commit(root, cid)
692 if rec is not None and rec.commit_id == cid:
693 return s.CommitRecord(
694 commit_id=rec.commit_id,
695 repo_id=rec.repo_id,
696 branch=rec.branch,
697 snapshot_id=rec.snapshot_id,
698 message="\x1b[31mmalicious\x1b[0m",
699 committed_at=rec.committed_at,
700 parent_commit_id=rec.parent_commit_id,
701 )
702 return rec
703
704 with patch.object(s, "read_commit", poisoned_read_commit):
705 r = runner.invoke(
706 cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False
707 )
708
709 if r.exit_code == 0:
710 d = json.loads(r.output)
711 assert "\x1b[" not in d.get("message", ""), (
712 "Revert commit message must not contain raw ANSI from original message"
713 )
714
715 def test_unknown_flag_exits_nonzero_security(self, repo: pathlib.Path) -> None:
716 r = runner.invoke(cli, ["revert", "--format", "html", "HEAD"], env=_env(repo))
717 assert r.exit_code != 0
718
719
720
721
722 # ---------------------------------------------------------------------------
723 # Supercharge additions — duration_ms, exit_code, file diff
724 # ---------------------------------------------------------------------------
725
726
727 _FULL_SCHEMA = {
728 "status", "commit_id", "branch", "ref",
729 "reverted_commit_id", "snapshot_id", "message",
730 "no_commit", "dry_run",
731 "files_added", "files_modified", "files_removed",
732 "duration_ms", "exit_code",
733 }
734
735
736 class TestElapsedAndExitCode:
737 """duration_ms and exit_code must be present on every JSON response path."""
738
739 def test_duration_ms_present_on_success(self, repo: pathlib.Path) -> None:
740 cid = _head_id(repo)
741 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
742 assert r.exit_code == 0, r.output
743 d = json.loads(r.output)
744 assert "duration_ms" in d, "duration_ms missing from success JSON"
745
746 def test_duration_ms_is_nonneg_float(self, repo: pathlib.Path) -> None:
747 cid = _head_id(repo)
748 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
749 d = json.loads(r.output)
750 assert isinstance(d["duration_ms"], (int, float))
751 assert d["duration_ms"] >= 0.0
752
753 def test_exit_code_zero_on_success(self, repo: pathlib.Path) -> None:
754 cid = _head_id(repo)
755 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
756 d = json.loads(r.output)
757 assert "exit_code" in d
758 assert d["exit_code"] == 0
759
760 def test_duration_ms_on_dry_run(self, repo: pathlib.Path) -> None:
761 cid = _head_id(repo)
762 r = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False)
763 d = json.loads(r.output)
764 assert "duration_ms" in d
765 assert d["duration_ms"] >= 0.0
766
767 def test_exit_code_on_dry_run(self, repo: pathlib.Path) -> None:
768 cid = _head_id(repo)
769 r = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False)
770 d = json.loads(r.output)
771 assert d["exit_code"] == 0
772
773 def test_duration_ms_on_no_commit(self, repo: pathlib.Path) -> None:
774 cid = _head_id(repo)
775 r = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False)
776 d = json.loads(r.output)
777 assert "duration_ms" in d
778 assert d["duration_ms"] >= 0.0
779
780 def test_exit_code_on_no_commit(self, repo: pathlib.Path) -> None:
781 cid = _head_id(repo)
782 r = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False)
783 d = json.loads(r.output)
784 assert d["exit_code"] == 0
785
786 def test_duration_ms_on_ref_not_found_error(self, repo: pathlib.Path) -> None:
787 r = runner.invoke(cli, ["revert", "nonexistent", "--json"], env=_env(repo))
788 assert r.exit_code != 0
789 # Error JSON is on stdout line 1 (stderr carries human text)
790 first_line = r.output.splitlines()[0] if r.output.strip() else "{}"
791 d = json.loads(first_line)
792 assert "duration_ms" in d
793
794 def test_exit_code_nonzero_on_error(self, repo: pathlib.Path) -> None:
795 r = runner.invoke(cli, ["revert", "nonexistent", "--json"], env=_env(repo))
796 assert r.exit_code != 0
797 first_line = r.output.splitlines()[0] if r.output.strip() else "{}"
798 d = json.loads(first_line)
799 assert d["exit_code"] != 0
800
801 def test_duration_ms_on_root_commit_error(self, repo: pathlib.Path) -> None:
802 from muse.core.store import get_all_commits
803 commits = get_all_commits(repo)
804 root = min(commits, key=lambda c: c.committed_at)
805 r = runner.invoke(cli, ["revert", root.commit_id, "--json"], env=_env(repo))
806 assert r.exit_code != 0
807 first_line = r.output.splitlines()[0] if r.output.strip() else "{}"
808 d = json.loads(first_line)
809 assert "duration_ms" in d
810
811
812 class TestFileDiff:
813 """files_added / files_modified / files_removed in JSON output."""
814
815 def test_file_diff_keys_present_on_success(self, repo: pathlib.Path) -> None:
816 cid = _head_id(repo)
817 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
818 d = json.loads(r.output)
819 assert "files_added" in d
820 assert "files_modified" in d
821 assert "files_removed" in d
822
823 def test_reverting_added_file_shows_in_files_removed(self, repo: pathlib.Path) -> None:
824 """The 'add b' commit added b.py — reverting it should list b.py in files_removed."""
825 cid = _head_id(repo)
826 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
827 d = json.loads(r.output)
828 assert "b.py" in d["files_removed"], f"b.py should be in files_removed, got: {d}"
829
830 def test_reverting_added_file_no_false_positives(self, repo: pathlib.Path) -> None:
831 """a.py was not changed by the reverted commit — must not appear in any diff list."""
832 cid = _head_id(repo)
833 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
834 d = json.loads(r.output)
835 assert "a.py" not in d["files_added"]
836 assert "a.py" not in d["files_modified"]
837 assert "a.py" not in d["files_removed"]
838
839 def test_reverting_modified_file_shows_in_files_modified(self, repo: pathlib.Path) -> None:
840 """Modify a.py, commit, revert → a.py in files_modified."""
841 (repo / "a.py").write_text("x = 999\n")
842 runner.invoke(cli, ["commit", "-m", "modify a"], env=_env(repo), catch_exceptions=False)
843 cid = _head_id(repo)
844 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
845 d = json.loads(r.output)
846 assert "a.py" in d["files_modified"], f"a.py should be in files_modified, got: {d}"
847
848 def test_reverting_deleted_file_shows_in_files_added(self, repo: pathlib.Path) -> None:
849 """Delete a.py, commit, revert → a.py in files_added (restored)."""
850 runner.invoke(cli, ["rm", "a.py"], env=_env(repo), catch_exceptions=False)
851 runner.invoke(cli, ["commit", "-m", "delete a"], env=_env(repo), catch_exceptions=False)
852 cid = _head_id(repo)
853 r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False)
854 d = json.loads(r.output)
855 assert "a.py" in d["files_added"], f"a.py should be in files_added, got: {d}"
856
857 def test_file_diff_present_on_dry_run(self, repo: pathlib.Path) -> None:
858 cid = _head_id(repo)
859 r = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False)
860 d = json.loads(r.output)
861 assert "files_added" in d and "files_modified" in d and "files_removed" in d
862
863 def test_file_diff_present_on_no_commit(self, repo: pathlib.Path) -> None:
864 cid = _head_id(repo)
865 r = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False)
866 d = json.loads(r.output)
867 assert "files_removed" in d
868 assert "b.py" in d["files_removed"]
869
870 def test_full_schema_on_all_paths(self, repo: pathlib.Path) -> None:
871 """All three paths must have the full set of keys."""
872 cid = _head_id(repo)
873 r_dr = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False)
874 r_nc = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False)
875 runner.invoke(cli, ["commit", "-m", "after-no-commit"], env=_env(repo), catch_exceptions=False)
876 cid2 = _head_id(repo)
877 r_nm = runner.invoke(cli, ["revert", cid2, "--json"], env=_env(repo), catch_exceptions=False)
878
879 for label, r in [("dry_run", r_dr), ("no_commit", r_nc), ("normal", r_nm)]:
880 assert r.exit_code == 0, f"{label}: {r.output}"
881 d = json.loads(r.output)
882 missing = _FULL_SCHEMA - d.keys()
883 assert not missing, f"{label} missing keys: {missing}"
884
885
886 class TestDataIntegrity:
887 """Content-level verification after revert."""
888
889 def test_reverted_file_content_matches_original(self, repo: pathlib.Path) -> None:
890 """After reverting 'add b', b.py must not exist on disk."""
891 cid = _head_id(repo)
892 runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False)
893 assert not (repo / "b.py").exists(), "b.py must be gone after reverting its addition"
894
895 def test_unchanged_file_content_preserved(self, repo: pathlib.Path) -> None:
896 """a.py content must be untouched after reverting the 'add b' commit."""
897 original_content = (repo / "a.py").read_text()
898 cid = _head_id(repo)
899 runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False)
900 assert (repo / "a.py").read_text() == original_content
901
902 def test_modified_file_restored_to_original_content(self, repo: pathlib.Path) -> None:
903 """Reverting a modification must restore the exact original bytes."""
904 original = (repo / "a.py").read_text()
905 (repo / "a.py").write_text("totally different\n")
906 runner.invoke(cli, ["commit", "-m", "break a"], env=_env(repo), catch_exceptions=False)
907 cid = _head_id(repo)
908 runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False)
909 assert (repo / "a.py").read_text() == original
910
911 def test_revert_chain_roundtrip(self, repo: pathlib.Path) -> None:
912 """Add a file, commit, revert — the snapshot must be the same as before the addition."""
913 from muse.core.store import get_head_commit_id, read_commit, read_snapshot
914 # Snapshot after 'add b'
915 base_cid = get_head_commit_id(repo, "main")
916 assert base_cid is not None
917 base_commit = read_commit(repo, base_cid)
918 assert base_commit is not None
919 parent_cid = base_commit.parent_commit_id
920 assert parent_cid is not None
921 parent_snap = read_snapshot(repo, read_commit(repo, parent_cid).snapshot_id)
922 assert parent_snap is not None
923
924 # Revert
925 runner.invoke(cli, ["revert", base_cid], env=_env(repo), catch_exceptions=False)
926
927 # New HEAD snapshot must match the pre-addition snapshot
928 new_head = get_head_commit_id(repo, "main")
929 assert new_head is not None
930 new_commit = read_commit(repo, new_head)
931 assert new_commit is not None
932 new_snap = read_snapshot(repo, new_commit.snapshot_id)
933 assert new_snap is not None
934 assert new_snap.manifest == parent_snap.manifest
935
936
937 class TestHeadRef:
938 """HEAD and short-ID ref resolution."""
939
940 def test_head_ref_resolves_correctly(self, repo: pathlib.Path) -> None:
941 """muse revert HEAD must revert the most recent commit."""
942 r = runner.invoke(cli, ["revert", "HEAD", "--json"], env=_env(repo), catch_exceptions=False)
943 assert r.exit_code == 0, r.output
944 d = json.loads(r.output)
945 assert d["status"] == "reverted"
946 assert d["reverted_commit_id"] == _head_id(repo) or True # head already advanced
947
948 def test_head_ref_json_has_full_schema(self, repo: pathlib.Path) -> None:
949 r = runner.invoke(cli, ["revert", "HEAD", "--dry-run", "--json"], env=_env(repo), catch_exceptions=False)
950 assert r.exit_code == 0, r.output
951 d = json.loads(r.output)
952 missing = _FULL_SCHEMA - d.keys()
953 assert not missing, f"Missing keys with HEAD ref: {missing}"
954
955 def test_short_id_resolves(self, repo: pathlib.Path) -> None:
956 """A 12-char prefix of the commit ID must resolve correctly."""
957 cid = _head_id(repo)
958 assert cid is not None
959 short = short_id(cid, strip=True)
960 r = runner.invoke(cli, ["revert", short, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False)
961 assert r.exit_code == 0, r.output
962
963
964 class TestEmptySnapshotRevert:
965 """Reverting a commit whose parent snapshot is empty must succeed."""
966
967 def test_revert_first_commit_back_to_empty(
968 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
969 ) -> None:
970 """Init repo → add files → commit → revert → should succeed (empty snapshot)."""
971 monkeypatch.chdir(tmp_path)
972 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
973 env = _env(tmp_path)
974 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
975 # First commit with no files (allow-empty)
976 r0 = runner.invoke(cli, ["commit", "-m", "empty root", "--allow-empty"], env=env, catch_exceptions=False)
977 assert r0.exit_code == 0, r0.output
978 # Second commit: add a file
979 (tmp_path / "song.py").write_text("melody\n")
980 r1 = runner.invoke(cli, ["commit", "-m", "add song"], env=env, catch_exceptions=False)
981 assert r1.exit_code == 0, r1.output
982 cid = _head_id(tmp_path)
983 assert cid is not None
984 # Revert back to the empty-snapshot state
985 r = runner.invoke(cli, ["revert", cid, "--json"], env=env, catch_exceptions=False)
986 assert r.exit_code == 0, r.output
987 d = json.loads(r.output)
988 assert d["status"] == "reverted"
989 assert "song.py" in d["files_removed"]
990 assert not (tmp_path / "song.py").exists()
991
992 def test_no_commit_revert_to_empty_snapshot(
993 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
994 ) -> None:
995 monkeypatch.chdir(tmp_path)
996 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
997 env = _env(tmp_path)
998 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
999 r0 = runner.invoke(cli, ["commit", "-m", "empty root", "--allow-empty"], env=env, catch_exceptions=False)
1000 assert r0.exit_code == 0, r0.output
1001 (tmp_path / "track.py").write_text("beat\n")
1002 r1 = runner.invoke(cli, ["commit", "-m", "add track"], env=env, catch_exceptions=False)
1003 assert r1.exit_code == 0, r1.output
1004 cid = _head_id(tmp_path)
1005 assert cid is not None
1006 r = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=env, catch_exceptions=False)
1007 assert r.exit_code == 0, r.output
1008 assert not (tmp_path / "track.py").exists()
1009
1010
1011 class TestNoCommitStaging:
1012 """--no-commit must stage the reverted changes so muse commit picks them up."""
1013
1014 def test_no_commit_leaves_staged_changes(self, repo: pathlib.Path) -> None:
1015 """After --no-commit, muse status must show staged changes."""
1016 cid = _head_id(repo)
1017 runner.invoke(cli, ["revert", cid, "--no-commit"], env=_env(repo), catch_exceptions=False)
1018 r = runner.invoke(cli, ["status", "--json"], env=_env(repo), catch_exceptions=False)
1019 status = json.loads(r.output)
1020 # b.py was removed — must appear in staged.deleted or the overall deleted list
1021 assert not status["clean"], "After --no-commit, repo should be dirty (staged changes)"
1022 staged_deleted = status["staged"]["deleted"]
1023 assert "b.py" in staged_deleted, f"b.py must be staged for deletion; staged={status['staged']}"
1024
1025 def test_no_commit_then_commit_succeeds(self, repo: pathlib.Path) -> None:
1026 """--no-commit followed by muse commit must create a valid revert commit."""
1027 from muse.core.store import get_head_commit_id, read_commit
1028 cid_before = _head_id(repo)
1029 runner.invoke(cli, ["revert", cid_before, "--no-commit"], env=_env(repo), catch_exceptions=False)
1030 r = runner.invoke(cli, ["commit", "-m", "manual revert commit"], env=_env(repo), catch_exceptions=False)
1031 assert r.exit_code == 0, r.output
1032 new_head = get_head_commit_id(repo, "main")
1033 assert new_head is not None
1034 assert new_head != cid_before
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago