gabriel / muse public
test_cmd_reset_revert.py python
192 lines 7.5 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 7 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.commits import (
54 CommitRecord,
55 write_commit,
56 )
57 from muse.core.snapshots import (
58 SnapshotRecord,
59 write_snapshot,
60 )
61 from muse.core.ids import hash_snapshot, hash_commit
62
63 ref_file = heads_dir(root) / "main"
64 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
65 manifest: Manifest = {}
66 snap_id = hash_snapshot(manifest)
67 committed_at = datetime.datetime.now(datetime.timezone.utc)
68 commit_id = hash_commit( parent_ids=[parent_id] if parent_id else [],
69 snapshot_id=snap_id, message=message,
70 committed_at_iso=committed_at.isoformat(),
71 )
72 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
73 write_commit(root, CommitRecord(
74 commit_id=commit_id, branch="main",
75 snapshot_id=snap_id, message=message, committed_at=committed_at,
76 parent_commit_id=parent_id,
77 ))
78 ref_file.parent.mkdir(parents=True, exist_ok=True)
79 ref_file.write_text(commit_id, encoding="utf-8")
80 return commit_id
81
82
83 # ---------------------------------------------------------------------------
84 # Reset tests
85 # ---------------------------------------------------------------------------
86
87 class TestResetCLI:
88 def test_reset_hard_to_previous_commit(self, tmp_path: pathlib.Path) -> None:
89 root, repo_id = _init_repo(tmp_path)
90 commit1 = _make_commit(root, repo_id, message="first")
91 _make_commit(root, repo_id, message="second")
92 result = runner.invoke(
93 cli, ["reset", "--hard", commit1], env=_env(root), catch_exceptions=False
94 )
95 assert result.exit_code == 0
96 ref = (heads_dir(root) / "main").read_text().strip()
97 assert ref == commit1
98
99 def test_reset_soft_to_previous_commit(self, tmp_path: pathlib.Path) -> None:
100 root, repo_id = _init_repo(tmp_path)
101 commit1 = _make_commit(root, repo_id, message="first")
102 _make_commit(root, repo_id, message="second")
103 result = runner.invoke(
104 cli, ["reset", "--soft", commit1], env=_env(root), catch_exceptions=False
105 )
106 assert result.exit_code == 0
107
108 def test_reset_to_head_tilde_syntax(self, tmp_path: pathlib.Path) -> None:
109 root, repo_id = _init_repo(tmp_path)
110 _make_commit(root, repo_id, message="first")
111 _make_commit(root, repo_id, message="second")
112 result = runner.invoke(cli, ["reset", "--hard", "HEAD~1"], env=_env(root), catch_exceptions=False)
113 # HEAD~1 syntax may not be supported by resolve_commit_ref; skip if not
114 assert result.exit_code in (0, 1)
115
116 def test_reset_invalid_ref_fails(self, tmp_path: pathlib.Path) -> None:
117 root, repo_id = _init_repo(tmp_path)
118 _make_commit(root, repo_id)
119 result = runner.invoke(cli, ["reset", "nonexistent-ref"], env=_env(root))
120 assert result.exit_code != 0
121
122 def test_reset_to_full_commit_id(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(cli, ["reset", commit1], env=_env(root), catch_exceptions=False)
127 assert result.exit_code == 0
128
129 def test_reset_format_json(self, tmp_path: pathlib.Path) -> None:
130 root, repo_id = _init_repo(tmp_path)
131 commit1 = _make_commit(root, repo_id, message="first")
132 _make_commit(root, repo_id, message="second")
133 result = runner.invoke(
134 cli, ["reset", "--json", commit1],
135 env=_env(root), catch_exceptions=False
136 )
137 assert result.exit_code == 0
138 data = json.loads(result.output)
139 assert isinstance(data, dict)
140
141
142 class TestResetStress:
143 def test_reset_across_many_commits(self, tmp_path: pathlib.Path) -> None:
144 root, repo_id = _init_repo(tmp_path)
145 first = _make_commit(root, repo_id, message="first")
146 for i in range(20):
147 _make_commit(root, repo_id, message=f"commit {i}")
148 result = runner.invoke(cli, ["reset", "--hard", first], env=_env(root), catch_exceptions=False)
149 assert result.exit_code == 0
150 ref = (heads_dir(root) / "main").read_text().strip()
151 assert ref == first
152
153
154 # ---------------------------------------------------------------------------
155 # Revert tests
156 # ---------------------------------------------------------------------------
157
158 class TestRevertCLI:
159 def test_revert_most_recent_commit(self, tmp_path: pathlib.Path) -> None:
160 root, repo_id = _init_repo(tmp_path)
161 _make_commit(root, repo_id, message="first")
162 commit2 = _make_commit(root, repo_id, message="second")
163 result = runner.invoke(cli, ["revert", commit2], env=_env(root), catch_exceptions=False)
164 assert result.exit_code == 0
165
166 def test_revert_invalid_commit_fails(self, tmp_path: pathlib.Path) -> None:
167 root, repo_id = _init_repo(tmp_path)
168 _make_commit(root, repo_id)
169 result = runner.invoke(cli, ["revert", "deadbeef" * 8], env=_env(root))
170 assert result.exit_code != 0
171
172 def test_revert_creates_new_commit(self, tmp_path: pathlib.Path) -> None:
173 root, repo_id = _init_repo(tmp_path)
174 commit1 = _make_commit(root, repo_id, message="first")
175 commit2 = _make_commit(root, repo_id, message="second")
176 runner.invoke(cli, ["revert", commit2], env=_env(root), catch_exceptions=False)
177 from muse.core.commits import get_all_commits
178 commits = get_all_commits(root)
179 # Should have 3 commits now (original two + revert commit)
180 assert len(commits) >= 2
181
182 def test_revert_format_json(self, tmp_path: pathlib.Path) -> None:
183 root, repo_id = _init_repo(tmp_path)
184 _make_commit(root, repo_id, message="first")
185 commit2 = _make_commit(root, repo_id, message="second")
186 result = runner.invoke(
187 cli, ["revert", "--json", commit2],
188 env=_env(root), catch_exceptions=False
189 )
190 assert result.exit_code == 0
191 data = json.loads(result.output)
192 assert isinstance(data, dict)
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 7 days ago