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