gabriel / muse public
test_reflog_supercharge.py python
409 lines 16.7 KB
Raw
sha256:37af4ab271a64f11d7bf14b4871ebaaf5dd68fb0247521007c6b034d35aa01be Merge branch 'fix/hub-user-read-update-path' into dev Human 10 days ago
1 """Supercharge tests for ``muse reflog``.
2
3 Coverage tiers
4 --------------
5 - Unit: _short_id helper — bare hex and sha256:-prefixed inputs
6 - Integration: duration_ms + exit_code in both JSON output paths
7 - Data integrity: new_id/old_id sha256:-prefixed in JSON; full IDs in text
8 - Filter behaviour: total reflects post-filter count; date range edge cases
9 - Security: null-ID shown as sha256:000…, ANSI in IDs sanitised in text
10 - Performance: empty reflog and 100-entry reflog timing
11 """
12 from __future__ import annotations
13
14 import datetime
15 import json
16 import pathlib
17 import re
18 import time
19
20 from muse.core.errors import ExitCode
21 from muse.core.reflog import append_reflog
22 from tests.cli_test_helper import CliRunner, InvokeResult
23 from muse.core.types import NULL_COMMIT_ID, NULL_LONG_ID, long_id, fake_id, short_id as _short_id
24 from muse.core.paths import logs_dir, muse_dir
25
26 runner = CliRunner()
27
28 _NULL_ID = NULL_COMMIT_ID
29 _SHA_A = long_id("a" * 64)
30 _SHA_B = long_id("b" * 64)
31
32 _SHA256_FULL = re.compile(r"^sha256:[0-9a-f]{64}$")
33 _SHA256_SHORT_19 = re.compile(r"^sha256:[0-9a-f]{12}$")
34
35 _TS = datetime.datetime(2026, 1, 15, 12, 0, tzinfo=datetime.timezone.utc)
36
37
38 # ---------------------------------------------------------------------------
39 # Helpers
40 # ---------------------------------------------------------------------------
41
42
43 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
44 repo = tmp_path / "repo"
45 dot_muse = muse_dir(repo)
46 for sub in ("objects", "commits", "snapshots", "refs/heads",
47 "logs/refs/heads", "logs"):
48 (dot_muse / sub).mkdir(parents=True, exist_ok=True)
49 (dot_muse / "HEAD").write_text("ref: refs/heads/main")
50 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
51 return repo
52
53
54 def _append(
55 repo: pathlib.Path,
56 *,
57 branch: str = "main",
58 old_id: str = _NULL_ID,
59 new_id: str = _SHA_A,
60 author: str = "gabriel",
61 operation: str = "commit: test",
62 timestamp: datetime.datetime | None = None,
63 ) -> None:
64 """Write one reflog entry.
65
66 When *timestamp* is given, write the raw log line directly so tests can
67 control the stored timestamp precisely. Otherwise delegate to
68 ``append_reflog`` which stamps with the current time.
69 """
70 if timestamp is None:
71 append_reflog(repo, branch, old_id=old_id, new_id=new_id,
72 author=author, operation=operation)
73 return
74 # Write raw log line to both HEAD and branch logs (mirrors append_reflog).
75 ts_unix = int(timestamp.timestamp())
76 safe_op = operation.replace("\n", "").replace("\r", "")
77 safe_author = author.replace("\n", "").replace("\r", "").replace("\t", "")
78 line = f"{old_id} {new_id} {safe_author} {ts_unix} +0000\t{safe_op}\n"
79 log_dir = logs_dir(repo)
80 head_log = log_dir / "HEAD"
81 head_log.write_text((head_log.read_text(encoding="utf-8") if head_log.exists() else "") + line,
82 encoding="utf-8")
83 if branch:
84 branch_log = log_dir / "refs" / "heads" / branch
85 branch_log.write_text(
86 (branch_log.read_text(encoding="utf-8") if branch_log.exists() else "") + line,
87 encoding="utf-8",
88 )
89
90
91 def _invoke(repo: pathlib.Path, *args: str) -> InvokeResult:
92 from muse.cli.app import main as cli
93 return runner.invoke(cli, ["reflog", *args], env={"MUSE_REPO_ROOT": str(repo)})
94
95
96 # ---------------------------------------------------------------------------
97 # Unit — _short_id
98 # ---------------------------------------------------------------------------
99
100
101 class TestShortId:
102 """_short_id handles both bare-hex (reflog on-disk format) and sha256:-prefixed input."""
103
104 def test_bare_hex_prepends_sha256_prefix(self) -> None:
105 result = _short_id(_SHA_A)
106 assert result.startswith("sha256:")
107
108 def test_bare_hex_12_hex_chars_after_prefix(self) -> None:
109 result = _short_id(_SHA_A)
110 assert result == long_id("a" * 12)
111
112 def test_bare_hex_total_length_is_19(self) -> None:
113 assert len(_short_id(_SHA_B)) == 19
114
115 def test_sha256_prefixed_input_handled(self) -> None:
116 prefixed = long_id("deadbeef" * 8)
117 result = _short_id(prefixed)
118 assert result.startswith("sha256:")
119 assert len(result) == 19
120
121 def test_null_id_shows_zeros(self) -> None:
122 result = _short_id(_NULL_ID)
123 assert result == "0" * 12
124
125 def test_matches_short_regex(self) -> None:
126 assert _SHA256_SHORT_19.match(_short_id(_SHA_A))
127
128
129 # ---------------------------------------------------------------------------
130 # Integration — text format short IDs
131 # ---------------------------------------------------------------------------
132
133
134 class TestTextFormatLongId:
135 """Text format must show the full sha256:<64-hex> for new_id and old_id
136 (muse#52 -- short IDs in human text output were an inconsistency with
137 every other Muse command and a prefix-collision risk)."""
138
139 def _full_tokens(self, line: str) -> list[str]:
140 return [tok for tok in line.split() if _SHA256_FULL.match(tok)]
141
142 def test_new_id_shown_as_sha256_full_in_text(self, tmp_path: pathlib.Path) -> None:
143 repo = _make_repo(tmp_path)
144 _append(repo, new_id=_SHA_A)
145 result = _invoke(repo)
146 assert result.exit_code == 0
147 assert _SHA_A in result.output, \
148 f"full new_id not in text output:\n{result.output}"
149
150 def test_old_id_shown_as_sha256_full_in_text(self, tmp_path: pathlib.Path) -> None:
151 repo = _make_repo(tmp_path)
152 _append(repo, old_id=_SHA_B, new_id=_SHA_A)
153 result = _invoke(repo)
154 assert result.exit_code == 0
155 assert _SHA_B in result.output, \
156 f"full old_id not in text output:\n{result.output}"
157
158 def test_initial_entry_shows_initial_keyword(self, tmp_path: pathlib.Path) -> None:
159 """Null old_id must render as 'initial', not sha256:000…."""
160 repo = _make_repo(tmp_path)
161 _append(repo, old_id=_NULL_ID)
162 result = _invoke(repo)
163 assert "initial" in result.output
164
165 def test_text_id_length_is_64_hex(self, tmp_path: pathlib.Path) -> None:
166 repo = _make_repo(tmp_path)
167 _append(repo, new_id=_SHA_A)
168 result = _invoke(repo)
169 tokens = self._full_tokens(result.output)
170 assert tokens, f"expected at least one sha256:<64-hex> token:\n{result.output}"
171 for tok in tokens:
172 hex_part = tok.split(":", 1)[1]
173 assert len(hex_part) == 64, f"ID token has wrong hex length: {tok!r}"
174
175
176 # ---------------------------------------------------------------------------
177 # Data integrity — JSON IDs
178 # ---------------------------------------------------------------------------
179
180
181 class TestJsonIds:
182 """JSON new_id / old_id must be sha256:<64-hex> canonical form."""
183
184 def test_new_id_sha256_prefixed_in_json(self, tmp_path: pathlib.Path) -> None:
185 repo = _make_repo(tmp_path)
186 _append(repo, new_id=_SHA_A)
187 data = json.loads(_invoke(repo, "--json").output)
188 entry = data["entries"][0]
189 assert entry["new_id"].startswith("sha256:"), \
190 f"new_id must have sha256: prefix, got {entry['new_id']!r}"
191
192 def test_new_id_is_full_sha256_in_json(self, tmp_path: pathlib.Path) -> None:
193 repo = _make_repo(tmp_path)
194 _append(repo, new_id=_SHA_A)
195 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
196 assert _SHA256_FULL.match(entry["new_id"]), \
197 f"new_id must be sha256:<64hex>, got {entry['new_id']!r}"
198
199 def test_old_id_sha256_prefixed_in_json(self, tmp_path: pathlib.Path) -> None:
200 repo = _make_repo(tmp_path)
201 _append(repo, old_id=_SHA_B, new_id=_SHA_A)
202 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
203 assert entry["old_id"].startswith("sha256:")
204
205 def test_old_id_is_full_sha256_in_json(self, tmp_path: pathlib.Path) -> None:
206 repo = _make_repo(tmp_path)
207 _append(repo, old_id=_SHA_B, new_id=_SHA_A)
208 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
209 assert _SHA256_FULL.match(entry["old_id"])
210
211 def test_null_old_id_sha256_zeros_in_json(self, tmp_path: pathlib.Path) -> None:
212 """Initial commit: old_id = sha256:0000…0000 (64 zeros)."""
213 repo = _make_repo(tmp_path)
214 _append(repo, old_id=_NULL_ID)
215 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
216 assert entry["old_id"] == NULL_LONG_ID
217
218 def test_new_id_value_round_trips(self, tmp_path: pathlib.Path) -> None:
219 """sha256: prefix wraps the exact bare hex stored in the reflog."""
220 repo = _make_repo(tmp_path)
221 _append(repo, new_id=_SHA_B)
222 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
223 assert entry["new_id"] == long_id(_SHA_B)
224
225
226 # ---------------------------------------------------------------------------
227 # Integration — duration_ms and exit_code
228 # ---------------------------------------------------------------------------
229
230
231 class TestDurationAndExitCode:
232 def test_duration_ms_present_in_json(self, tmp_path: pathlib.Path) -> None:
233 repo = _make_repo(tmp_path)
234 _append(repo)
235 data = json.loads(_invoke(repo, "--json").output)
236 assert "duration_ms" in data
237
238 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
239 repo = _make_repo(tmp_path)
240 _append(repo)
241 data = json.loads(_invoke(repo, "--json").output)
242 assert data["exit_code"] == 0
243
244 def test_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None:
245 repo = _make_repo(tmp_path)
246 _append(repo)
247 data = json.loads(_invoke(repo, "--json").output)
248 assert isinstance(data["duration_ms"], float)
249
250 def test_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None:
251 repo = _make_repo(tmp_path)
252 _append(repo)
253 assert json.loads(_invoke(repo, "--json").output)["duration_ms"] >= 0.0
254
255 def test_duration_ms_3dp_precision(self, tmp_path: pathlib.Path) -> None:
256 repo = _make_repo(tmp_path)
257 _append(repo)
258 ms = json.loads(_invoke(repo, "--json").output)["duration_ms"]
259 assert round(ms, 3) == ms
260
261 def test_all_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
262 repo = _make_repo(tmp_path)
263 _append(repo, branch="main")
264 data = json.loads(_invoke(repo, "--all", "--json").output)
265 assert "duration_ms" in data
266 assert data["exit_code"] == 0
267
268 def test_filtered_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
269 repo = _make_repo(tmp_path)
270 _append(repo, operation="commit: feature")
271 _append(repo, operation="checkout: dev",
272 timestamp=_TS + datetime.timedelta(seconds=1))
273 data = json.loads(_invoke(repo, "--json", "--operation", "commit").output)
274 assert "duration_ms" in data
275 assert data["exit_code"] == 0
276
277 def test_empty_reflog_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
278 """Even with no entries the JSON output must include duration_ms."""
279 repo = _make_repo(tmp_path)
280 data = json.loads(_invoke(repo, "--json").output)
281 assert "duration_ms" in data
282 assert data["exit_code"] == 0
283
284
285 # ---------------------------------------------------------------------------
286 # Filter behaviour
287 # ---------------------------------------------------------------------------
288
289
290 class TestFilterBehaviour:
291 def test_total_reflects_post_filter_count(self, tmp_path: pathlib.Path) -> None:
292 """total in JSON is the number of entries that pass all filters,
293 before --limit is applied."""
294 repo = _make_repo(tmp_path)
295 for i in range(5):
296 _append(repo, operation="commit: work",
297 timestamp=_TS + datetime.timedelta(seconds=i))
298 for i in range(3):
299 _append(repo, operation="checkout: branch",
300 timestamp=_TS + datetime.timedelta(seconds=10 + i))
301 data = json.loads(_invoke(repo, "--json", "--operation", "commit", "--limit", "2").output)
302 assert data["total"] == 5, "total must count all matching entries, not just displayed"
303 assert len(data["entries"]) == 2, "entries must be capped by --limit"
304
305 def test_since_until_single_day(self, tmp_path: pathlib.Path) -> None:
306 """--since and --until set to same day returns entries on that day."""
307 repo = _make_repo(tmp_path)
308 day = datetime.datetime(2026, 3, 10, tzinfo=datetime.timezone.utc)
309 _append(repo, operation="commit: on-day", timestamp=day)
310 _append(repo, operation="commit: day-before",
311 timestamp=day - datetime.timedelta(days=1))
312 _append(repo, operation="commit: day-after",
313 timestamp=day + datetime.timedelta(days=1))
314 data = json.loads(
315 _invoke(repo, "--json", "--since", "2026-03-10", "--until", "2026-03-10").output
316 )
317 assert data["total"] == 1
318 assert data["entries"][0]["operation"] == "commit: on-day"
319
320 def test_since_after_until_errors(self, tmp_path: pathlib.Path) -> None:
321 """--since after --until must exit USER_ERROR."""
322 repo = _make_repo(tmp_path)
323 result = _invoke(repo, "--since", "2026-06-01", "--until", "2026-01-01")
324 assert result.exit_code == ExitCode.USER_ERROR
325
326 def test_limit_applied_after_all_filters(self, tmp_path: pathlib.Path) -> None:
327 """--limit caps displayed entries but total reflects full filtered count."""
328 repo = _make_repo(tmp_path)
329 for i in range(10):
330 _append(repo, operation="commit: x",
331 timestamp=_TS + datetime.timedelta(seconds=i))
332 data = json.loads(_invoke(repo, "--json", "--limit", "3").output)
333 assert data["total"] == 10
334 assert len(data["entries"]) == 3
335 assert data["limit"] == 3
336
337 def test_operation_and_author_filters_combined(self, tmp_path: pathlib.Path) -> None:
338 repo = _make_repo(tmp_path)
339 _append(repo, author="alice", operation="commit: feature")
340 _append(repo, author="bob", operation="commit: feature",
341 timestamp=_TS + datetime.timedelta(seconds=1))
342 _append(repo, author="alice", operation="checkout: main",
343 timestamp=_TS + datetime.timedelta(seconds=2))
344 data = json.loads(
345 _invoke(repo, "--json", "--operation", "commit", "--author", "alice").output
346 )
347 assert data["total"] == 1
348 assert data["entries"][0]["author"] == "alice"
349 assert "commit" in data["entries"][0]["operation"]
350
351
352 # ---------------------------------------------------------------------------
353 # Security
354 # ---------------------------------------------------------------------------
355
356
357 class TestSecuritySupercharge:
358 def test_ansi_in_new_id_sanitized_in_text(self, tmp_path: pathlib.Path) -> None:
359 """ANSI in a stored new_id is stripped before terminal output."""
360 repo = _make_repo(tmp_path)
361 malicious_id = f"\x1b[31m{'a' * 60}" # starts with ANSI, then hex
362 _append(repo, new_id=malicious_id)
363 result = _invoke(repo)
364 assert result.exit_code == 0
365 assert "\x1b" not in result.output
366
367 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
368 repo = _make_repo(tmp_path)
369 result = _invoke(repo, "--format", "msgpack")
370 assert result.exit_code in (ExitCode.USER_ERROR, 2)
371 assert "Traceback" not in result.output
372
373 def test_no_traceback_on_bad_date(self, tmp_path: pathlib.Path) -> None:
374 repo = _make_repo(tmp_path)
375 result = _invoke(repo, "--since", "not-a-date")
376 assert result.exit_code == ExitCode.USER_ERROR
377 assert "Traceback" not in result.output
378
379
380 # ---------------------------------------------------------------------------
381 # Performance
382 # ---------------------------------------------------------------------------
383
384
385 class TestPerformanceSupercharge:
386 def test_empty_reflog_under_100ms(self, tmp_path: pathlib.Path) -> None:
387 repo = _make_repo(tmp_path)
388 t0 = time.monotonic()
389 result = _invoke(repo, "--json")
390 duration_ms = (time.monotonic() - t0) * 1000
391 assert result.exit_code == 0
392 assert duration_ms < 100
393
394 def test_100_entries_under_500ms(self, tmp_path: pathlib.Path) -> None:
395 repo = _make_repo(tmp_path)
396 for i in range(100):
397 _append(repo, operation=f"commit: entry {i}",
398 timestamp=_TS + datetime.timedelta(seconds=i))
399 t0 = time.monotonic()
400 result = _invoke(repo, "--json", "--limit", "100")
401 duration_ms = (time.monotonic() - t0) * 1000
402 assert result.exit_code == 0
403 assert duration_ms < 500
404
405 def test_duration_ms_plausible(self, tmp_path: pathlib.Path) -> None:
406 repo = _make_repo(tmp_path)
407 _append(repo)
408 data = json.loads(_invoke(repo, "--json").output)
409 assert data["duration_ms"] < 500
File History 11 commits
sha256:37af4ab271a64f11d7bf14b4871ebaaf5dd68fb0247521007c6b034d35aa01be Merge branch 'fix/hub-user-read-update-path' into dev Human 10 days ago
sha256:b7be56ec091919a612cffe7f3c8b600209d5155517443fdac0e16954c8c7d81f Merge 'task/git-export-ignored-file-deletion' into 'dev' — … Human 12 days ago
sha256:8de4334a98c945aace420969d389ad678aa926d4ab4e886b2ac4c4241cb3bf2b revert: keep pyproject.toml in canonical PEP 440 form Sonnet 4.6 patch 15 days ago
sha256:50bb615c573aa4b928fca75b60f82ba2a659910066507cec6a95c412ae22ccb9 docs: add issue docs for push have-negotiation bug (#55) an… Sonnet 4.6 19 days ago
sha256:d90d175cded68aae1d4ffcf4858917854195d0cd8ce1fe73cee4dbc02541cb74 chore: trigger push to surface null-OID paths Sonnet 4.6 24 days ago
sha256:e452ad9a6ace6ccc6d875a35e06caf9da5576a970c1c36133b69a891ce5fefa8 chore: prebuild timing test Sonnet 4.6 35 days ago
sha256:0008ab6695e3e064b3e236b24fd19e538fef6a588eb0d211622f4466d919c0b1 merge: pull staging/dev — advance to 0.2.0rc12 Sonnet 4.6 patch 36 days ago
sha256:9c33d61749fff814c5226d5386aa2af7064c2c02788594a25fdd709358132eea fix: _PROPOSAL_PREFIX_RESOLVE_LIMIT 200 → 100 to match hub … Sonnet 4.6 48 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 50 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 57 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor 57 days ago