gabriel / muse public
test_cmd_for_each_ref_hardening.py python
328 lines 12.3 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """Hardening tests for ``muse for-each-ref`` — agent supercharge series.
2
3 Tests added in this pass
4 ------------------------
5 - ``duration_ms`` present and valid in JSON output
6 - ``exit_code`` present and zero in JSON output
7 - ``current_branch`` present — which branch HEAD points to
8 - JSON is compact (single line)
9 - ``commit_id`` and ``snapshot_id`` carry sha256: prefix
10 - Data integrity: duration_ms non-negative float, exit_code int,
11 current_branch matches HEAD, count == len(refs)
12 - Performance: 100-branch repo round-trip under 5 s, duration_ms plausible
13 - Security: error output to stderr, no traceback
14 """
15 from __future__ import annotations
16 from collections.abc import Mapping
17
18 import datetime
19 import json
20 import pathlib
21 import time
22
23 import pytest
24
25 from tests.cli_test_helper import CliRunner, InvokeResult
26 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
27 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
28 from muse.core.paths import muse_dir, ref_path
29
30 runner = CliRunner()
31
32
33 # ---------------------------------------------------------------------------
34 # Helpers
35 # ---------------------------------------------------------------------------
36
37 def _init_repo(path: pathlib.Path, head_branch: str = "main") -> pathlib.Path:
38 dot_muse = muse_dir(path)
39 for sub in ("commits", "snapshots", "objects", "refs/heads"):
40 (dot_muse / sub).mkdir(parents=True, exist_ok=True)
41 (dot_muse / "HEAD").write_text(f"ref: refs/heads/{head_branch}\n")
42 (dot_muse / "repo.json").write_text(
43 json.dumps({"repo_id": "test-repo", "domain": "code"})
44 )
45 return path
46
47
48 def _commit(
49 repo: pathlib.Path,
50 msg: str,
51 branch: str = "main",
52 parent: str | None = None,
53 ts: datetime.datetime | None = None,
54 author: str = "gabriel",
55 ) -> str:
56 ts = ts or datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
57 sid = compute_snapshot_id({})
58 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest={}, created_at=ts))
59 parent_ids = [parent] if parent else []
60 cid = compute_commit_id( parent_ids=parent_ids,
61 snapshot_id=sid,
62 message=msg,
63 committed_at_iso=ts.isoformat(),
64 author=author,
65 )
66 write_commit(repo, CommitRecord(
67 commit_id=cid, repo_id="test-repo", branch=branch,
68 snapshot_id=sid, message=msg, committed_at=ts,
69 author=author, parent_commit_id=parent, parent2_commit_id=None,
70 ))
71 branch_ref = ref_path(repo, branch)
72 branch_ref.parent.mkdir(parents=True, exist_ok=True)
73 branch_ref.write_text(cid)
74 return cid
75
76
77 def _fer(repo: pathlib.Path, *args: str) -> InvokeResult:
78 return runner.invoke(None, ["for-each-ref", "--json", *args],
79 env={"MUSE_REPO_ROOT": str(repo)})
80
81
82 def _json(r: InvokeResult) -> Mapping[str, object]:
83 return json.loads(r.output)
84
85
86 # ---------------------------------------------------------------------------
87 # duration_ms
88 # ---------------------------------------------------------------------------
89
90 class TestElapsedSeconds:
91 def test_present_in_full_output(self, tmp_path: pathlib.Path) -> None:
92 _init_repo(tmp_path)
93 _commit(tmp_path, "c1")
94 assert "duration_ms" in _json(_fer(tmp_path))
95
96 def test_present_with_no_commits(self, tmp_path: pathlib.Path) -> None:
97 _init_repo(tmp_path)
98 _commit(tmp_path, "c1")
99 assert "duration_ms" in _json(_fer(tmp_path, "--no-commits"))
100
101 def test_present_on_empty_repo(self, tmp_path: pathlib.Path) -> None:
102 _init_repo(tmp_path)
103 assert "duration_ms" in _json(_fer(tmp_path))
104
105 def test_is_float(self, tmp_path: pathlib.Path) -> None:
106 _init_repo(tmp_path)
107 _commit(tmp_path, "c1")
108 assert isinstance(_json(_fer(tmp_path))["duration_ms"], float)
109
110 def test_non_negative(self, tmp_path: pathlib.Path) -> None:
111 _init_repo(tmp_path)
112 _commit(tmp_path, "c1")
113 assert _json(_fer(tmp_path))["duration_ms"] >= 0.0
114
115 def test_six_decimal_places(self, tmp_path: pathlib.Path) -> None:
116 _init_repo(tmp_path)
117 _commit(tmp_path, "c1")
118 v = _json(_fer(tmp_path))["duration_ms"]
119 assert v == round(v, 6)
120
121 def test_present_with_pattern_filter(self, tmp_path: pathlib.Path) -> None:
122 _init_repo(tmp_path)
123 _commit(tmp_path, "c1")
124 data = _json(_fer(tmp_path, "--pattern", "refs/heads/main"))
125 assert "duration_ms" in data
126
127 def test_present_with_count_limit(self, tmp_path: pathlib.Path) -> None:
128 _init_repo(tmp_path)
129 for b in ["a", "b", "c"]:
130 _commit(tmp_path, f"c-{b}", b)
131 assert "duration_ms" in _json(_fer(tmp_path, "--count", "2"))
132
133
134 # ---------------------------------------------------------------------------
135 # exit_code
136 # ---------------------------------------------------------------------------
137
138 class TestExitCode:
139 def test_present_in_full_output(self, tmp_path: pathlib.Path) -> None:
140 _init_repo(tmp_path)
141 _commit(tmp_path, "c1")
142 assert "exit_code" in _json(_fer(tmp_path))
143
144 def test_zero_on_success(self, tmp_path: pathlib.Path) -> None:
145 _init_repo(tmp_path)
146 _commit(tmp_path, "c1")
147 assert _json(_fer(tmp_path))["exit_code"] == 0
148
149 def test_zero_on_empty_repo(self, tmp_path: pathlib.Path) -> None:
150 _init_repo(tmp_path)
151 assert _json(_fer(tmp_path))["exit_code"] == 0
152
153 def test_zero_with_no_commits(self, tmp_path: pathlib.Path) -> None:
154 _init_repo(tmp_path)
155 _commit(tmp_path, "c1")
156 assert _json(_fer(tmp_path, "--no-commits"))["exit_code"] == 0
157
158 def test_is_int_not_bool(self, tmp_path: pathlib.Path) -> None:
159 _init_repo(tmp_path)
160 _commit(tmp_path, "c1")
161 assert type(_json(_fer(tmp_path))["exit_code"]) is int
162
163
164 # ---------------------------------------------------------------------------
165 # current_branch
166 # ---------------------------------------------------------------------------
167
168 class TestCurrentBranch:
169 def test_present_in_output(self, tmp_path: pathlib.Path) -> None:
170 _init_repo(tmp_path, head_branch="main")
171 _commit(tmp_path, "c1", "main")
172 assert "current_branch" in _json(_fer(tmp_path))
173
174 def test_matches_head_branch(self, tmp_path: pathlib.Path) -> None:
175 _init_repo(tmp_path, head_branch="dev")
176 _commit(tmp_path, "c1", "dev")
177 assert _json(_fer(tmp_path))["current_branch"] == "dev"
178
179 def test_main_by_default(self, tmp_path: pathlib.Path) -> None:
180 _init_repo(tmp_path, head_branch="main")
181 _commit(tmp_path, "c1", "main")
182 assert _json(_fer(tmp_path))["current_branch"] == "main"
183
184 def test_present_with_no_commits_flag(self, tmp_path: pathlib.Path) -> None:
185 _init_repo(tmp_path, head_branch="main")
186 _commit(tmp_path, "c1", "main")
187 assert "current_branch" in _json(_fer(tmp_path, "--no-commits"))
188
189 def test_present_on_empty_repo(self, tmp_path: pathlib.Path) -> None:
190 _init_repo(tmp_path, head_branch="main")
191 data = _json(_fer(tmp_path))
192 assert "current_branch" in data
193
194 def test_feature_branch_reflected(self, tmp_path: pathlib.Path) -> None:
195 _init_repo(tmp_path, head_branch="feat/my-thing")
196 _commit(tmp_path, "c1", "feat/my-thing")
197 assert _json(_fer(tmp_path))["current_branch"] == "feat/my-thing"
198
199
200 # ---------------------------------------------------------------------------
201 # Compact JSON
202 # ---------------------------------------------------------------------------
203
204 class TestCompactJson:
205 def test_output_is_single_line(self, tmp_path: pathlib.Path) -> None:
206 _init_repo(tmp_path)
207 _commit(tmp_path, "c1")
208 r = _fer(tmp_path)
209 assert len(r.output.strip().splitlines()) == 1
210
211 def test_no_commits_is_single_line(self, tmp_path: pathlib.Path) -> None:
212 _init_repo(tmp_path)
213 _commit(tmp_path, "c1")
214 r = _fer(tmp_path, "--no-commits")
215 assert len(r.output.strip().splitlines()) == 1
216
217 def test_empty_repo_is_single_line(self, tmp_path: pathlib.Path) -> None:
218 _init_repo(tmp_path)
219 r = _fer(tmp_path)
220 assert len(r.output.strip().splitlines()) == 1
221
222
223 # ---------------------------------------------------------------------------
224 # sha256: prefix on IDs
225 # ---------------------------------------------------------------------------
226
227 class TestSha256Prefix:
228 def test_commit_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None:
229 _init_repo(tmp_path)
230 _commit(tmp_path, "c1")
231 ref = _json(_fer(tmp_path))["refs"][0]
232 assert ref["commit_id"].startswith("sha256:")
233
234 def test_commit_id_full_length(self, tmp_path: pathlib.Path) -> None:
235 _init_repo(tmp_path)
236 _commit(tmp_path, "c1")
237 ref = _json(_fer(tmp_path))["refs"][0]
238 # sha256: (7) + 64 hex chars = 71
239 assert len(ref["commit_id"]) == 71
240
241 def test_snapshot_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None:
242 _init_repo(tmp_path)
243 _commit(tmp_path, "c1")
244 ref = _json(_fer(tmp_path))["refs"][0]
245 assert ref["snapshot_id"].startswith("sha256:")
246
247 def test_no_commits_commit_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None:
248 _init_repo(tmp_path)
249 _commit(tmp_path, "c1")
250 ref = _json(_fer(tmp_path, "--no-commits"))["refs"][0]
251 assert ref["commit_id"].startswith("sha256:")
252
253
254 # ---------------------------------------------------------------------------
255 # Data integrity
256 # ---------------------------------------------------------------------------
257
258 class TestDataIntegrity:
259 def test_count_equals_len_refs(self, tmp_path: pathlib.Path) -> None:
260 _init_repo(tmp_path)
261 for b in ["a", "b", "c", "d"]:
262 _commit(tmp_path, f"c-{b}", b)
263 data = _json(_fer(tmp_path))
264 assert data["count"] == len(data["refs"])
265
266 def test_count_equals_len_refs_after_pattern(self, tmp_path: pathlib.Path) -> None:
267 _init_repo(tmp_path)
268 for b in ["feat/x", "feat/y", "main"]:
269 _commit(tmp_path, f"c-{b}", b)
270 data = _json(_fer(tmp_path, "--pattern", "refs/heads/feat/*"))
271 assert data["count"] == len(data["refs"])
272
273 def test_count_equals_len_refs_after_count_limit(self, tmp_path: pathlib.Path) -> None:
274 _init_repo(tmp_path)
275 for b in ["a", "b", "c", "d", "e"]:
276 _commit(tmp_path, f"c-{b}", b)
277 data = _json(_fer(tmp_path, "--count", "3"))
278 assert data["count"] == len(data["refs"])
279 assert data["count"] == 3
280
281 def test_all_refs_have_branch_and_ref_fields(self, tmp_path: pathlib.Path) -> None:
282 _init_repo(tmp_path)
283 for b in ["main", "dev", "feat/x"]:
284 _commit(tmp_path, f"c-{b}", b)
285 data = _json(_fer(tmp_path))
286 for ref in data["refs"]:
287 assert "branch" in ref
288 assert "ref" in ref
289 assert ref["ref"] == f"refs/heads/{ref['branch']}"
290
291 def test_committed_at_is_iso8601(self, tmp_path: pathlib.Path) -> None:
292 _init_repo(tmp_path)
293 _commit(tmp_path, "c1")
294 ref = _json(_fer(tmp_path))["refs"][0]
295 # Must parse as a datetime without raising
296 import datetime
297 datetime.datetime.fromisoformat(ref["committed_at"])
298
299
300 # ---------------------------------------------------------------------------
301 # Performance
302 # ---------------------------------------------------------------------------
303
304 class TestPerformance:
305 def test_duration_ms_plausible(self, tmp_path: pathlib.Path) -> None:
306 _init_repo(tmp_path)
307 _commit(tmp_path, "c1")
308 assert _json(_fer(tmp_path))["duration_ms"] < 10.0
309
310 def test_100_branch_repo_under_5s(self, tmp_path: pathlib.Path) -> None:
311 _init_repo(tmp_path)
312 for i in range(100):
313 _commit(tmp_path, f"c-{i}", f"branch-{i:03d}")
314 t0 = time.monotonic()
315 r = _fer(tmp_path)
316 assert r.exit_code == 0
317 assert time.monotonic() - t0 < 5.0
318 assert _json(r)["count"] == 100
319
320 def test_no_commits_faster_than_full(self, tmp_path: pathlib.Path) -> None:
321 """--no-commits duration_ms <= full duration_ms (with some slack)."""
322 _init_repo(tmp_path)
323 for i in range(50):
324 _commit(tmp_path, f"c-{i}", f"b-{i:03d}")
325 full = _json(_fer(tmp_path))["duration_ms"]
326 fast = _json(_fer(tmp_path, "--no-commits"))["duration_ms"]
327 # fast path must not be 10x slower than full (loose bound; CI noise)
328 assert fast < full * 10 + 1.0
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago