gabriel / muse public
test_log_supercharge.py python
493 lines 19.0 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 6 days ago
1 """Supercharge tests for ``muse log``.
2
3 Coverage tiers
4 --------------
5 I JSON envelope schema — status, error, branch, repo_id, total, duration_ms, exit_code
6 II Error payload shape — consistent {status, error, exit_code}; no prose in JSON mode
7 III Canonical ref resolution — sha256: prefix accepted by muse log
8 IV Data integrity — all commit_ids in response are sha256:-prefixed
9 V Truncation behaviour — explicit -n suppresses warning; probe_truncated flag set correctly
10 VI TypedDicts — _LogJson and _LogErrorJson exist with correct annotations
11 VII Docstring — documents new envelope fields
12 VIII No prose pollution in --json mode
13 """
14 from __future__ import annotations
15 from collections.abc import Mapping
16
17 import json
18 import os
19 import pathlib
20 import sys
21
22 import pytest
23
24 from tests.cli_test_helper import CliRunner, InvokeResult
25 from muse.core.types import long_id
26
27 runner = CliRunner()
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33 _REQUIRED_ENVELOPE_KEYS = {
34 "status", "error", "truncated", "total", "branch",
35 "repo_id", "commits", "duration_ms", "exit_code",
36 "timestamp", "warnings", "schema", "muse_version",
37 }
38
39
40 def _env(root: pathlib.Path) -> Mapping[str, str]:
41 return {"MUSE_REPO_ROOT": str(root)}
42
43
44 def _invoke(root: pathlib.Path, args: list[str]) -> InvokeResult:
45 saved = os.getcwd()
46 try:
47 os.chdir(root)
48 return runner.invoke(None, args, env=_env(root))
49 finally:
50 os.chdir(saved)
51
52
53 def _log(root: pathlib.Path, *extra: str) -> InvokeResult:
54 return _invoke(root, ["log", *extra])
55
56
57 def _log_json(root: pathlib.Path, *extra: str) -> Mapping[str, object]:
58 result = _log(root, "--json", *extra)
59 assert result.exit_code == 0, f"log --json failed:\n{result.output}"
60 return json.loads(result.output.strip())
61
62
63 @pytest.fixture()
64 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
65 """Initialised code repo with two commits."""
66 os.chdir(tmp_path)
67 r = _invoke(tmp_path, ["init"])
68 assert r.exit_code == 0, r.output
69 (tmp_path / "a.py").write_text("a = 1\n")
70 _invoke(tmp_path, ["code", "add", "a.py"])
71 r = _invoke(tmp_path, ["commit", "-m", "first"])
72 assert r.exit_code == 0, r.output
73 (tmp_path / "b.py").write_text("b = 2\n")
74 _invoke(tmp_path, ["code", "add", "b.py"])
75 r = _invoke(tmp_path, ["commit", "-m", "second"])
76 assert r.exit_code == 0, r.output
77 return tmp_path
78
79
80 @pytest.fixture()
81 def single_commit_repo(tmp_path: pathlib.Path) -> pathlib.Path:
82 os.chdir(tmp_path)
83 _invoke(tmp_path, ["init"])
84 (tmp_path / "x.py").write_text("x = 0\n")
85 _invoke(tmp_path, ["code", "add", "x.py"])
86 r = _invoke(tmp_path, ["commit", "-m", "init commit"])
87 assert r.exit_code == 0, r.output
88 return tmp_path
89
90
91 # ---------------------------------------------------------------------------
92 # I JSON envelope schema
93 # ---------------------------------------------------------------------------
94
95
96 class TestJsonEnvelopeSchema:
97 def test_all_required_envelope_keys_present(self, repo: pathlib.Path) -> None:
98 """Envelope always has all required top-level keys."""
99 data = _log_json(repo)
100 missing = _REQUIRED_ENVELOPE_KEYS - set(data.keys())
101 assert not missing, f"Missing envelope keys: {missing}"
102
103 def test_no_extra_undocumented_keys(self, repo: pathlib.Path) -> None:
104 """Envelope contains no undocumented extra keys."""
105 data = _log_json(repo)
106 extra = set(data.keys()) - _REQUIRED_ENVELOPE_KEYS
107 assert not extra, f"Undocumented extra keys: {extra}"
108
109 def test_status_is_ok_on_success(self, repo: pathlib.Path) -> None:
110 data = _log_json(repo)
111 assert data["status"] == "ok"
112
113 def test_error_is_empty_string_on_success(self, repo: pathlib.Path) -> None:
114 data = _log_json(repo)
115 assert data["error"] == ""
116
117 def test_exit_code_is_zero_on_success(self, repo: pathlib.Path) -> None:
118 data = _log_json(repo)
119 assert data["exit_code"] == 0
120
121 def test_branch_field_matches_current_branch(self, repo: pathlib.Path) -> None:
122 data = _log_json(repo)
123 assert data["branch"] == "main"
124
125 def test_branch_field_respects_explicit_ref(self, tmp_path: pathlib.Path) -> None:
126 os.chdir(tmp_path)
127 _invoke(tmp_path, ["init"])
128 (tmp_path / "f.py").write_text("f = 1\n")
129 _invoke(tmp_path, ["code", "add", "f.py"])
130 _invoke(tmp_path, ["commit", "-m", "base"])
131 _invoke(tmp_path, ["checkout", "-b", "dev"])
132 (tmp_path / "g.py").write_text("g = 2\n")
133 _invoke(tmp_path, ["code", "add", "g.py"])
134 _invoke(tmp_path, ["commit", "-m", "on dev"])
135 _invoke(tmp_path, ["checkout", "main"])
136 # Explicitly ask for dev log
137 data = _log_json(tmp_path, "dev")
138 assert data["branch"] == "dev"
139
140 def test_repo_id_is_nonempty_string(self, repo: pathlib.Path) -> None:
141 data = _log_json(repo)
142 assert isinstance(data["repo_id"], str)
143 assert len(data["repo_id"]) > 0
144
145 def test_total_matches_len_commits(self, repo: pathlib.Path) -> None:
146 data = _log_json(repo)
147 assert data["total"] == len(data["commits"])
148
149 def test_total_reflects_limit(self, repo: pathlib.Path) -> None:
150 data = _log_json(repo, "-n", "1")
151 assert data["total"] == 1
152 assert len(data["commits"]) == 1
153
154 def test_duration_ms_is_nonnegative_float(self, repo: pathlib.Path) -> None:
155 data = _log_json(repo)
156 assert isinstance(data["duration_ms"], float)
157 assert data["duration_ms"] >= 0.0
158
159 def test_commits_is_a_list(self, repo: pathlib.Path) -> None:
160 data = _log_json(repo)
161 assert isinstance(data["commits"], list)
162
163 def test_truncated_is_bool(self, repo: pathlib.Path) -> None:
164 data = _log_json(repo)
165 assert isinstance(data["truncated"], bool)
166
167 def test_empty_repo_returns_empty_commits(self, tmp_path: pathlib.Path) -> None:
168 os.chdir(tmp_path)
169 _invoke(tmp_path, ["init"])
170 data = _log_json(tmp_path)
171 assert data["status"] == "ok"
172 assert data["commits"] == []
173 assert data["total"] == 0
174
175
176 # ---------------------------------------------------------------------------
177 # II Error payload shape
178 # ---------------------------------------------------------------------------
179
180
181 class TestErrorPayloadShape:
182 def test_invalid_since_returns_error_payload(self, repo: pathlib.Path) -> None:
183 result = _log(repo, "--json", "--since", "not-a-date")
184 assert result.exit_code != 0
185 data = json.loads(result.output.strip())
186 assert data["status"] == "error"
187 assert isinstance(data["error"], str) and len(data["error"]) > 0
188 assert isinstance(data["exit_code"], int) and data["exit_code"] != 0
189
190 def test_invalid_until_returns_error_payload(self, repo: pathlib.Path) -> None:
191 result = _log(repo, "--json", "--until", "bad")
192 assert result.exit_code != 0
193 data = json.loads(result.output.strip())
194 assert data["status"] == "error"
195 assert "error" in data
196
197 def test_invalid_format_returns_error_payload(self, repo: pathlib.Path) -> None:
198 result = _log(repo, "--format", "xml")
199 assert result.exit_code != 0
200
201 def test_error_payload_has_exactly_three_keys(self, repo: pathlib.Path) -> None:
202 """Error payload: exactly {status, error, exit_code}."""
203 result = _log(repo, "--json", "--since", "not-a-date")
204 data = json.loads(result.output.strip())
205 assert {"status", "error", "exit_code"}.issubset(data.keys())
206
207 def test_no_prose_on_stderr_for_invalid_since_in_json_mode(
208 self, repo: pathlib.Path
209 ) -> None:
210 """When --json is active, errors go to JSON on stdout, not prose on stderr."""
211 result = _log(repo, "--json", "--since", "not-a-date")
212 # stdout must be valid JSON
213 data = json.loads(result.output.strip())
214 assert data["status"] == "error"
215 # No human-readable error emoji in stdout
216 assert "❌" not in result.output
217
218 def test_no_prose_on_stderr_for_invalid_until_in_json_mode(
219 self, repo: pathlib.Path
220 ) -> None:
221 result = _log(repo, "--json", "--until", "not-a-date")
222 data = json.loads(result.output.strip())
223 assert data["status"] == "error"
224 assert "❌" not in result.output
225
226
227 # ---------------------------------------------------------------------------
228 # III Canonical sha256: ref resolution
229 # ---------------------------------------------------------------------------
230
231
232 class TestSha256RefResolution:
233 def test_sha256_full_commit_id_as_ref(self, repo: pathlib.Path) -> None:
234 """muse log sha256:<cid> should walk from that commit, not fall into pathspec."""
235 # Get HEAD commit id
236 all_data = _log_json(repo)
237 head_cid = all_data["commits"][0]["commit_id"]
238 assert head_cid.startswith("sha256:")
239
240 # Walk from that specific commit
241 data = _log_json(repo, head_cid)
242 assert data["status"] == "ok"
243 assert any(c["commit_id"] == head_cid for c in data["commits"])
244
245 def test_sha256_prefix_as_ref(self, repo: pathlib.Path) -> None:
246 """sha256:<first-8-hex> should resolve and not be treated as a pathspec."""
247 all_data = _log_json(repo)
248 head_cid = all_data["commits"][0]["commit_id"]
249 short_ref = long_id(head_cid[7:15])# sha256: + 8 hex chars
250
251 data = _log_json(repo, short_ref)
252 assert data["status"] == "ok"
253 assert len(data["commits"]) >= 1
254
255 def test_sha256_ref_not_treated_as_pathspec(self, repo: pathlib.Path) -> None:
256 """When sha256:<cid> is given as ref, commits list must not be empty."""
257 all_data = _log_json(repo)
258 head_cid = all_data["commits"][0]["commit_id"]
259 data = _log_json(repo, head_cid)
260 # If it fell into pathspec, no commits would touch a file named sha256:…
261 assert len(data["commits"]) > 0
262
263
264 # ---------------------------------------------------------------------------
265 # IV Data integrity — all commit_ids are sha256:-prefixed
266 # ---------------------------------------------------------------------------
267
268
269 class TestDataIntegrity:
270 def test_all_commit_ids_sha256_prefixed(self, repo: pathlib.Path) -> None:
271 data = _log_json(repo)
272 for c in data["commits"]:
273 assert c["commit_id"].startswith("sha256:"), (
274 f"commit_id not sha256:-prefixed: {c['commit_id']!r}"
275 )
276
277 def test_parent_commit_id_sha256_or_null(self, repo: pathlib.Path) -> None:
278 data = _log_json(repo)
279 for c in data["commits"]:
280 pid = c["parent_commit_id"]
281 assert pid is None or pid.startswith("sha256:"), (
282 f"parent_commit_id not sha256:-prefixed: {pid!r}"
283 )
284
285 def test_snapshot_id_sha256_or_null(self, repo: pathlib.Path) -> None:
286 data = _log_json(repo)
287 for c in data["commits"]:
288 sid = c["snapshot_id"]
289 assert sid is None or sid.startswith("sha256:"), (
290 f"snapshot_id not sha256:-prefixed: {sid!r}"
291 )
292
293 def test_committed_at_is_iso8601(self, repo: pathlib.Path) -> None:
294 from datetime import datetime
295 data = _log_json(repo)
296 for c in data["commits"]:
297 ts = c["committed_at"]
298 # Must parse as ISO-8601
299 datetime.fromisoformat(ts)
300
301 def test_files_added_is_list(self, repo: pathlib.Path) -> None:
302 data = _log_json(repo)
303 for c in data["commits"]:
304 assert isinstance(c["files_added"], list)
305
306 def test_files_modified_is_list(self, repo: pathlib.Path) -> None:
307 data = _log_json(repo)
308 for c in data["commits"]:
309 assert isinstance(c["files_modified"], list)
310
311 def test_files_removed_is_list(self, repo: pathlib.Path) -> None:
312 data = _log_json(repo)
313 for c in data["commits"]:
314 assert isinstance(c["files_removed"], list)
315
316 def test_agent_id_is_string(self, repo: pathlib.Path) -> None:
317 data = _log_json(repo)
318 for c in data["commits"]:
319 assert isinstance(c["agent_id"], str)
320
321 def test_model_id_is_string(self, repo: pathlib.Path) -> None:
322 data = _log_json(repo)
323 for c in data["commits"]:
324 assert isinstance(c["model_id"], str)
325
326
327 # ---------------------------------------------------------------------------
328 # V Truncation behaviour
329 # ---------------------------------------------------------------------------
330
331
332 class TestTruncationBehaviour:
333 def test_explicit_n_does_not_emit_warning_in_text(
334 self, tmp_path: pathlib.Path
335 ) -> None:
336 """When user explicitly passes -n, no truncation warning is printed."""
337 os.chdir(tmp_path)
338 _invoke(tmp_path, ["init"])
339 for i in range(5):
340 (tmp_path / f"f{i}.py").write_text(f"x={i}\n")
341 _invoke(tmp_path, ["code", "add", f"f{i}.py"])
342 _invoke(tmp_path, ["commit", "-m", f"commit {i}"])
343
344 result = _log(tmp_path, "--oneline", "-n", "2")
345 assert result.exit_code == 0
346 assert "truncated" not in result.output.lower(), (
347 f"Unexpected truncation warning when -n was explicit:\n{result.output}"
348 )
349 lines = [l for l in result.output.strip().splitlines() if l.strip()]
350 assert len(lines) == 2
351
352 def test_truncated_true_in_json_when_limit_hit(
353 self, tmp_path: pathlib.Path
354 ) -> None:
355 """truncated=true in JSON envelope when there are more commits than limit."""
356 os.chdir(tmp_path)
357 _invoke(tmp_path, ["init"])
358 for i in range(4):
359 (tmp_path / f"f{i}.py").write_text(f"x={i}\n")
360 _invoke(tmp_path, ["code", "add", f"f{i}.py"])
361 _invoke(tmp_path, ["commit", "-m", f"commit {i}"])
362
363 data = _log_json(tmp_path, "-n", "2")
364 assert data["truncated"] is True
365 assert data["total"] == 2
366
367 def test_truncated_false_when_all_commits_returned(
368 self, repo: pathlib.Path
369 ) -> None:
370 data = _log_json(repo)
371 assert data["truncated"] is False
372
373 def test_default_limit_truncation_warning_fires_in_text(
374 self, tmp_path: pathlib.Path
375 ) -> None:
376 """When default limit is hit (not explicitly set), warning is shown."""
377 os.chdir(tmp_path)
378 _invoke(tmp_path, ["init"])
379 for i in range(3):
380 (tmp_path / f"f{i}.py").write_text(f"x={i}\n")
381 _invoke(tmp_path, ["code", "add", f"f{i}.py"])
382 _invoke(tmp_path, ["commit", "-m", f"commit {i}"])
383
384 # Patch the default limit to 2 so the warning fires on a 3-commit repo
385 import muse.cli.commands.log as log_mod
386 orig = log_mod._DEFAULT_LIMIT
387 try:
388 log_mod._DEFAULT_LIMIT = 2
389 result = _log(tmp_path, "--oneline")
390 # Warning should appear because default was hit, not explicit -n
391 assert "truncated" in result.output.lower()
392 finally:
393 log_mod._DEFAULT_LIMIT = orig
394
395
396 # ---------------------------------------------------------------------------
397 # VI TypedDicts
398 # ---------------------------------------------------------------------------
399
400
401 class TestTypedDicts:
402 def test_log_json_typed_dict_exists(self) -> None:
403 from muse.cli.commands.log import _LogJson # type: ignore[attr-defined]
404 assert _LogJson is not None
405
406 def test_log_error_json_typed_dict_exists(self) -> None:
407 from muse.cli.commands.log import _LogErrorJson # type: ignore[attr-defined]
408 assert _LogErrorJson is not None
409
410 def test_log_json_has_required_annotations(self) -> None:
411 from muse.cli.commands.log import _LogJson # type: ignore[attr-defined]
412 hints = _LogJson.__annotations__
413 required = {"status", "error", "truncated", "total", "branch",
414 "repo_id", "commits", "duration_ms", "exit_code"}
415 missing = required - set(hints)
416 assert not missing, f"_LogJson missing annotations: {missing}"
417
418 def test_log_error_json_has_required_annotations(self) -> None:
419 from muse.cli.commands.log import _LogErrorJson # type: ignore[attr-defined]
420 hints = _LogErrorJson.__annotations__
421 required = {"status", "error", "exit_code"}
422 missing = required - set(hints)
423 assert not missing, f"_LogErrorJson missing annotations: {missing}"
424
425
426 # ---------------------------------------------------------------------------
427 # VII Docstring
428 # ---------------------------------------------------------------------------
429
430
431 class TestDocstring:
432 def test_module_docstring_documents_status(self) -> None:
433 import muse.cli.commands.log as log_mod
434 assert "status" in (log_mod.__doc__ or "")
435
436 def test_module_docstring_documents_branch(self) -> None:
437 import muse.cli.commands.log as log_mod
438 assert '"branch"' in (log_mod.__doc__ or "")
439
440 def test_module_docstring_documents_repo_id(self) -> None:
441 import muse.cli.commands.log as log_mod
442 assert '"repo_id"' in (log_mod.__doc__ or "")
443
444 def test_module_docstring_documents_total(self) -> None:
445 import muse.cli.commands.log as log_mod
446 assert '"total"' in (log_mod.__doc__ or "")
447
448 def test_module_docstring_documents_duration_ms(self) -> None:
449 import muse.cli.commands.log as log_mod
450 assert "duration_ms" in (log_mod.__doc__ or "")
451
452 def test_module_docstring_documents_exit_code(self) -> None:
453 import muse.cli.commands.log as log_mod
454 assert "exit_code" in (log_mod.__doc__ or "")
455
456
457 # ---------------------------------------------------------------------------
458 # VIII No prose pollution in --json mode
459 # ---------------------------------------------------------------------------
460
461
462 class TestNoProsePollution:
463 def test_success_stdout_is_valid_json(self, repo: pathlib.Path) -> None:
464 result = _log(repo, "--json")
465 assert result.exit_code == 0
466 data = json.loads(result.output.strip()) # must not raise
467 assert isinstance(data, dict)
468
469 def test_no_emoji_in_json_success_output(self, repo: pathlib.Path) -> None:
470 result = _log(repo, "--json")
471 assert "✅" not in result.output
472 assert "⚠️" not in result.output
473
474 def test_no_emoji_in_json_error_output(self, repo: pathlib.Path) -> None:
475 result = _log(repo, "--json", "--since", "bad-date")
476 # Must still be parseable JSON
477 data = json.loads(result.output.strip())
478 assert "❌" not in result.output
479 assert data["status"] == "error"
480
481 def test_ansi_in_commit_message_is_json_escaped(
482 self, tmp_path: pathlib.Path
483 ) -> None:
484 """ANSI escape in commit message must be JSON-encoded, not echoed raw."""
485 os.chdir(tmp_path)
486 _invoke(tmp_path, ["init"])
487 (tmp_path / "malicious.py").write_text("x = 1\n")
488 _invoke(tmp_path, ["code", "add", "malicious.py"])
489 _invoke(tmp_path, ["commit", "-m", "safe\x1b[31mmalicious\x1b[0m"])
490 result = _log(tmp_path, "--json")
491 assert "\x1b" not in result.output
492 data = json.loads(result.output.strip())
493 assert data["status"] == "ok"
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 6 days ago