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