gabriel / muse public
test_cmd_reset_revert.py python
185 lines 7.5 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 126 days ago
1 """Comprehensive tests for ``muse reset`` and ``muse revert``.
2
3 Covers:
4 - reset: --soft / --hard / --mixed, HEAD~N syntax
5 - revert: revert a specific commit
6 - Security: reject path-traversal commit refs
7 - Stress: reset across many commits
8 """
9
10 from __future__ import annotations
11
12 import datetime
13 import json
14 import pathlib
15
16 import pytest
17 from tests.cli_test_helper import CliRunner
18 from muse.core.types import fake_id
19 from muse.core.paths import heads_dir, muse_dir
20
21 cli = None # argparse migration — CliRunner ignores this arg
22
23 runner = CliRunner()
24
25
26 # ---------------------------------------------------------------------------
27 # Shared helpers
28 # ---------------------------------------------------------------------------
29
30 def _env(root: pathlib.Path) -> Manifest:
31 return {"MUSE_REPO_ROOT": str(root)}
32
33
34 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
35 dot_muse = muse_dir(tmp_path)
36 dot_muse.mkdir()
37 repo_id = fake_id("repo")
38 (dot_muse / "repo.json").write_text(json.dumps({
39 "repo_id": repo_id,
40 "domain": "midi",
41 "default_branch": "main",
42 "created_at": "2025-01-01T00:00:00+00:00",
43 }), encoding="utf-8")
44 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
45 (dot_muse / "refs" / "heads").mkdir(parents=True)
46 (dot_muse / "snapshots").mkdir()
47 (dot_muse / "commits").mkdir()
48 (dot_muse / "objects").mkdir()
49 return tmp_path, repo_id
50
51
52 def _make_commit(root: pathlib.Path, repo_id: str, message: str = "test") -> str:
53 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
54 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
55
56 ref_file = heads_dir(root) / "main"
57 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
58 manifest: Manifest = {}
59 snap_id = compute_snapshot_id(manifest)
60 committed_at = datetime.datetime.now(datetime.timezone.utc)
61 commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [],
62 snapshot_id=snap_id, message=message,
63 committed_at_iso=committed_at.isoformat(),
64 )
65 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
66 write_commit(root, CommitRecord(
67 commit_id=commit_id, repo_id=repo_id, branch="main",
68 snapshot_id=snap_id, message=message, committed_at=committed_at,
69 parent_commit_id=parent_id,
70 ))
71 ref_file.parent.mkdir(parents=True, exist_ok=True)
72 ref_file.write_text(commit_id, encoding="utf-8")
73 return commit_id
74
75
76 # ---------------------------------------------------------------------------
77 # Reset tests
78 # ---------------------------------------------------------------------------
79
80 class TestResetCLI:
81 def test_reset_hard_to_previous_commit(self, tmp_path: pathlib.Path) -> None:
82 root, repo_id = _init_repo(tmp_path)
83 commit1 = _make_commit(root, repo_id, message="first")
84 _make_commit(root, repo_id, message="second")
85 result = runner.invoke(
86 cli, ["reset", "--hard", commit1], env=_env(root), catch_exceptions=False
87 )
88 assert result.exit_code == 0
89 ref = (heads_dir(root) / "main").read_text().strip()
90 assert ref == commit1
91
92 def test_reset_soft_to_previous_commit(self, tmp_path: pathlib.Path) -> None:
93 root, repo_id = _init_repo(tmp_path)
94 commit1 = _make_commit(root, repo_id, message="first")
95 _make_commit(root, repo_id, message="second")
96 result = runner.invoke(
97 cli, ["reset", "--soft", commit1], env=_env(root), catch_exceptions=False
98 )
99 assert result.exit_code == 0
100
101 def test_reset_to_head_tilde_syntax(self, tmp_path: pathlib.Path) -> None:
102 root, repo_id = _init_repo(tmp_path)
103 _make_commit(root, repo_id, message="first")
104 _make_commit(root, repo_id, message="second")
105 result = runner.invoke(cli, ["reset", "--hard", "HEAD~1"], env=_env(root), catch_exceptions=False)
106 # HEAD~1 syntax may not be supported by resolve_commit_ref; skip if not
107 assert result.exit_code in (0, 1)
108
109 def test_reset_invalid_ref_fails(self, tmp_path: pathlib.Path) -> None:
110 root, repo_id = _init_repo(tmp_path)
111 _make_commit(root, repo_id)
112 result = runner.invoke(cli, ["reset", "nonexistent-ref"], env=_env(root))
113 assert result.exit_code != 0
114
115 def test_reset_to_full_commit_id(self, tmp_path: pathlib.Path) -> None:
116 root, repo_id = _init_repo(tmp_path)
117 commit1 = _make_commit(root, repo_id, message="first")
118 _make_commit(root, repo_id, message="second")
119 result = runner.invoke(cli, ["reset", commit1], env=_env(root), catch_exceptions=False)
120 assert result.exit_code == 0
121
122 def test_reset_format_json(self, tmp_path: pathlib.Path) -> None:
123 root, repo_id = _init_repo(tmp_path)
124 commit1 = _make_commit(root, repo_id, message="first")
125 _make_commit(root, repo_id, message="second")
126 result = runner.invoke(
127 cli, ["reset", "--json", commit1],
128 env=_env(root), catch_exceptions=False
129 )
130 assert result.exit_code == 0
131 data = json.loads(result.output)
132 assert isinstance(data, dict)
133
134
135 class TestResetStress:
136 def test_reset_across_many_commits(self, tmp_path: pathlib.Path) -> None:
137 root, repo_id = _init_repo(tmp_path)
138 first = _make_commit(root, repo_id, message="first")
139 for i in range(20):
140 _make_commit(root, repo_id, message=f"commit {i}")
141 result = runner.invoke(cli, ["reset", "--hard", first], env=_env(root), catch_exceptions=False)
142 assert result.exit_code == 0
143 ref = (heads_dir(root) / "main").read_text().strip()
144 assert ref == first
145
146
147 # ---------------------------------------------------------------------------
148 # Revert tests
149 # ---------------------------------------------------------------------------
150
151 class TestRevertCLI:
152 def test_revert_most_recent_commit(self, tmp_path: pathlib.Path) -> None:
153 root, repo_id = _init_repo(tmp_path)
154 _make_commit(root, repo_id, message="first")
155 commit2 = _make_commit(root, repo_id, message="second")
156 result = runner.invoke(cli, ["revert", commit2], env=_env(root), catch_exceptions=False)
157 assert result.exit_code == 0
158
159 def test_revert_invalid_commit_fails(self, tmp_path: pathlib.Path) -> None:
160 root, repo_id = _init_repo(tmp_path)
161 _make_commit(root, repo_id)
162 result = runner.invoke(cli, ["revert", "deadbeef" * 8], env=_env(root))
163 assert result.exit_code != 0
164
165 def test_revert_creates_new_commit(self, tmp_path: pathlib.Path) -> None:
166 root, repo_id = _init_repo(tmp_path)
167 commit1 = _make_commit(root, repo_id, message="first")
168 commit2 = _make_commit(root, repo_id, message="second")
169 runner.invoke(cli, ["revert", commit2], env=_env(root), catch_exceptions=False)
170 from muse.core.store import get_all_commits
171 commits = get_all_commits(root)
172 # Should have 3 commits now (original two + revert commit)
173 assert len(commits) >= 2
174
175 def test_revert_format_json(self, tmp_path: pathlib.Path) -> None:
176 root, repo_id = _init_repo(tmp_path)
177 _make_commit(root, repo_id, message="first")
178 commit2 = _make_commit(root, repo_id, message="second")
179 result = runner.invoke(
180 cli, ["revert", "--json", commit2],
181 env=_env(root), catch_exceptions=False
182 )
183 assert result.exit_code == 0
184 data = json.loads(result.output)
185 assert isinstance(data, dict)
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 126 days ago