"""Comprehensive hardening tests for ``muse revert``. Covers all changes introduced in the revert command review: Unit ---- - Parser flags: --dry-run, --force, --no-commit, --json/-j - Dead-code removal: _read_branch absent, pathlib not imported - All flags present and correctly typed in register() Integration ----------- - Error messages routed to stderr, stdout clean - JSON schema identical and complete for all code paths (normal, --no-commit, --dry-run) - --dry-run performs no writes (branch ref, workdir, reflog unchanged) - --no-commit applies workdir changes without advancing the branch ref - Reflog entry appended after normal revert - Write ordering: write_commit fires before apply_manifest in source - validate_branch_name called in run() - target.message sanitized before embedding in revert commit message - ref sanitized in "not found" error Agent-UX (supercharge additions) --------------------------------- - duration_ms present in all JSON responses (success and error) - exit_code present in all JSON responses (success and error) - files_added / files_modified / files_removed in all success JSON - Correct file-level diff for added, modified, deleted file reverts - --no-commit stages changes so muse commit picks them up - Reverting to an empty snapshot (no parent files) works without crash - HEAD ref resolves correctly - Data integrity: file content verified after revert End-to-end ---------- - Text output format - JSON output format with full schema verification - --force bypasses dirty-workdir guard Security -------- - ANSI escape codes in ref rejected / sanitized in error - ANSI in original commit message not propagated to revert commit message - Unknown flags exit non-zero Stress ------ - Revert across a chain of 200 commits - 50 sequential reverts in the same repo - Concurrent reverts to isolated repos """ from __future__ import annotations from collections.abc import Mapping import argparse import inspect import json import pathlib import subprocess import time import pytest from tests.cli_test_helper import CliRunner from muse.core.types import short_id from muse.core.paths import heads_dir cli = None # argparse migration — CliRunner ignores this arg runner = CliRunner() # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> Mapping[str, str]: return {"MUSE_REPO_ROOT": str(root)} @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Minimal real muse repo with two commits: base + target.""" monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False) assert r.exit_code == 0, r.output (tmp_path / "a.py").write_text("x = 1\n") runner.invoke(cli, ["code", "add", "a.py"], env=_env(tmp_path), catch_exceptions=False) r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False) assert r.exit_code == 0, r.output (tmp_path / "b.py").write_text("y = 2\n") runner.invoke(cli, ["code", "add", "b.py"], env=_env(tmp_path), catch_exceptions=False) r = runner.invoke(cli, ["commit", "-m", "add b"], env=_env(tmp_path), catch_exceptions=False) assert r.exit_code == 0, r.output return tmp_path def _head_id(repo: pathlib.Path) -> str | None: from muse.core.refs import get_head_commit_id return get_head_commit_id(repo, "main") def _ref_file(repo: pathlib.Path) -> pathlib.Path: return heads_dir(repo) / "main" # --------------------------------------------------------------------------- # Unit — parser flags and dead-code removal # --------------------------------------------------------------------------- class TestRegisterFlags: """Parser registration emits all expected flags.""" @pytest.fixture(autouse=True) def _ns(self) -> None: import argparse import muse.cli.commands.revert as m p = argparse.ArgumentParser() sub = p.add_subparsers() m.register(sub) self._sub = sub def _parse(self, *args: str) -> argparse.Namespace: import argparse import muse.cli.commands.revert as m p = argparse.ArgumentParser() sub = p.add_subparsers() m.register(sub) return p.parse_args(["revert", *args]) def test_dry_run_flag(self) -> None: import argparse ns = self._parse("abc123", "--dry-run") assert ns.dry_run is True def test_dry_run_default_false(self) -> None: import argparse ns = self._parse("abc123") assert ns.dry_run is False def test_dry_run_short_flag(self) -> None: import argparse ns = self._parse("abc123", "-n") assert ns.dry_run is True def test_no_commit_long_flag(self) -> None: import argparse ns = self._parse("abc123", "--no-commit") assert ns.no_commit is True def test_force_flag(self) -> None: import argparse ns = self._parse("abc123", "--force") assert ns.force is True def test_json_flag_sets_json_out(self) -> None: ns = self._parse("abc123", "--json") assert ns.json_out is True def test_j_shorthand_sets_json_out(self) -> None: ns = self._parse("abc123", "-j") assert ns.json_out is True def test_default_json_out_is_false(self) -> None: ns = self._parse("abc123") assert ns.json_out is False def test_message_short(self) -> None: import argparse ns = self._parse("abc123", "-m", "my message") assert ns.message == "my message" def test_ref_positional(self) -> None: import argparse ns = self._parse("deadbeef") assert ns.ref == "deadbeef" class TestDeadCodeRemoval: def test_no_read_branch_wrapper(self) -> None: import muse.cli.commands.revert as m assert not hasattr(m, "_read_branch"), "_read_branch must be deleted" def test_pathlib_used_for_path_annotations(self) -> None: import muse.cli.commands.revert as m src = inspect.getsource(m) assert "pathlib.Path" in src def test_validate_branch_name_called_in_run(self) -> None: import muse.cli.commands.revert as m src = inspect.getsource(m.run) assert "validate_branch_name" in src def test_write_commit_before_apply_manifest(self) -> None: """Normal path must write_commit before _apply_manifest_safe and write_branch_ref.""" import muse.cli.commands.revert as m # Filter out comment lines so we check executable ordering only. src_lines = [ (i, l) for i, l in enumerate(inspect.getsource(m.run).split("\n"), 1) if l.strip() and not l.strip().startswith("#") ] write_commit_line = next( i for i, l in src_lines if "write_commit(" in l ) apply_manifest_lines = [i for i, l in src_lines if "_apply_manifest_safe(" in l] write_branch_ref_line = next( i for i, l in src_lines if "write_branch_ref(" in l ) # There may be two _apply_manifest_safe calls (no_commit and normal path). # The LAST _apply_manifest_safe must come after write_commit. last_apply = max(apply_manifest_lines) assert write_commit_line < last_apply, ( f"write_commit ({write_commit_line}) must precede _apply_manifest_safe ({last_apply})" ) assert last_apply < write_branch_ref_line, ( f"_apply_manifest_safe ({last_apply}) must precede write_branch_ref ({write_branch_ref_line})" ) def test_target_message_sanitized_in_run(self) -> None: import muse.cli.commands.revert as m src = inspect.getsource(m.run) assert "sanitize_display(target.message" in src def test_ref_sanitized_in_error(self) -> None: import muse.cli.commands.revert as m src = inspect.getsource(m.run) assert "sanitize_display(ref)" in src # --------------------------------------------------------------------------- # Integration — error routing and behaviour # --------------------------------------------------------------------------- class TestErrorRouting: def test_not_found_to_stderr(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["revert", "badref"], env=_env(repo)) assert r.exit_code != 0 # Error message must be in stderr; stdout should be clean. assert "not found" in (r.stderr or "").lower() assert "badref" in (r.stderr or "") def test_root_commit_error_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.commits import get_all_commits commits = get_all_commits(repo) root = min(commits, key=lambda c: c.committed_at) r = runner.invoke(cli, ["revert", root.commit_id], env=_env(repo)) assert r.exit_code != 0 assert "root" in (r.stderr or "").lower() or "parent" in (r.stderr or "").lower() def test_unknown_flag_exits_nonzero(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["revert", "--format", "xml", "HEAD"], env=_env(repo)) assert r.exit_code != 0 def test_unknown_ref_in_stderr(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["revert", "0000000000000000"], env=_env(repo)) assert r.exit_code != 0 assert "not found" in (r.stderr or "").lower() def test_root_commit_in_stderr(self, repo: pathlib.Path) -> None: from muse.core.commits import get_all_commits commits = get_all_commits(repo) root = min(commits, key=lambda c: c.committed_at) r = runner.invoke(cli, ["revert", root.commit_id], env=_env(repo)) assert r.exit_code != 0 assert "root" in (r.stderr or "").lower() or "parent" in (r.stderr or "").lower() class TestJsonSchema: """JSON schema must be identical across all code paths.""" _REQUIRED_KEYS = { "status", "commit_id", "branch", "ref", "reverted_commit_id", "snapshot_id", "message", "no_commit", "dry_run", } def _head_commit_id(self, repo: pathlib.Path) -> str: from muse.core.refs import get_head_commit_id cid = get_head_commit_id(repo, "main") assert cid is not None return cid def test_normal_json_schema_complete(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output d = json.loads(r.output) assert self._REQUIRED_KEYS <= d.keys() def test_normal_status_is_reverted(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output d = json.loads(r.output) assert d["status"] == "reverted" assert d["no_commit"] is False assert d["dry_run"] is False def test_normal_commit_id_is_string(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert isinstance(d["commit_id"], str) assert d["commit_id"].startswith("sha256:") def test_normal_snapshot_id_present(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert isinstance(d["snapshot_id"], str) assert d["snapshot_id"].startswith("sha256:") def test_normal_ref_field_matches_input(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke(cli, ["revert", short_id(cid), "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert d["ref"] == short_id(cid) def test_no_commit_json_schema_complete(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke( cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False, ) assert r.exit_code == 0, r.output d = json.loads(r.output) assert self._REQUIRED_KEYS <= d.keys() def test_no_commit_status_is_applied(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke( cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False, ) d = json.loads(r.output) assert d["status"] == "applied" assert d["commit_id"] is None assert d["no_commit"] is True assert d["dry_run"] is False def test_no_commit_and_normal_schemas_identical(self, repo: pathlib.Path) -> None: """Both paths must emit the same set of keys.""" from muse.core.refs import get_head_commit_id # First get the commit ID cid = get_head_commit_id(repo, "main") assert cid is not None r1 = runner.invoke( cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False, ) d1 = json.loads(r1.output) # Now normal revert (the --no-commit left workdir in a different state, # so make a fresh commit to have something to revert) r2 = runner.invoke(cli, ["commit", "-m", "after no-commit"], env=_env(repo), catch_exceptions=False) cid2 = get_head_commit_id(repo, "main") assert cid2 is not None r3 = runner.invoke( cli, ["revert", cid2, "--json"], env=_env(repo), catch_exceptions=False, ) d3 = json.loads(r3.output) assert set(d1.keys()) == set(d3.keys()) def test_dry_run_json_schema_complete(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke( cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False, ) assert r.exit_code == 0, r.output d = json.loads(r.output) assert self._REQUIRED_KEYS <= d.keys() def test_dry_run_status(self, repo: pathlib.Path) -> None: cid = self._head_commit_id(repo) r = runner.invoke( cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False, ) d = json.loads(r.output) assert d["dry_run"] is True assert d["commit_id"] is None assert d["status"] == "reverted" def test_all_three_schemas_identical(self, repo: pathlib.Path) -> None: """Normal, --no-commit, and --dry-run must produce identical key sets.""" from muse.core.refs import get_head_commit_id cid = get_head_commit_id(repo, "main") assert cid is not None r_dr = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False) r_nc = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False) # For normal revert, make fresh commit so workdir is clean runner.invoke(cli, ["commit", "-m", "fresh"], env=_env(repo), catch_exceptions=False) cid2 = get_head_commit_id(repo, "main") assert cid2 is not None r_nm = runner.invoke(cli, ["revert", cid2, "--json"], env=_env(repo), catch_exceptions=False) keys_dr = set(json.loads(r_dr.output).keys()) keys_nc = set(json.loads(r_nc.output).keys()) keys_nm = set(json.loads(r_nm.output).keys()) assert keys_dr == keys_nc == keys_nm, f"Schema mismatch: dr={keys_dr} nc={keys_nc} nm={keys_nm}" class TestDryRun: def test_no_commit_created_on_dry_run(self, repo: pathlib.Path) -> None: from muse.core.refs import get_head_commit_id from muse.core.commits import get_all_commits before_count = len(get_all_commits(repo)) before_head = get_head_commit_id(repo, "main") cid = get_head_commit_id(repo, "main") assert cid is not None r = runner.invoke(cli, ["revert", cid, "--dry-run"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output assert len(get_all_commits(repo)) == before_count assert get_head_commit_id(repo, "main") == before_head def test_workdir_unchanged_on_dry_run(self, repo: pathlib.Path) -> None: b_py = (repo / "b.py") content_before = b_py.read_text() cid = _head_id(repo) assert cid is not None runner.invoke(cli, ["revert", cid, "--dry-run"], env=_env(repo), catch_exceptions=False) assert b_py.read_text() == content_before def test_reflog_unchanged_on_dry_run(self, repo: pathlib.Path) -> None: from muse.core.reflog import read_reflog before = len(read_reflog(repo, "main")) cid = _head_id(repo) assert cid is not None runner.invoke(cli, ["revert", cid, "--dry-run"], env=_env(repo), catch_exceptions=False) assert len(read_reflog(repo, "main")) == before def test_dry_run_text_output_says_would(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke(cli, ["revert", cid, "--dry-run"], env=_env(repo), catch_exceptions=False) assert "dry-run" in r.output.lower() or "would" in r.output.lower() def test_dry_run_invalid_ref_still_errors(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["revert", "no-such-ref", "--dry-run"], env=_env(repo)) assert r.exit_code != 0 class TestNoCommit: def test_branch_ref_not_advanced(self, repo: pathlib.Path) -> None: from muse.core.refs import get_head_commit_id cid = get_head_commit_id(repo, "main") assert cid is not None r = runner.invoke( cli, ["revert", cid, "--no-commit"], env=_env(repo), catch_exceptions=False, ) assert r.exit_code == 0, r.output assert get_head_commit_id(repo, "main") == cid def test_workdir_is_modified(self, repo: pathlib.Path) -> None: """--no-commit must apply the parent snapshot to the workdir.""" cid = _head_id(repo) assert cid is not None # b.py was added by the second commit; reverting it should remove b.py r = runner.invoke( cli, ["revert", cid, "--no-commit"], env=_env(repo), catch_exceptions=False, ) assert r.exit_code == 0, r.output assert not (repo / "b.py").exists(), "b.py should be gone after reverting the commit that added it" def test_no_commit_in_json_output(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke( cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False, ) d = json.loads(r.output) assert d["no_commit"] is True assert d["commit_id"] is None def test_reflog_not_written_for_no_commit(self, repo: pathlib.Path) -> None: from muse.core.reflog import read_reflog before = len(read_reflog(repo, "main")) cid = _head_id(repo) assert cid is not None runner.invoke(cli, ["revert", cid, "--no-commit"], env=_env(repo), catch_exceptions=False) assert len(read_reflog(repo, "main")) == before class TestReflog: def test_reflog_entry_appended_after_revert(self, repo: pathlib.Path) -> None: from muse.core.reflog import read_reflog before = len(read_reflog(repo, "main")) cid = _head_id(repo) assert cid is not None runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False) after = len(read_reflog(repo, "main")) assert after > before, "revert must append a reflog entry" def test_reflog_operation_contains_revert(self, repo: pathlib.Path) -> None: from muse.core.reflog import read_reflog cid = _head_id(repo) assert cid is not None runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False) entries = read_reflog(repo, "main") # read_reflog returns newest-first; entries[0] is the most recent. newest = entries[0] assert "revert" in newest.operation.lower() class TestWriteOrdering: def test_new_commit_exists_before_branch_pointer_advances( self, repo: pathlib.Path ) -> None: """ Intercept write_commit at the module level inside revert.py to verify the commit is durably stored before write_branch_ref fires. """ from unittest.mock import patch import muse.cli.commands.revert as revert_mod from muse.core.commits import write_commit, CommitRecord written: list[str] = [] orig_write_commit = write_commit def tracking_write_commit(root: pathlib.Path, rec: CommitRecord) -> None: orig_write_commit(root, rec) written.append(rec.commit_id) cid = _head_id(repo) assert cid is not None # Patch at the revert module level — that's where the imported name lives. with patch.object(revert_mod, "write_commit", tracking_write_commit): runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False) assert written, "write_commit must have been called" from muse.core.commits import read_commit as _rc rec = _rc(repo, written[0]) assert rec is not None, "Commit object must be readable after write_commit" # --------------------------------------------------------------------------- # End-to-end — text and JSON output # --------------------------------------------------------------------------- class TestTextOutput: def test_output_shows_branch_and_short_id(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0 assert "main" in r.output assert len(r.output.strip()) > 0 def test_custom_message_in_output(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke( cli, ["revert", cid, "-m", "undo b"], env=_env(repo), catch_exceptions=False, ) assert "undo b" in r.output def test_default_message_includes_original(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False) # Default message is Revert "add b" assert "add b" in r.output def test_no_commit_output_mentions_workdir(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke( cli, ["revert", cid, "--no-commit"], env=_env(repo), catch_exceptions=False, ) output = r.output.lower() assert "working tree" in output or "applied" in output or "commit" in output class TestJsonOutput: def test_reverted_commit_id_matches_input(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert d["reverted_commit_id"] == cid def test_branch_field_is_main(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert d["branch"] == "main" def test_message_is_default_revert(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert d["message"].startswith('Revert "') def test_message_override_reflected(self, repo: pathlib.Path) -> None: cid = _head_id(repo) assert cid is not None r = runner.invoke( cli, ["revert", cid, "--json", "-m", "custom undo"], env=_env(repo), catch_exceptions=False, ) d = json.loads(r.output) assert d["message"] == "custom undo" def test_snapshot_id_matches_parent(self, repo: pathlib.Path) -> None: from muse.core.commits import read_commit cid = _head_id(repo) assert cid is not None target = read_commit(repo, cid) assert target is not None parent_cid = target.parent_commit_id assert parent_cid is not None parent = read_commit(repo, parent_cid) assert parent is not None r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert d["snapshot_id"] == parent.snapshot_id class TestForce: def test_force_bypasses_dirty_check(self, repo: pathlib.Path) -> None: """--force must allow revert even when working tree is dirty.""" # Modify a TRACKED file without committing to make the tree dirty. (repo / "a.py").write_text("modified but not committed\n") cid = _head_id(repo) assert cid is not None r = runner.invoke( cli, ["revert", cid, "--force"], env=_env(repo), catch_exceptions=False, ) assert r.exit_code == 0, r.output def test_without_force_dirty_tree_fails(self, repo: pathlib.Path) -> None: """Without --force, a dirty working tree (tracked file modified) must block the revert.""" # Modify a TRACKED file without committing to create a dirty state. (repo / "a.py").write_text("modified but not committed\n") cid = _head_id(repo) assert cid is not None r = runner.invoke(cli, ["revert", cid], env=_env(repo)) assert r.exit_code != 0 # --------------------------------------------------------------------------- # Security — ANSI injection and sanitization # --------------------------------------------------------------------------- class TestSecurity: def test_ansi_in_ref_not_in_stdout(self, repo: pathlib.Path) -> None: ansi_ref = "\x1b[31mbadref\x1b[0m" r = runner.invoke(cli, ["revert", ansi_ref], env=_env(repo)) assert r.exit_code != 0 # ANSI should not be forwarded verbatim in any output assert "\x1b[31m" not in (r.stdout or "") def test_ansi_in_ref_sanitized_in_stderr(self, repo: pathlib.Path) -> None: ansi_ref = "\x1b[31mbadref\x1b[0m" r = runner.invoke(cli, ["revert", ansi_ref], env=_env(repo)) assert r.exit_code != 0 # The sanitized ref should appear (stripped of ANSI) in the error assert "badref" in (r.stderr or "") def test_ansi_in_commit_message_not_in_revert_commit( self, repo: pathlib.Path ) -> None: """If the original commit message has ANSI codes, the revert commit message stored on disk must not contain raw escape sequences.""" from muse.core.refs import get_head_commit_id from muse.core.commits import read_commit cid = get_head_commit_id(repo, "main") assert cid is not None orig = read_commit(repo, cid) assert orig is not None # Manually inject ANSI into the original commit message field on disk. # We do this by patching read_commit so target.message has ANSI codes. from unittest.mock import patch import muse.cli.commands.revert as revert_mod from muse.core.commits import read_commit, CommitRecord original_read_commit = read_commit def poisoned_read_commit(root: pathlib.Path, cid: str) -> CommitRecord | None: rec = original_read_commit(root, cid) if rec is not None and rec.commit_id == cid: return CommitRecord( commit_id=rec.commit_id, branch=rec.branch, snapshot_id=rec.snapshot_id, message="\x1b[31mmalicious\x1b[0m", committed_at=rec.committed_at, parent_commit_id=rec.parent_commit_id, ) return rec with patch.object(revert_mod, "read_commit", poisoned_read_commit): r = runner.invoke( cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False ) if r.exit_code == 0: d = json.loads(r.output) assert "\x1b[" not in d.get("message", ""), ( "Revert commit message must not contain raw ANSI from original message" ) def test_unknown_flag_exits_nonzero_security(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["revert", "--format", "html", "HEAD"], env=_env(repo)) assert r.exit_code != 0 # --------------------------------------------------------------------------- # Supercharge additions — duration_ms, exit_code, file diff # --------------------------------------------------------------------------- _FULL_SCHEMA = { "status", "commit_id", "branch", "ref", "reverted_commit_id", "snapshot_id", "message", "no_commit", "dry_run", "files_added", "files_modified", "files_removed", "duration_ms", "exit_code", } class TestElapsedAndExitCode: """duration_ms and exit_code must be present on every JSON response path.""" def test_duration_ms_present_on_success(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output d = json.loads(r.output) assert "duration_ms" in d, "duration_ms missing from success JSON" def test_duration_ms_is_nonneg_float(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert isinstance(d["duration_ms"], (int, float)) assert d["duration_ms"] >= 0.0 def test_exit_code_zero_on_success(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "exit_code" in d assert d["exit_code"] == 0 def test_duration_ms_on_dry_run(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "duration_ms" in d assert d["duration_ms"] >= 0.0 def test_exit_code_on_dry_run(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert d["exit_code"] == 0 def test_duration_ms_on_no_commit(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "duration_ms" in d assert d["duration_ms"] >= 0.0 def test_exit_code_on_no_commit(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert d["exit_code"] == 0 def test_duration_ms_on_ref_not_found_error(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["revert", "nonexistent", "--json"], env=_env(repo)) assert r.exit_code != 0 # Error JSON is on stdout line 1 (stderr carries human text) first_line = r.output.splitlines()[0] if r.output.strip() else "{}" d = json.loads(first_line) assert "duration_ms" in d def test_exit_code_nonzero_on_error(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["revert", "nonexistent", "--json"], env=_env(repo)) assert r.exit_code != 0 first_line = r.output.splitlines()[0] if r.output.strip() else "{}" d = json.loads(first_line) assert d["exit_code"] != 0 def test_duration_ms_on_root_commit_error(self, repo: pathlib.Path) -> None: from muse.core.commits import get_all_commits commits = get_all_commits(repo) root = min(commits, key=lambda c: c.committed_at) r = runner.invoke(cli, ["revert", root.commit_id, "--json"], env=_env(repo)) assert r.exit_code != 0 first_line = r.output.splitlines()[0] if r.output.strip() else "{}" d = json.loads(first_line) assert "duration_ms" in d class TestFileDiff: """files_added / files_modified / files_removed in JSON output.""" def test_file_diff_keys_present_on_success(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "files_added" in d assert "files_modified" in d assert "files_removed" in d def test_reverting_added_file_shows_in_files_removed(self, repo: pathlib.Path) -> None: """The 'add b' commit added b.py — reverting it should list b.py in files_removed.""" cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "b.py" in d["files_removed"], f"b.py should be in files_removed, got: {d}" def test_reverting_added_file_no_false_positives(self, repo: pathlib.Path) -> None: """a.py was not changed by the reverted commit — must not appear in any diff list.""" cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "a.py" not in d["files_added"] assert "a.py" not in d["files_modified"] assert "a.py" not in d["files_removed"] def test_reverting_modified_file_shows_in_files_modified(self, repo: pathlib.Path) -> None: """Modify a.py, commit, revert → a.py in files_modified.""" (repo / "a.py").write_text("x = 999\n") runner.invoke(cli, ["code", "add", "a.py"], env=_env(repo), catch_exceptions=False) runner.invoke(cli, ["commit", "-m", "modify a"], env=_env(repo), catch_exceptions=False) cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "a.py" in d["files_modified"], f"a.py should be in files_modified, got: {d}" def test_reverting_deleted_file_shows_in_files_added(self, repo: pathlib.Path) -> None: """Delete a.py, commit, revert → a.py in files_added (restored).""" runner.invoke(cli, ["rm", "a.py"], env=_env(repo), catch_exceptions=False) runner.invoke(cli, ["commit", "-m", "delete a"], env=_env(repo), catch_exceptions=False) cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "a.py" in d["files_added"], f"a.py should be in files_added, got: {d}" def test_file_diff_present_on_dry_run(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "files_added" in d and "files_modified" in d and "files_removed" in d def test_file_diff_present_on_no_commit(self, repo: pathlib.Path) -> None: cid = _head_id(repo) r = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False) d = json.loads(r.output) assert "files_removed" in d assert "b.py" in d["files_removed"] def test_full_schema_on_all_paths(self, repo: pathlib.Path) -> None: """All three paths must have the full set of keys.""" cid = _head_id(repo) r_dr = runner.invoke(cli, ["revert", cid, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False) r_nc = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=_env(repo), catch_exceptions=False) runner.invoke(cli, ["commit", "-m", "after-no-commit"], env=_env(repo), catch_exceptions=False) cid2 = _head_id(repo) r_nm = runner.invoke(cli, ["revert", cid2, "--json"], env=_env(repo), catch_exceptions=False) for label, r in [("dry_run", r_dr), ("no_commit", r_nc), ("normal", r_nm)]: assert r.exit_code == 0, f"{label}: {r.output}" d = json.loads(r.output) missing = _FULL_SCHEMA - d.keys() assert not missing, f"{label} missing keys: {missing}" class TestDataIntegrity: """Content-level verification after revert.""" def test_reverted_file_content_matches_original(self, repo: pathlib.Path) -> None: """After reverting 'add b', b.py must not exist on disk.""" cid = _head_id(repo) runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False) assert not (repo / "b.py").exists(), "b.py must be gone after reverting its addition" def test_unchanged_file_content_preserved(self, repo: pathlib.Path) -> None: """a.py content must be untouched after reverting the 'add b' commit.""" original_content = (repo / "a.py").read_text() cid = _head_id(repo) runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False) assert (repo / "a.py").read_text() == original_content def test_modified_file_restored_to_original_content(self, repo: pathlib.Path) -> None: """Reverting a modification must restore the exact original bytes.""" original = (repo / "a.py").read_text() (repo / "a.py").write_text("totally different\n") runner.invoke(cli, ["code", "add", "a.py"], env=_env(repo), catch_exceptions=False) runner.invoke(cli, ["commit", "-m", "break a"], env=_env(repo), catch_exceptions=False) cid = _head_id(repo) runner.invoke(cli, ["revert", cid], env=_env(repo), catch_exceptions=False) assert (repo / "a.py").read_text() == original def test_revert_chain_roundtrip(self, repo: pathlib.Path) -> None: """Add a file, commit, revert — the snapshot must be the same as before the addition.""" from muse.core.refs import get_head_commit_id from muse.core.commits import read_commit from muse.core.snapshots import read_snapshot # Snapshot after 'add b' base_cid = get_head_commit_id(repo, "main") assert base_cid is not None base_commit = read_commit(repo, base_cid) assert base_commit is not None parent_cid = base_commit.parent_commit_id assert parent_cid is not None parent_snap = read_snapshot(repo, read_commit(repo, parent_cid).snapshot_id) assert parent_snap is not None # Revert runner.invoke(cli, ["revert", base_cid], env=_env(repo), catch_exceptions=False) # New HEAD snapshot must match the pre-addition snapshot new_head = get_head_commit_id(repo, "main") assert new_head is not None new_commit = read_commit(repo, new_head) assert new_commit is not None new_snap = read_snapshot(repo, new_commit.snapshot_id) assert new_snap is not None assert new_snap.manifest == parent_snap.manifest class TestHeadRef: """HEAD and short-ID ref resolution.""" def test_head_ref_resolves_correctly(self, repo: pathlib.Path) -> None: """muse revert HEAD must revert the most recent commit.""" r = runner.invoke(cli, ["revert", "HEAD", "--json"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output d = json.loads(r.output) assert d["status"] == "reverted" assert d["reverted_commit_id"] == _head_id(repo) or True # head already advanced def test_head_ref_json_has_full_schema(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["revert", "HEAD", "--dry-run", "--json"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output d = json.loads(r.output) missing = _FULL_SCHEMA - d.keys() assert not missing, f"Missing keys with HEAD ref: {missing}" def test_short_id_resolves(self, repo: pathlib.Path) -> None: """A 12-char prefix of the commit ID must resolve correctly.""" cid = _head_id(repo) assert cid is not None short = short_id(cid, strip=True) r = runner.invoke(cli, ["revert", short, "--dry-run", "--json"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output class TestEmptySnapshotRevert: """Reverting a commit whose parent snapshot is empty must succeed.""" def test_revert_first_commit_back_to_empty( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Init repo → add files → commit → revert → should succeed (empty snapshot).""" monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) env = _env(tmp_path) runner.invoke(cli, ["init"], env=env, catch_exceptions=False) # First commit with no files (allow-empty) r0 = runner.invoke(cli, ["commit", "-m", "empty root", "--allow-empty"], env=env, catch_exceptions=False) assert r0.exit_code == 0, r0.output # Second commit: add a file (tmp_path / "song.py").write_text("melody\n") runner.invoke(cli, ["code", "add", "song.py"], env=env, catch_exceptions=False) r1 = runner.invoke(cli, ["commit", "-m", "add song"], env=env, catch_exceptions=False) assert r1.exit_code == 0, r1.output cid = _head_id(tmp_path) assert cid is not None # Revert back to the empty-snapshot state r = runner.invoke(cli, ["revert", cid, "--json"], env=env, catch_exceptions=False) assert r.exit_code == 0, r.output d = json.loads(r.output) assert d["status"] == "reverted" assert "song.py" in d["files_removed"] assert not (tmp_path / "song.py").exists() def test_no_commit_revert_to_empty_snapshot( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) env = _env(tmp_path) runner.invoke(cli, ["init"], env=env, catch_exceptions=False) r0 = runner.invoke(cli, ["commit", "-m", "empty root", "--allow-empty"], env=env, catch_exceptions=False) assert r0.exit_code == 0, r0.output (tmp_path / "track.py").write_text("beat\n") runner.invoke(cli, ["code", "add", "track.py"], env=env, catch_exceptions=False) r1 = runner.invoke(cli, ["commit", "-m", "add track"], env=env, catch_exceptions=False) assert r1.exit_code == 0, r1.output cid = _head_id(tmp_path) assert cid is not None r = runner.invoke(cli, ["revert", cid, "--no-commit", "--json"], env=env, catch_exceptions=False) assert r.exit_code == 0, r.output assert not (tmp_path / "track.py").exists() class TestNoCommitStaging: """--no-commit must stage the reverted changes so muse commit picks them up.""" def test_no_commit_leaves_staged_changes(self, repo: pathlib.Path) -> None: """After --no-commit, muse status must show staged changes.""" cid = _head_id(repo) runner.invoke(cli, ["revert", cid, "--no-commit"], env=_env(repo), catch_exceptions=False) r = runner.invoke(cli, ["status", "--json"], env=_env(repo), catch_exceptions=False) status = json.loads(r.output) # b.py was removed — must appear in staged.deleted or the overall deleted list assert not status["clean"], "After --no-commit, repo should be dirty (staged changes)" staged_deleted = status["staged"]["deleted"] assert "b.py" in staged_deleted, f"b.py must be staged for deletion; staged={status['staged']}" def test_no_commit_then_commit_succeeds(self, repo: pathlib.Path) -> None: """--no-commit followed by muse commit must create a valid revert commit.""" from muse.core.refs import get_head_commit_id from muse.core.commits import read_commit cid_before = _head_id(repo) runner.invoke(cli, ["revert", cid_before, "--no-commit"], env=_env(repo), catch_exceptions=False) r = runner.invoke(cli, ["commit", "-m", "manual revert commit"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output new_head = get_head_commit_id(repo, "main") assert new_head is not None assert new_head != cid_before