gabriel / muse public
test_cmd_reflog.py python
220 lines 8.8 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 122 days ago
1 """Comprehensive tests for ``muse reflog``.
2
3 Covers:
4 - Unit: _fmt_entry sanitizes operation field
5 - Integration: reflog populated by commits, --all flag
6 - E2E: full CLI via CliRunner
7 - Security: branch name validated before use as path, operation sanitized
8 - Stress: large reflog with limit
9 """
10
11 from __future__ import annotations
12
13 import datetime
14 import json
15 import pathlib
16
17 import pytest
18 from tests.cli_test_helper import CliRunner
19 from muse.core.types import NULL_COMMIT_ID, fake_id
20 from muse.core.paths import muse_dir, ref_path
21
22 cli = None # argparse migration — CliRunner ignores this arg
23
24 runner = CliRunner()
25
26
27 # ---------------------------------------------------------------------------
28 # Helpers
29 # ---------------------------------------------------------------------------
30
31 def _env(root: pathlib.Path) -> Manifest:
32 return {"MUSE_REPO_ROOT": str(root)}
33
34
35 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
36 dot_muse = muse_dir(tmp_path)
37 dot_muse.mkdir()
38 repo_id = fake_id("repo")
39 (dot_muse / "repo.json").write_text(json.dumps({
40 "repo_id": repo_id,
41 "domain": "midi",
42 "default_branch": "main",
43 "created_at": "2025-01-01T00:00:00+00:00",
44 }), encoding="utf-8")
45 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
46 (dot_muse / "refs" / "heads").mkdir(parents=True)
47 (dot_muse / "snapshots").mkdir()
48 (dot_muse / "commits").mkdir()
49 (dot_muse / "objects").mkdir()
50 return tmp_path, repo_id
51
52
53 def _make_commit_with_reflog(
54 root: pathlib.Path, repo_id: str, message: str = "commit", branch: str = "main"
55 ) -> str:
56 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
57 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
58 from muse.core.reflog import append_reflog
59
60 ref_file = ref_path(root, branch)
61 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
62 manifest: Manifest = {}
63 snap_id = compute_snapshot_id(manifest)
64 committed_at = datetime.datetime.now(datetime.timezone.utc)
65 commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [],
66 snapshot_id=snap_id, message=message,
67 committed_at_iso=committed_at.isoformat(),
68 )
69 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
70 write_commit(root, CommitRecord(
71 commit_id=commit_id, repo_id=repo_id, branch=branch,
72 snapshot_id=snap_id, message=message, committed_at=committed_at,
73 parent_commit_id=parent_id,
74 ))
75 ref_file.parent.mkdir(parents=True, exist_ok=True)
76 ref_file.write_text(commit_id, encoding="utf-8")
77 append_reflog(root, branch, old_id=parent_id or NULL_COMMIT_ID, new_id=commit_id,
78 author="user", operation=f"commit: {message}")
79 return commit_id
80
81
82 # ---------------------------------------------------------------------------
83 # Unit tests
84 # ---------------------------------------------------------------------------
85
86
87 class TestRegisterFlags:
88 def _parse(self, *args: str) -> "argparse.Namespace":
89 import argparse
90 from muse.cli.commands.reflog import register
91 p = argparse.ArgumentParser()
92 sub = p.add_subparsers()
93 register(sub)
94 return p.parse_args(["reflog", *args])
95
96 def test_default_json_out_is_false(self) -> None:
97 ns = self._parse()
98 assert ns.json_out is False
99
100 def test_json_flag_sets_json_out(self) -> None:
101 ns = self._parse("--json")
102 assert ns.json_out is True
103
104 def test_j_shorthand_sets_json_out(self) -> None:
105 ns = self._parse("-j")
106 assert ns.json_out is True
107
108
109 class TestReflogUnit:
110 def test_fmt_entry_sanitizes_operation(self) -> None:
111 from muse.cli.commands.reflog import _fmt_entry
112 from muse.core.reflog import ReflogEntry
113
114 entry = ReflogEntry(
115 old_id=NULL_COMMIT_ID, new_id="a" * 64, author="user",
116 timestamp=datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc),
117 operation="commit: Hello\x1b[31mRED\x1b[0m",
118 )
119 result = _fmt_entry(0, entry)
120 assert "\x1b" not in result
121
122 def test_fmt_entry_initial_shown_as_initial(self) -> None:
123 from muse.cli.commands.reflog import _fmt_entry
124 from muse.core.reflog import ReflogEntry
125
126 entry = ReflogEntry(
127 old_id=NULL_COMMIT_ID, new_id="b" * 64, author="user",
128 timestamp=datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc),
129 operation="branch: created",
130 )
131 result = _fmt_entry(0, entry)
132 assert "initial" in result
133
134
135 # ---------------------------------------------------------------------------
136 # Integration tests
137 # ---------------------------------------------------------------------------
138
139 class TestReflogIntegration:
140 def test_reflog_empty_repo(self, tmp_path: pathlib.Path) -> None:
141 root, _ = _init_repo(tmp_path)
142 result = runner.invoke(cli, ["reflog"], env=_env(root), catch_exceptions=False)
143 assert result.exit_code == 0
144 assert "No reflog entries" in result.output
145
146 def test_reflog_after_commit(self, tmp_path: pathlib.Path) -> None:
147 root, repo_id = _init_repo(tmp_path)
148 _make_commit_with_reflog(root, repo_id, message="my first commit")
149 result = runner.invoke(cli, ["reflog"], env=_env(root), catch_exceptions=False)
150 assert result.exit_code == 0
151 assert "@{0" in result.output
152 assert "commit: my first commit" in result.output
153
154 def test_reflog_limit(self, tmp_path: pathlib.Path) -> None:
155 root, repo_id = _init_repo(tmp_path)
156 for i in range(10):
157 _make_commit_with_reflog(root, repo_id, message=f"commit {i}")
158 result = runner.invoke(cli, ["reflog", "--limit", "3"], env=_env(root), catch_exceptions=False)
159 assert result.exit_code == 0
160 lines = [l for l in result.output.splitlines() if "@{" in l]
161 assert len(lines) <= 3
162
163 def test_reflog_branch_flag(self, tmp_path: pathlib.Path) -> None:
164 root, repo_id = _init_repo(tmp_path)
165 _make_commit_with_reflog(root, repo_id, message="on main")
166 result = runner.invoke(cli, ["reflog", "--branch", "main"], env=_env(root), catch_exceptions=False)
167 assert result.exit_code == 0
168 assert "main" in result.output
169
170 def test_reflog_short_flags(self, tmp_path: pathlib.Path) -> None:
171 root, repo_id = _init_repo(tmp_path)
172 for i in range(5):
173 _make_commit_with_reflog(root, repo_id, message=f"commit {i}")
174 result = runner.invoke(cli, ["reflog", "--limit", "2", "-b", "main"], env=_env(root), catch_exceptions=False)
175 assert result.exit_code == 0
176 lines = [l for l in result.output.splitlines() if "@{" in l]
177 assert len(lines) <= 2
178
179 def test_reflog_all_flag_lists_refs(self, tmp_path: pathlib.Path) -> None:
180 root, repo_id = _init_repo(tmp_path)
181 _make_commit_with_reflog(root, repo_id, message="first")
182 result = runner.invoke(cli, ["reflog", "--all"], env=_env(root), catch_exceptions=False)
183 assert result.exit_code == 0
184
185
186 # ---------------------------------------------------------------------------
187 # Security tests
188 # ---------------------------------------------------------------------------
189
190 class TestReflogSecurity:
191 def test_invalid_branch_name_rejected(self, tmp_path: pathlib.Path) -> None:
192 root, repo_id = _init_repo(tmp_path)
193 _make_commit_with_reflog(root, repo_id)
194 result = runner.invoke(cli, ["reflog", "--branch", "../../../etc/passwd"], env=_env(root))
195 assert result.exit_code != 0
196
197 def test_operation_with_control_chars_sanitized(self, tmp_path: pathlib.Path) -> None:
198 root, repo_id = _init_repo(tmp_path)
199 from muse.core.reflog import append_reflog
200 _make_commit_with_reflog(root, repo_id, message="clean")
201 append_reflog(root, "main", old_id=NULL_COMMIT_ID, new_id="a" * 64,
202 author="user", operation="malicious\x1b[31mRED\x1b[0m op")
203 result = runner.invoke(cli, ["reflog"], env=_env(root), catch_exceptions=False)
204 assert result.exit_code == 0
205 assert "\x1b" not in result.output
206
207
208 # ---------------------------------------------------------------------------
209 # Stress tests
210 # ---------------------------------------------------------------------------
211
212 class TestReflogStress:
213 def test_large_reflog_with_limit(self, tmp_path: pathlib.Path) -> None:
214 root, repo_id = _init_repo(tmp_path)
215 for i in range(50):
216 _make_commit_with_reflog(root, repo_id, message=f"commit {i:03d}")
217 result = runner.invoke(cli, ["reflog", "--limit", "5"], env=_env(root), catch_exceptions=False)
218 assert result.exit_code == 0
219 lines = [l for l in result.output.splitlines() if "@{" in l]
220 assert len(lines) <= 5
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 122 days ago