"""Comprehensive tests for ``muse code reset``. Review findings addressed -------------------------- Performance bug fixed * write_stage was called unconditionally even when no files were actually unstaged (all-"not staged" case). Now skipped when to_unstage is empty. Security * Paths outside the repo root now produce an explicit warning and are reported in the not_staged list instead of silently falling back to treating the raw string as a stage key. New capabilities (added this review) * ``-n`` / ``--dry-run`` — preview what would be unstaged, no write. * Glob pattern support — ``muse code reset '*.py'`` unstages all staged Python files matching the pattern. * ``not_staged`` key in JSON output — agents can see which files were requested but not found in the stage (enables partial-failure detection). Test categories --------------- I Core behaviour — clear all, specific file, HEAD syntax, nothing staged. II JSON output — all keys present, not_staged populated, dry_run flag. III Dry-run — no writes, correct preview output, JSON dry_run=true. IV Glob patterns — *.ext, ?, partial match, no match. V Security — out-of-root path, path traversal, sanitize_display. VI Performance — no write when nothing unstaged. VII Edge cases — duplicate paths, mixed staged/not-staged, fresh repo. VIII Stress — 500 files, 100 reset-cycles, multi-pattern glob. """ from __future__ import annotations import json import pathlib import pytest from muse.plugins.code.stage import StagedEntry, StagedFileMap, read_stage, stage_path, write_stage, make_entry from muse.core._types import Manifest from tests.cli_test_helper import CliRunner runner = CliRunner() cli = None # --------------------------------------------------------------------------- # Helpers and fixtures # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(root)} def _run(root: pathlib.Path, *args: str) -> tuple[int, str]: result = runner.invoke(cli, list(args), env=_env(root), catch_exceptions=False) return result.exit_code, result.output def _run_unchecked(root: pathlib.Path, *args: str) -> tuple[int, str]: result = runner.invoke(cli, list(args), env=_env(root)) return result.exit_code, result.output @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Fresh code-domain repo with main.py + models.py committed.""" monkeypatch.chdir(tmp_path) r = runner.invoke(cli, ["init", "--domain", "code"], env=_env(tmp_path)) assert r.exit_code == 0, r.output (tmp_path / "main.py").write_text("x = 1\n") (tmp_path / "models.py").write_text("class User: pass\n") r2 = runner.invoke(cli, ["commit", "--allow-empty", "-m", "init"], env=_env(tmp_path)) assert r2.exit_code == 0, r2.output return tmp_path def _stage(repo: pathlib.Path, *filenames: str) -> None: """Write modified content and stage the given files.""" for fn in filenames: path = repo / fn if not path.exists(): path.write_text(f"# new: {fn}\n") else: path.write_text(path.read_text() + "# modified\n") _run(repo, "code", "add", "-A") # =========================================================================== # I Core behaviour # =========================================================================== class TestCoreBehaviourI: """Baseline correctness: clear-all, specific-file, HEAD syntax.""" def test_I1_reset_all_clears_entire_stage(self, repo: pathlib.Path) -> None: """I1: reset with no args clears every staged file.""" _stage(repo, "main.py", "models.py") assert stage_path(repo).exists() code, out = _run(repo, "code", "reset") assert code == 0, out assert not stage_path(repo).exists() assert read_stage(repo) == {} def test_I2_reset_specific_file_leaves_others( self, repo: pathlib.Path ) -> None: """I2: resetting one file leaves other staged files intact.""" _stage(repo, "main.py", "models.py") code, out = _run(repo, "code", "reset", "main.py") assert code == 0, out stage = read_stage(repo) assert "main.py" not in stage assert "models.py" in stage def test_I3_reset_HEAD_syntax_works(self, repo: pathlib.Path) -> None: """I3: 'muse code reset HEAD ' is an alias for 'muse code reset '.""" _stage(repo, "main.py") code, out = _run(repo, "code", "reset", "HEAD", "main.py") assert code == 0, out assert not stage_path(repo).exists() def test_I4_reset_when_nothing_staged(self, repo: pathlib.Path) -> None: """I4: reset with no stage prints 'Nothing staged' and exits 0.""" code, out = _run(repo, "code", "reset") assert code == 0, out assert "Nothing staged" in out def test_I5_reset_not_staged_file_exits_zero( self, repo: pathlib.Path ) -> None: """I5: resetting a file that is not staged exits 0 and says 'not staged'.""" _stage(repo, "main.py") code, out = _run(repo, "code", "reset", "models.py") assert code == 0, out assert "not staged" in out # main.py must still be staged. assert "main.py" in read_stage(repo) def test_I6_reset_multiple_files_in_one_invocation( self, repo: pathlib.Path ) -> None: """I6: multiple files can be unstaged in a single command.""" _stage(repo, "main.py", "models.py") code, out = _run(repo, "code", "reset", "main.py", "models.py") assert code == 0, out assert read_stage(repo) == {} def test_I7_reset_preserves_working_tree(self, repo: pathlib.Path) -> None: """I7: unstaging does NOT revert the working-tree file.""" (repo / "main.py").write_text("changed content\n") # Stage directly without using _stage helper (which modifies content again). _run(repo, "code", "add", "main.py") _run(repo, "code", "reset", "main.py") assert (repo / "main.py").read_text() == "changed content\n" def test_I8_text_output_shows_unstaged_per_file( self, repo: pathlib.Path ) -> None: """I8: text output lists each unstaged file.""" _stage(repo, "main.py", "models.py") _, out = _run(repo, "code", "reset", "main.py") assert "main.py" in out def test_I9_reset_HEAD_alone_clears_all(self, repo: pathlib.Path) -> None: """I9: 'muse code reset HEAD' (no file) clears everything — HEAD is stripped.""" _stage(repo, "main.py", "models.py") code, out = _run(repo, "code", "reset", "HEAD") assert code == 0, out assert read_stage(repo) == {} # =========================================================================== # II JSON output # =========================================================================== class TestJsonOutputII: """``--format json`` must emit complete, accurate JSON.""" def test_II1_json_unstaged_specific_file( self, repo: pathlib.Path ) -> None: """II1: JSON output contains unstaged count, files list, and not_staged.""" _stage(repo, "main.py") _, out = _run(repo, "code", "reset", "--format", "json", "main.py") data = json.loads(out.strip()) assert data["unstaged"] == 1 assert "main.py" in data["files"] assert data["not_staged"] == [] assert data["dry_run"] is False def test_II2_json_not_staged_populated_for_missing_file( self, repo: pathlib.Path ) -> None: """II2: not_staged is populated when a requested file is not in the stage.""" _stage(repo, "main.py") _, out = _run( repo, "code", "reset", "--format", "json", "main.py", "ghost.py" ) data = json.loads(out.strip()) assert data["unstaged"] == 1 assert "main.py" in data["files"] assert "ghost.py" in data["not_staged"] def test_II3_json_clear_all_has_all_keys( self, repo: pathlib.Path ) -> None: """II3: clear-all (no args) JSON includes all required keys.""" _stage(repo, "main.py", "models.py") _, out = _run(repo, "code", "reset", "--format", "json") data = json.loads(out.strip()) assert "unstaged" in data assert "files" in data assert "not_staged" in data assert "dry_run" in data assert data["unstaged"] == 2 assert data["not_staged"] == [] def test_II4_json_nothing_staged(self, repo: pathlib.Path) -> None: """II4: nothing staged returns unstaged=0, files=[], not_staged=[].""" _, out = _run(repo, "code", "reset", "--format", "json") data = json.loads(out.strip()) assert data["unstaged"] == 0 assert data["files"] == [] assert data["not_staged"] == [] assert data["dry_run"] is False def test_II5_json_output_is_valid_json(self, repo: pathlib.Path) -> None: """II5: output is always parseable JSON regardless of outcome.""" _stage(repo, "main.py") _, out = _run(repo, "code", "reset", "--format", "json", "main.py") json.loads(out.strip()) # must not raise def test_II6_json_all_not_staged_when_none_found( self, repo: pathlib.Path ) -> None: """II6: when all requested files are not staged, files=[], all in not_staged.""" _stage(repo, "main.py") _, out = _run( repo, "code", "reset", "--format", "json", "ghost1.py", "ghost2.py" ) data = json.loads(out.strip()) assert data["unstaged"] == 0 assert data["files"] == [] assert "ghost1.py" in data["not_staged"] assert "ghost2.py" in data["not_staged"] def test_II7_json_glob_results_included(self, repo: pathlib.Path) -> None: """II7: glob results appear in JSON files list.""" _stage(repo, "main.py", "models.py") _, out = _run(repo, "code", "reset", "--format", "json", "*.py") data = json.loads(out.strip()) assert data["unstaged"] >= 2 assert "main.py" in data["files"] assert "models.py" in data["files"] # =========================================================================== # III Dry-run # =========================================================================== class TestDryRunIII: """``-n`` / ``--dry-run`` must preview without writing.""" def test_III1_dry_run_does_not_remove_from_stage( self, repo: pathlib.Path ) -> None: """III1: after dry-run, stage is unchanged.""" _stage(repo, "main.py") before = dict(read_stage(repo)) _run(repo, "code", "reset", "-n", "main.py") assert read_stage(repo) == before def test_III2_dry_run_lists_would_unstage( self, repo: pathlib.Path ) -> None: """III2: dry-run text output mentions files that would be unstaged.""" _stage(repo, "main.py", "models.py") _, out = _run(repo, "code", "reset", "--dry-run", "main.py") assert "main.py" in out assert "would" in out.lower() def test_III3_dry_run_json_flag_is_true( self, repo: pathlib.Path ) -> None: """III3: dry_run field is true in JSON output when -n given.""" _stage(repo, "main.py") _, out = _run( repo, "code", "reset", "-n", "--format", "json", "main.py" ) data = json.loads(out.strip()) assert data["dry_run"] is True assert data["unstaged"] == 1 assert "main.py" in data["files"] def test_III4_dry_run_clear_all_is_preview_only( self, repo: pathlib.Path ) -> None: """III4: dry-run on clear-all leaves stage intact.""" _stage(repo, "main.py", "models.py") _run(repo, "code", "reset", "--dry-run") assert stage_path(repo).exists() assert "main.py" in read_stage(repo) def test_III5_dry_run_nothing_staged(self, repo: pathlib.Path) -> None: """III5: dry-run when nothing staged exits 0 and reports 0.""" code, out = _run(repo, "code", "reset", "-n") assert code == 0, out assert "Nothing staged" in out def test_III6_dry_run_with_json_nothing_staged( self, repo: pathlib.Path ) -> None: """III6: dry-run + JSON + nothing staged emits valid JSON with zeros.""" _, out = _run(repo, "code", "reset", "-n", "--format", "json") data = json.loads(out.strip()) assert data["unstaged"] == 0 assert data["dry_run"] is True # =========================================================================== # IV Glob patterns # =========================================================================== class TestGlobPatternsIV: """Glob patterns must match staged paths using fnmatch semantics.""" def test_IV1_star_ext_unstages_matching_files( self, repo: pathlib.Path ) -> None: """IV1: '*.py' unstages all staged Python files.""" _stage(repo, "main.py", "models.py") code, out = _run(repo, "code", "reset", "*.py") assert code == 0, out assert read_stage(repo) == {} def test_IV2_star_ext_leaves_non_matching_staged( self, repo: pathlib.Path ) -> None: """IV2: '*.py' does not unstage non-Python staged files.""" _stage(repo, "main.py") # Manually add a .txt file to the stage. stage = dict(read_stage(repo)) stage["readme.txt"] = make_entry("a" * 64, "A") write_stage(repo, stage) _run(repo, "code", "reset", "*.py") remaining = read_stage(repo) assert "main.py" not in remaining assert "readme.txt" in remaining def test_IV3_glob_with_directory_prefix( self, repo: pathlib.Path ) -> None: """IV3: 'src/*.py' matches only files inside src/.""" # Manually build a stage with nested paths. stage: StagedFileMap = { "src/auth.py": make_entry("a" * 64, "A"), "src/models.py": make_entry("b" * 64, "A"), "top.py": make_entry("c" * 64, "M"), } write_stage(repo, stage) _run(repo, "code", "reset", "src/*.py") remaining = read_stage(repo) assert "src/auth.py" not in remaining assert "src/models.py" not in remaining assert "top.py" in remaining def test_IV4_glob_no_match_goes_to_not_staged( self, repo: pathlib.Path ) -> None: """IV4: glob with no matches appears in not_staged in JSON output.""" _stage(repo, "main.py") _, out = _run( repo, "code", "reset", "--format", "json", "*.toml" ) data = json.loads(out.strip()) assert data["unstaged"] == 0 assert "*.toml" in data["not_staged"] def test_IV5_question_mark_glob(self, repo: pathlib.Path) -> None: """IV5: '?.py' matches single-char filenames.""" stage: StagedFileMap = { "a.py": make_entry("a" * 64, "A"), "ab.py": make_entry("b" * 64, "A"), } write_stage(repo, stage) _run(repo, "code", "reset", "?.py") remaining = read_stage(repo) assert "a.py" not in remaining assert "ab.py" in remaining def test_IV6_multiple_globs_in_one_invocation( self, repo: pathlib.Path ) -> None: """IV6: multiple glob patterns can be combined in one reset call.""" stage: StagedFileMap = { "main.py": make_entry("a" * 64, "M"), "config.toml": make_entry("b" * 64, "M"), "readme.md": make_entry("c" * 64, "M"), } write_stage(repo, stage) _run(repo, "code", "reset", "*.py", "*.toml") remaining = read_stage(repo) assert "main.py" not in remaining assert "config.toml" not in remaining assert "readme.md" in remaining def test_IV7_literal_path_and_glob_combined( self, repo: pathlib.Path ) -> None: """IV7: literal path and glob can be combined in one invocation.""" _stage(repo, "main.py", "models.py") _, out = _run( repo, "code", "reset", "--format", "json", "main.py", "*.py" ) data = json.loads(out.strip()) # Both main.py and models.py should be unstaged. assert "main.py" in data["files"] assert "models.py" in data["files"] # =========================================================================== # V Security # =========================================================================== class TestSecurityV: """Out-of-root paths must be rejected with a warning, not silently used.""" def test_V1_path_outside_repo_goes_to_not_staged( self, repo: pathlib.Path ) -> None: """V1: an absolute path outside the repo appears in not_staged, not files.""" _stage(repo, "main.py") outside = str(repo.parent / "outside.py") _, out = _run_unchecked( repo, "code", "reset", "--format", "json", outside ) data = json.loads(out.strip()) assert data["unstaged"] == 0 assert data["files"] == [] assert any("outside" in p for p in data["not_staged"]) def test_V2_dotdot_traversal_rejected(self, repo: pathlib.Path) -> None: """V2: '../secret.py' is treated as outside-root and lands in not_staged.""" _stage(repo, "main.py") _, out = _run(repo, "code", "reset", "--format", "json", "../secret.py") data = json.loads(out.strip()) assert data["unstaged"] == 0 # main.py must remain staged — traversal did not corrupt state. assert "main.py" in read_stage(repo) def test_V3_out_of_root_does_not_corrupt_stage( self, repo: pathlib.Path ) -> None: """V3: out-of-root path does not modify any staged entry.""" _stage(repo, "main.py", "models.py") before = dict(read_stage(repo)) _run_unchecked(repo, "code", "reset", str(repo.parent / "evil.py")) assert read_stage(repo) == before # =========================================================================== # VI Performance — no write when nothing unstaged # =========================================================================== class TestPerformanceVI: """write_stage must NOT be called when nothing was actually unstaged.""" def test_VI1_stage_file_mtime_unchanged_when_nothing_unstaged( self, repo: pathlib.Path ) -> None: """VI1: stage.msgpack mtime stays the same when no files are unstaged.""" import time _stage(repo, "main.py") path = stage_path(repo) mtime_before = path.stat().st_mtime time.sleep(0.05) # ensure mtime would differ if written _run(repo, "code", "reset", "ghost.py") mtime_after = path.stat().st_mtime assert mtime_before == mtime_after, ( "Stage file was rewritten despite nothing being unstaged" ) def test_VI2_stage_file_mtime_changes_when_something_unstaged( self, repo: pathlib.Path ) -> None: """VI2: stage.msgpack mtime changes when a file is actually unstaged.""" import time _stage(repo, "main.py", "models.py") path = stage_path(repo) mtime_before = path.stat().st_mtime time.sleep(0.05) _run(repo, "code", "reset", "main.py") mtime_after = path.stat().st_mtime assert mtime_after > mtime_before, ( "Stage file was not rewritten after unstaging a file" ) def test_VI3_dry_run_never_writes(self, repo: pathlib.Path) -> None: """VI3: dry-run never touches the stage file.""" import time _stage(repo, "main.py") path = stage_path(repo) mtime_before = path.stat().st_mtime time.sleep(0.05) _run(repo, "code", "reset", "--dry-run", "main.py") assert path.stat().st_mtime == mtime_before # =========================================================================== # VII Edge cases # =========================================================================== class TestEdgeCasesVII: """Fresh repo, duplicate paths, and partial-staged scenarios.""" def test_VII1_fresh_repo_no_commits( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """VII1: reset works on a repo with no commits.""" monkeypatch.chdir(tmp_path) runner.invoke(cli, ["init", "--domain", "code"], env=_env(tmp_path)) (tmp_path / "f.py").write_text("x = 1\n") _run(tmp_path, "code", "add", "f.py") code, out = _run(tmp_path, "code", "reset", "f.py") assert code == 0, out assert read_stage(tmp_path) == {} def test_VII2_duplicate_paths_not_double_counted( self, repo: pathlib.Path ) -> None: """VII2: passing the same path twice only unstages it once.""" _stage(repo, "main.py") _, out = _run( repo, "code", "reset", "--format", "json", "main.py", "main.py" ) data = json.loads(out.strip()) assert data["unstaged"] == 1 def test_VII3_reset_file_then_restage_works( self, repo: pathlib.Path ) -> None: """VII3: unstage then restage a file produces a clean round-trip.""" (repo / "main.py").write_text("v1\n") _run(repo, "code", "add", "main.py") _run(repo, "code", "reset", "main.py") assert not stage_path(repo).exists() (repo / "main.py").write_text("v2\n") _run(repo, "code", "add", "main.py") assert "main.py" in read_stage(repo) def test_VII4_reset_all_then_readd_works( self, repo: pathlib.Path ) -> None: """VII4: clear stage then re-add produces a fresh stage.""" _stage(repo, "main.py") _run(repo, "code", "reset") assert read_stage(repo) == {} _stage(repo, "models.py") assert "models.py" in read_stage(repo) def test_VII5_reset_after_commit_clears_nothing( self, repo: pathlib.Path ) -> None: """VII5: after a commit the stage is clear — reset reports 'Nothing staged'.""" _stage(repo, "main.py") _run(repo, "commit", "-m", "committed") assert not stage_path(repo).exists() code, out = _run(repo, "code", "reset") assert code == 0 assert "Nothing staged" in out def test_VII6_reset_mode_D_staged_deletion( self, repo: pathlib.Path ) -> None: """VII6: a staged deletion (mode D) can be unstaged.""" (repo / "models.py").unlink() _run(repo, "code", "add", "-u") assert "models.py" in read_stage(repo) _run(repo, "code", "reset", "models.py") assert "models.py" not in read_stage(repo) def test_VII7_json_and_text_formats_consistent( self, repo: pathlib.Path ) -> None: """VII7: JSON and text modes agree on what was unstaged.""" _stage(repo, "main.py", "models.py") # Reset in JSON mode and capture what was unstaged. _, out_json = _run( repo, "code", "add", "-A" ) # Re-stage after the read above. _stage(repo, "main.py", "models.py") _, out_j = _run(repo, "code", "reset", "--format", "json", "main.py") data = json.loads(out_j.strip()) assert data["unstaged"] == 1 # Re-stage again and reset in text mode. _stage(repo, "main.py") _, out_t = _run(repo, "code", "reset", "main.py") assert "main.py" in out_t # =========================================================================== # VIII Stress # =========================================================================== class TestStressVIII: """High-volume and repeated-operation scenarios.""" def test_VIII1_reset_500_staged_files(self, repo: pathlib.Path) -> None: """VIII1: unstaging 500 files in one shot clears the stage completely.""" for i in range(500): (repo / f"f_{i:04d}.py").write_text(f"X = {i}\n") _run(repo, "code", "add", "-A") assert len(read_stage(repo)) >= 500 code, out = _run(repo, "code", "reset") assert code == 0, out assert read_stage(repo) == {} def test_VIII2_glob_reset_500_files(self, repo: pathlib.Path) -> None: """VIII2: glob '*.py' unstages 500 staged Python files.""" for i in range(500): (repo / f"m_{i:04d}.py").write_text(f"V = {i}\n") _run(repo, "code", "add", "-A") _, out = _run(repo, "code", "reset", "--format", "json", "*.py") data = json.loads(out.strip()) assert data["unstaged"] >= 500 assert read_stage(repo) == {} def test_VIII3_100_reset_cycles_leave_clean_stage( self, repo: pathlib.Path ) -> None: """VIII3: 100 add/reset cycles leave a clean stage every time.""" for cycle in range(100): # Use content that never matches the committed version (x = 1). (repo / "main.py").write_text(f"# cycle {cycle:04d}\nx = {cycle + 1000}\n") _run(repo, "code", "add", "main.py") assert stage_path(repo).exists(), f"Cycle {cycle}: stage not written" _run(repo, "code", "reset") assert not stage_path(repo).exists(), f"Cycle {cycle}: stage not cleared" def test_VIII4_dry_run_on_500_staged_files_accurate( self, repo: pathlib.Path ) -> None: """VIII4: dry-run JSON count matches actual stage size.""" for i in range(500): (repo / f"d_{i:04d}.py").write_text(f"Z = {i}\n") _run(repo, "code", "add", "-A") stage_size = len(read_stage(repo)) _, out = _run(repo, "code", "reset", "-n", "--format", "json") data = json.loads(out.strip()) assert data["dry_run"] is True assert data["unstaged"] == stage_size # Stage must be untouched. assert len(read_stage(repo)) == stage_size def test_VIII5_multi_pattern_glob_500_files( self, repo: pathlib.Path ) -> None: """VIII5: multiple globs combined unstage all matching files.""" for i in range(250): (repo / f"py_{i:03d}.py").write_text(f"A = {i}\n") (repo / f"md_{i:03d}.md").write_text(f"# doc {i}\n") _run(repo, "code", "add", "-A") _, out = _run( repo, "code", "reset", "--format", "json", "*.py", "*.md" ) data = json.loads(out.strip()) assert data["unstaged"] >= 500 assert read_stage(repo) == {}