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