gabriel / muse public
test_reflog_supercharge.py python
405 lines 16.5 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 133 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, fake_id, short_id as _short_id
24
25 runner = CliRunner()
26
27 _NULL_ID = "0" * 64
28 _SHA_A = long_id("a" * 64)
29 _SHA_B = long_id("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 result = _short_id(_SHA_A)
105 assert result.startswith("sha256:")
106
107 def test_bare_hex_12_hex_chars_after_prefix(self) -> None:
108 result = _short_id(_SHA_A)
109 assert result == long_id("a" * 12)
110
111 def test_bare_hex_total_length_is_19(self) -> None:
112 assert len(_short_id(_SHA_B)) == 19
113
114 def test_sha256_prefixed_input_handled(self) -> None:
115 prefixed = long_id("deadbeef" * 8)
116 result = _short_id(prefixed)
117 assert result.startswith("sha256:")
118 assert len(result) == 19
119
120 def test_null_id_shows_zeros(self) -> None:
121 result = _short_id(_NULL_ID)
122 assert result == "0" * 12
123
124 def test_matches_short_regex(self) -> None:
125 assert _SHA256_SHORT_19.match(_short_id(_SHA_A))
126
127
128 # ---------------------------------------------------------------------------
129 # Integration — text format short IDs
130 # ---------------------------------------------------------------------------
131
132
133 class TestTextFormatShortId:
134 """Text format must show sha256:<12-hex> for new_id and old_id."""
135
136 def _short_tokens(self, line: str) -> list[str]:
137 return [tok for tok in line.split() if _SHA256_SHORT_19.match(tok)]
138
139 def test_new_id_shown_as_sha256_short_in_text(self, tmp_path: pathlib.Path) -> None:
140 repo = _make_repo(tmp_path)
141 _append(repo, new_id=_SHA_A)
142 result = _invoke(repo)
143 assert result.exit_code == 0
144 tokens = self._short_tokens(result.output)
145 assert any(t.startswith(long_id("a" * 12)) for t in tokens), \
146 f"no sha256:aaa… token in text output:\n{result.output}"
147
148 def test_old_id_shown_as_sha256_short_in_text(self, tmp_path: pathlib.Path) -> None:
149 repo = _make_repo(tmp_path)
150 _append(repo, old_id=_SHA_B, new_id=_SHA_A)
151 result = _invoke(repo)
152 assert result.exit_code == 0
153 assert long_id("b" * 12) in result.output, \
154 f"sha256:bbb… not in text output:\n{result.output}"
155
156 def test_initial_entry_shows_initial_keyword(self, tmp_path: pathlib.Path) -> None:
157 """Null old_id must render as 'initial', not sha256:000…."""
158 repo = _make_repo(tmp_path)
159 _append(repo, old_id=_NULL_ID)
160 result = _invoke(repo)
161 assert "initial" in result.output
162
163 def test_text_short_id_length_is_19(self, tmp_path: pathlib.Path) -> None:
164 repo = _make_repo(tmp_path)
165 _append(repo, new_id=_SHA_A)
166 result = _invoke(repo)
167 tokens = self._short_tokens(result.output)
168 for tok in tokens:
169 assert len(tok) == 19, f"short ID token has wrong length: {tok!r}"
170
171
172 # ---------------------------------------------------------------------------
173 # Data integrity — JSON IDs
174 # ---------------------------------------------------------------------------
175
176
177 class TestJsonIds:
178 """JSON new_id / old_id must be sha256:<64-hex> canonical form."""
179
180 def test_new_id_sha256_prefixed_in_json(self, tmp_path: pathlib.Path) -> None:
181 repo = _make_repo(tmp_path)
182 _append(repo, new_id=_SHA_A)
183 data = json.loads(_invoke(repo, "--json").output)
184 entry = data["entries"][0]
185 assert entry["new_id"].startswith("sha256:"), \
186 f"new_id must have sha256: prefix, got {entry['new_id']!r}"
187
188 def test_new_id_is_full_sha256_in_json(self, tmp_path: pathlib.Path) -> None:
189 repo = _make_repo(tmp_path)
190 _append(repo, new_id=_SHA_A)
191 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
192 assert _SHA256_FULL.match(entry["new_id"]), \
193 f"new_id must be sha256:<64hex>, got {entry['new_id']!r}"
194
195 def test_old_id_sha256_prefixed_in_json(self, tmp_path: pathlib.Path) -> None:
196 repo = _make_repo(tmp_path)
197 _append(repo, old_id=_SHA_B, new_id=_SHA_A)
198 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
199 assert entry["old_id"].startswith("sha256:")
200
201 def test_old_id_is_full_sha256_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 _SHA256_FULL.match(entry["old_id"])
206
207 def test_null_old_id_sha256_zeros_in_json(self, tmp_path: pathlib.Path) -> None:
208 """Initial commit: old_id = sha256:0000…0000 (64 zeros)."""
209 repo = _make_repo(tmp_path)
210 _append(repo, old_id=_NULL_ID)
211 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
212 assert entry["old_id"] == long_id("0" * 64)
213
214 def test_new_id_value_round_trips(self, tmp_path: pathlib.Path) -> None:
215 """sha256: prefix wraps the exact bare hex stored in the reflog."""
216 repo = _make_repo(tmp_path)
217 _append(repo, new_id=_SHA_B)
218 entry = json.loads(_invoke(repo, "--json").output)["entries"][0]
219 assert entry["new_id"] == long_id(_SHA_B)
220
221
222 # ---------------------------------------------------------------------------
223 # Integration — duration_ms and exit_code
224 # ---------------------------------------------------------------------------
225
226
227 class TestDurationAndExitCode:
228 def test_duration_ms_present_in_json(self, tmp_path: pathlib.Path) -> None:
229 repo = _make_repo(tmp_path)
230 _append(repo)
231 data = json.loads(_invoke(repo, "--json").output)
232 assert "duration_ms" in data
233
234 def test_exit_code_zero_on_success(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 data["exit_code"] == 0
239
240 def test_duration_ms_is_float(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 isinstance(data["duration_ms"], float)
245
246 def test_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None:
247 repo = _make_repo(tmp_path)
248 _append(repo)
249 assert json.loads(_invoke(repo, "--json").output)["duration_ms"] >= 0.0
250
251 def test_duration_ms_3dp_precision(self, tmp_path: pathlib.Path) -> None:
252 repo = _make_repo(tmp_path)
253 _append(repo)
254 ms = json.loads(_invoke(repo, "--json").output)["duration_ms"]
255 assert round(ms, 3) == ms
256
257 def test_all_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
258 repo = _make_repo(tmp_path)
259 _append(repo, branch="main")
260 data = json.loads(_invoke(repo, "--all", "--json").output)
261 assert "duration_ms" in data
262 assert data["exit_code"] == 0
263
264 def test_filtered_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
265 repo = _make_repo(tmp_path)
266 _append(repo, operation="commit: feature")
267 _append(repo, operation="checkout: dev",
268 timestamp=_TS + datetime.timedelta(seconds=1))
269 data = json.loads(_invoke(repo, "--json", "--operation", "commit").output)
270 assert "duration_ms" in data
271 assert data["exit_code"] == 0
272
273 def test_empty_reflog_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
274 """Even with no entries the JSON output must include duration_ms."""
275 repo = _make_repo(tmp_path)
276 data = json.loads(_invoke(repo, "--json").output)
277 assert "duration_ms" in data
278 assert data["exit_code"] == 0
279
280
281 # ---------------------------------------------------------------------------
282 # Filter behaviour
283 # ---------------------------------------------------------------------------
284
285
286 class TestFilterBehaviour:
287 def test_total_reflects_post_filter_count(self, tmp_path: pathlib.Path) -> None:
288 """total in JSON is the number of entries that pass all filters,
289 before --limit is applied."""
290 repo = _make_repo(tmp_path)
291 for i in range(5):
292 _append(repo, operation="commit: work",
293 timestamp=_TS + datetime.timedelta(seconds=i))
294 for i in range(3):
295 _append(repo, operation="checkout: branch",
296 timestamp=_TS + datetime.timedelta(seconds=10 + i))
297 data = json.loads(_invoke(repo, "--json", "--operation", "commit", "--limit", "2").output)
298 assert data["total"] == 5, "total must count all matching entries, not just displayed"
299 assert len(data["entries"]) == 2, "entries must be capped by --limit"
300
301 def test_since_until_single_day(self, tmp_path: pathlib.Path) -> None:
302 """--since and --until set to same day returns entries on that day."""
303 repo = _make_repo(tmp_path)
304 day = datetime.datetime(2026, 3, 10, tzinfo=datetime.timezone.utc)
305 _append(repo, operation="commit: on-day", timestamp=day)
306 _append(repo, operation="commit: day-before",
307 timestamp=day - datetime.timedelta(days=1))
308 _append(repo, operation="commit: day-after",
309 timestamp=day + datetime.timedelta(days=1))
310 data = json.loads(
311 _invoke(repo, "--json", "--since", "2026-03-10", "--until", "2026-03-10").output
312 )
313 assert data["total"] == 1
314 assert data["entries"][0]["operation"] == "commit: on-day"
315
316 def test_since_after_until_errors(self, tmp_path: pathlib.Path) -> None:
317 """--since after --until must exit USER_ERROR."""
318 repo = _make_repo(tmp_path)
319 result = _invoke(repo, "--since", "2026-06-01", "--until", "2026-01-01")
320 assert result.exit_code == ExitCode.USER_ERROR
321
322 def test_limit_applied_after_all_filters(self, tmp_path: pathlib.Path) -> None:
323 """--limit caps displayed entries but total reflects full filtered count."""
324 repo = _make_repo(tmp_path)
325 for i in range(10):
326 _append(repo, operation="commit: x",
327 timestamp=_TS + datetime.timedelta(seconds=i))
328 data = json.loads(_invoke(repo, "--json", "--limit", "3").output)
329 assert data["total"] == 10
330 assert len(data["entries"]) == 3
331 assert data["limit"] == 3
332
333 def test_operation_and_author_filters_combined(self, tmp_path: pathlib.Path) -> None:
334 repo = _make_repo(tmp_path)
335 _append(repo, author="alice", operation="commit: feature")
336 _append(repo, author="bob", operation="commit: feature",
337 timestamp=_TS + datetime.timedelta(seconds=1))
338 _append(repo, author="alice", operation="checkout: main",
339 timestamp=_TS + datetime.timedelta(seconds=2))
340 data = json.loads(
341 _invoke(repo, "--json", "--operation", "commit", "--author", "alice").output
342 )
343 assert data["total"] == 1
344 assert data["entries"][0]["author"] == "alice"
345 assert "commit" in data["entries"][0]["operation"]
346
347
348 # ---------------------------------------------------------------------------
349 # Security
350 # ---------------------------------------------------------------------------
351
352
353 class TestSecuritySupercharge:
354 def test_ansi_in_new_id_sanitized_in_text(self, tmp_path: pathlib.Path) -> None:
355 """ANSI in a stored new_id is stripped before terminal output."""
356 repo = _make_repo(tmp_path)
357 evil_id = "\x1b[31m" + "a" * 60 # starts with ANSI, then hex
358 _append(repo, new_id=evil_id)
359 result = _invoke(repo)
360 assert result.exit_code == 0
361 assert "\x1b" not in result.output
362
363 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
364 repo = _make_repo(tmp_path)
365 result = _invoke(repo, "--format", "msgpack")
366 assert result.exit_code in (ExitCode.USER_ERROR, 2)
367 assert "Traceback" not in result.output
368
369 def test_no_traceback_on_bad_date(self, tmp_path: pathlib.Path) -> None:
370 repo = _make_repo(tmp_path)
371 result = _invoke(repo, "--since", "not-a-date")
372 assert result.exit_code == ExitCode.USER_ERROR
373 assert "Traceback" not in result.output
374
375
376 # ---------------------------------------------------------------------------
377 # Performance
378 # ---------------------------------------------------------------------------
379
380
381 class TestPerformanceSupercharge:
382 def test_empty_reflog_under_100ms(self, tmp_path: pathlib.Path) -> None:
383 repo = _make_repo(tmp_path)
384 t0 = time.monotonic()
385 result = _invoke(repo, "--json")
386 duration_ms = (time.monotonic() - t0) * 1000
387 assert result.exit_code == 0
388 assert duration_ms < 100
389
390 def test_100_entries_under_500ms(self, tmp_path: pathlib.Path) -> None:
391 repo = _make_repo(tmp_path)
392 for i in range(100):
393 _append(repo, operation=f"commit: entry {i}",
394 timestamp=_TS + datetime.timedelta(seconds=i))
395 t0 = time.monotonic()
396 result = _invoke(repo, "--json", "--limit", "100")
397 duration_ms = (time.monotonic() - t0) * 1000
398 assert result.exit_code == 0
399 assert duration_ms < 500
400
401 def test_duration_ms_plausible(self, tmp_path: pathlib.Path) -> None:
402 repo = _make_repo(tmp_path)
403 _append(repo)
404 data = json.loads(_invoke(repo, "--json").output)
405 assert data["duration_ms"] < 500
File History 2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 133 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 139 days ago