"""Supercharge tests for ``muse log``. Coverage tiers -------------- I JSON envelope schema — status, error, branch, repo_id, total, duration_ms, exit_code II Error payload shape — consistent {status, error, exit_code}; no prose in JSON mode III Canonical ref resolution — sha256: prefix accepted by muse log IV Data integrity — all commit_ids in response are sha256:-prefixed V Truncation behaviour — explicit -n suppresses warning; probe_truncated flag set correctly VI TypedDicts — _LogJson and _LogErrorJson exist with correct annotations VII Docstring — documents new envelope fields VIII No prose pollution in --json mode """ from __future__ import annotations from collections.abc import Mapping import json import os import pathlib import sys import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.types import long_id runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _REQUIRED_ENVELOPE_KEYS = { "status", "error", "truncated", "total", "branch", "repo_id", "commits", "duration_ms", "exit_code", "timestamp", "warnings", "schema", "muse_version", } def _env(root: pathlib.Path) -> Mapping[str, str]: return {"MUSE_REPO_ROOT": str(root)} def _invoke(root: pathlib.Path, args: list[str]) -> InvokeResult: saved = os.getcwd() try: os.chdir(root) return runner.invoke(None, args, env=_env(root)) finally: os.chdir(saved) def _log(root: pathlib.Path, *extra: str) -> InvokeResult: return _invoke(root, ["log", *extra]) def _log_json(root: pathlib.Path, *extra: str) -> Mapping[str, object]: result = _log(root, "--json", *extra) assert result.exit_code == 0, f"log --json failed:\n{result.output}" return json.loads(result.output.strip()) @pytest.fixture() def repo(tmp_path: pathlib.Path) -> pathlib.Path: """Initialised code repo with two commits.""" os.chdir(tmp_path) r = _invoke(tmp_path, ["init"]) assert r.exit_code == 0, r.output (tmp_path / "a.py").write_text("a = 1\n") _invoke(tmp_path, ["code", "add", "a.py"]) r = _invoke(tmp_path, ["commit", "-m", "first"]) assert r.exit_code == 0, r.output (tmp_path / "b.py").write_text("b = 2\n") _invoke(tmp_path, ["code", "add", "b.py"]) r = _invoke(tmp_path, ["commit", "-m", "second"]) assert r.exit_code == 0, r.output return tmp_path @pytest.fixture() def single_commit_repo(tmp_path: pathlib.Path) -> pathlib.Path: os.chdir(tmp_path) _invoke(tmp_path, ["init"]) (tmp_path / "x.py").write_text("x = 0\n") _invoke(tmp_path, ["code", "add", "x.py"]) r = _invoke(tmp_path, ["commit", "-m", "init commit"]) assert r.exit_code == 0, r.output return tmp_path # --------------------------------------------------------------------------- # I JSON envelope schema # --------------------------------------------------------------------------- class TestJsonEnvelopeSchema: def test_all_required_envelope_keys_present(self, repo: pathlib.Path) -> None: """Envelope always has all required top-level keys.""" data = _log_json(repo) missing = _REQUIRED_ENVELOPE_KEYS - set(data.keys()) assert not missing, f"Missing envelope keys: {missing}" def test_no_extra_undocumented_keys(self, repo: pathlib.Path) -> None: """Envelope contains no undocumented extra keys.""" data = _log_json(repo) extra = set(data.keys()) - _REQUIRED_ENVELOPE_KEYS assert not extra, f"Undocumented extra keys: {extra}" def test_status_is_ok_on_success(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert data["status"] == "ok" def test_error_is_empty_string_on_success(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert data["error"] == "" def test_exit_code_is_zero_on_success(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert data["exit_code"] == 0 def test_branch_field_matches_current_branch(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert data["branch"] == "main" def test_branch_field_respects_explicit_ref(self, tmp_path: pathlib.Path) -> None: os.chdir(tmp_path) _invoke(tmp_path, ["init"]) (tmp_path / "f.py").write_text("f = 1\n") _invoke(tmp_path, ["code", "add", "f.py"]) _invoke(tmp_path, ["commit", "-m", "base"]) _invoke(tmp_path, ["checkout", "-b", "dev"]) (tmp_path / "g.py").write_text("g = 2\n") _invoke(tmp_path, ["code", "add", "g.py"]) _invoke(tmp_path, ["commit", "-m", "on dev"]) _invoke(tmp_path, ["checkout", "main"]) # Explicitly ask for dev log data = _log_json(tmp_path, "dev") assert data["branch"] == "dev" def test_repo_id_is_nonempty_string(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert isinstance(data["repo_id"], str) assert len(data["repo_id"]) > 0 def test_total_matches_len_commits(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert data["total"] == len(data["commits"]) def test_total_reflects_limit(self, repo: pathlib.Path) -> None: data = _log_json(repo, "-n", "1") assert data["total"] == 1 assert len(data["commits"]) == 1 def test_duration_ms_is_nonnegative_float(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert isinstance(data["duration_ms"], float) assert data["duration_ms"] >= 0.0 def test_commits_is_a_list(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert isinstance(data["commits"], list) def test_truncated_is_bool(self, repo: pathlib.Path) -> None: data = _log_json(repo) assert isinstance(data["truncated"], bool) def test_empty_repo_returns_empty_commits(self, tmp_path: pathlib.Path) -> None: os.chdir(tmp_path) _invoke(tmp_path, ["init"]) data = _log_json(tmp_path) assert data["status"] == "ok" assert data["commits"] == [] assert data["total"] == 0 # --------------------------------------------------------------------------- # II Error payload shape # --------------------------------------------------------------------------- class TestErrorPayloadShape: def test_invalid_since_returns_error_payload(self, repo: pathlib.Path) -> None: result = _log(repo, "--json", "--since", "not-a-date") assert result.exit_code != 0 data = json.loads(result.output.strip()) assert data["status"] == "error" assert isinstance(data["error"], str) and len(data["error"]) > 0 assert isinstance(data["exit_code"], int) and data["exit_code"] != 0 def test_invalid_until_returns_error_payload(self, repo: pathlib.Path) -> None: result = _log(repo, "--json", "--until", "bad") assert result.exit_code != 0 data = json.loads(result.output.strip()) assert data["status"] == "error" assert "error" in data def test_invalid_format_returns_error_payload(self, repo: pathlib.Path) -> None: result = _log(repo, "--format", "xml") assert result.exit_code != 0 def test_error_payload_has_exactly_three_keys(self, repo: pathlib.Path) -> None: """Error payload: exactly {status, error, exit_code}.""" result = _log(repo, "--json", "--since", "not-a-date") data = json.loads(result.output.strip()) assert {"status", "error", "exit_code"}.issubset(data.keys()) def test_no_prose_on_stderr_for_invalid_since_in_json_mode( self, repo: pathlib.Path ) -> None: """When --json is active, errors go to JSON on stdout, not prose on stderr.""" result = _log(repo, "--json", "--since", "not-a-date") # stdout must be valid JSON data = json.loads(result.output.strip()) assert data["status"] == "error" # No human-readable error emoji in stdout assert "❌" not in result.output def test_no_prose_on_stderr_for_invalid_until_in_json_mode( self, repo: pathlib.Path ) -> None: result = _log(repo, "--json", "--until", "not-a-date") data = json.loads(result.output.strip()) assert data["status"] == "error" assert "❌" not in result.output # --------------------------------------------------------------------------- # III Canonical sha256: ref resolution # --------------------------------------------------------------------------- class TestSha256RefResolution: def test_sha256_full_commit_id_as_ref(self, repo: pathlib.Path) -> None: """muse log sha256: should walk from that commit, not fall into pathspec.""" # Get HEAD commit id all_data = _log_json(repo) head_cid = all_data["commits"][0]["commit_id"] assert head_cid.startswith("sha256:") # Walk from that specific commit data = _log_json(repo, head_cid) assert data["status"] == "ok" assert any(c["commit_id"] == head_cid for c in data["commits"]) def test_sha256_prefix_as_ref(self, repo: pathlib.Path) -> None: """sha256: should resolve and not be treated as a pathspec.""" all_data = _log_json(repo) head_cid = all_data["commits"][0]["commit_id"] short_ref = long_id(head_cid[7:15])# sha256: + 8 hex chars data = _log_json(repo, short_ref) assert data["status"] == "ok" assert len(data["commits"]) >= 1 def test_sha256_ref_not_treated_as_pathspec(self, repo: pathlib.Path) -> None: """When sha256: is given as ref, commits list must not be empty.""" all_data = _log_json(repo) head_cid = all_data["commits"][0]["commit_id"] data = _log_json(repo, head_cid) # If it fell into pathspec, no commits would touch a file named sha256:… assert len(data["commits"]) > 0 # --------------------------------------------------------------------------- # IV Data integrity — all commit_ids are sha256:-prefixed # --------------------------------------------------------------------------- class TestDataIntegrity: def test_all_commit_ids_sha256_prefixed(self, repo: pathlib.Path) -> None: data = _log_json(repo) for c in data["commits"]: assert c["commit_id"].startswith("sha256:"), ( f"commit_id not sha256:-prefixed: {c['commit_id']!r}" ) def test_parent_commit_id_sha256_or_null(self, repo: pathlib.Path) -> None: data = _log_json(repo) for c in data["commits"]: pid = c["parent_commit_id"] assert pid is None or pid.startswith("sha256:"), ( f"parent_commit_id not sha256:-prefixed: {pid!r}" ) def test_snapshot_id_sha256_or_null(self, repo: pathlib.Path) -> None: data = _log_json(repo) for c in data["commits"]: sid = c["snapshot_id"] assert sid is None or sid.startswith("sha256:"), ( f"snapshot_id not sha256:-prefixed: {sid!r}" ) def test_committed_at_is_iso8601(self, repo: pathlib.Path) -> None: from datetime import datetime data = _log_json(repo) for c in data["commits"]: ts = c["committed_at"] # Must parse as ISO-8601 datetime.fromisoformat(ts) def test_files_added_is_list(self, repo: pathlib.Path) -> None: data = _log_json(repo) for c in data["commits"]: assert isinstance(c["files_added"], list) def test_files_modified_is_list(self, repo: pathlib.Path) -> None: data = _log_json(repo) for c in data["commits"]: assert isinstance(c["files_modified"], list) def test_files_removed_is_list(self, repo: pathlib.Path) -> None: data = _log_json(repo) for c in data["commits"]: assert isinstance(c["files_removed"], list) def test_agent_id_is_string(self, repo: pathlib.Path) -> None: data = _log_json(repo) for c in data["commits"]: assert isinstance(c["agent_id"], str) def test_model_id_is_string(self, repo: pathlib.Path) -> None: data = _log_json(repo) for c in data["commits"]: assert isinstance(c["model_id"], str) # --------------------------------------------------------------------------- # V Truncation behaviour # --------------------------------------------------------------------------- class TestTruncationBehaviour: def test_explicit_n_does_not_emit_warning_in_text( self, tmp_path: pathlib.Path ) -> None: """When user explicitly passes -n, no truncation warning is printed.""" os.chdir(tmp_path) _invoke(tmp_path, ["init"]) for i in range(5): (tmp_path / f"f{i}.py").write_text(f"x={i}\n") _invoke(tmp_path, ["code", "add", f"f{i}.py"]) _invoke(tmp_path, ["commit", "-m", f"commit {i}"]) result = _log(tmp_path, "--oneline", "-n", "2") assert result.exit_code == 0 assert "truncated" not in result.output.lower(), ( f"Unexpected truncation warning when -n was explicit:\n{result.output}" ) lines = [l for l in result.output.strip().splitlines() if l.strip()] assert len(lines) == 2 def test_truncated_true_in_json_when_limit_hit( self, tmp_path: pathlib.Path ) -> None: """truncated=true in JSON envelope when there are more commits than limit.""" os.chdir(tmp_path) _invoke(tmp_path, ["init"]) for i in range(4): (tmp_path / f"f{i}.py").write_text(f"x={i}\n") _invoke(tmp_path, ["code", "add", f"f{i}.py"]) _invoke(tmp_path, ["commit", "-m", f"commit {i}"]) data = _log_json(tmp_path, "-n", "2") assert data["truncated"] is True assert data["total"] == 2 def test_truncated_false_when_all_commits_returned( self, repo: pathlib.Path ) -> None: data = _log_json(repo) assert data["truncated"] is False def test_default_limit_truncation_warning_fires_in_text( self, tmp_path: pathlib.Path ) -> None: """When default limit is hit (not explicitly set), warning is shown.""" os.chdir(tmp_path) _invoke(tmp_path, ["init"]) for i in range(3): (tmp_path / f"f{i}.py").write_text(f"x={i}\n") _invoke(tmp_path, ["code", "add", f"f{i}.py"]) _invoke(tmp_path, ["commit", "-m", f"commit {i}"]) # Patch the default limit to 2 so the warning fires on a 3-commit repo import muse.cli.commands.log as log_mod orig = log_mod._DEFAULT_LIMIT try: log_mod._DEFAULT_LIMIT = 2 result = _log(tmp_path, "--oneline") # Warning should appear because default was hit, not explicit -n assert "truncated" in result.output.lower() finally: log_mod._DEFAULT_LIMIT = orig # --------------------------------------------------------------------------- # VI TypedDicts # --------------------------------------------------------------------------- class TestTypedDicts: def test_log_json_typed_dict_exists(self) -> None: from muse.cli.commands.log import _LogJson # type: ignore[attr-defined] assert _LogJson is not None def test_log_error_json_typed_dict_exists(self) -> None: from muse.cli.commands.log import _LogErrorJson # type: ignore[attr-defined] assert _LogErrorJson is not None def test_log_json_has_required_annotations(self) -> None: from muse.cli.commands.log import _LogJson # type: ignore[attr-defined] hints = _LogJson.__annotations__ required = {"status", "error", "truncated", "total", "branch", "repo_id", "commits", "duration_ms", "exit_code"} missing = required - set(hints) assert not missing, f"_LogJson missing annotations: {missing}" def test_log_error_json_has_required_annotations(self) -> None: from muse.cli.commands.log import _LogErrorJson # type: ignore[attr-defined] hints = _LogErrorJson.__annotations__ required = {"status", "error", "exit_code"} missing = required - set(hints) assert not missing, f"_LogErrorJson missing annotations: {missing}" # --------------------------------------------------------------------------- # VII Docstring # --------------------------------------------------------------------------- class TestDocstring: def test_module_docstring_documents_status(self) -> None: import muse.cli.commands.log as log_mod assert "status" in (log_mod.__doc__ or "") def test_module_docstring_documents_branch(self) -> None: import muse.cli.commands.log as log_mod assert '"branch"' in (log_mod.__doc__ or "") def test_module_docstring_documents_repo_id(self) -> None: import muse.cli.commands.log as log_mod assert '"repo_id"' in (log_mod.__doc__ or "") def test_module_docstring_documents_total(self) -> None: import muse.cli.commands.log as log_mod assert '"total"' in (log_mod.__doc__ or "") def test_module_docstring_documents_duration_ms(self) -> None: import muse.cli.commands.log as log_mod assert "duration_ms" in (log_mod.__doc__ or "") def test_module_docstring_documents_exit_code(self) -> None: import muse.cli.commands.log as log_mod assert "exit_code" in (log_mod.__doc__ or "") # --------------------------------------------------------------------------- # VIII No prose pollution in --json mode # --------------------------------------------------------------------------- class TestNoProsePollution: def test_success_stdout_is_valid_json(self, repo: pathlib.Path) -> None: result = _log(repo, "--json") assert result.exit_code == 0 data = json.loads(result.output.strip()) # must not raise assert isinstance(data, dict) def test_no_emoji_in_json_success_output(self, repo: pathlib.Path) -> None: result = _log(repo, "--json") assert "✅" not in result.output assert "⚠️" not in result.output def test_no_emoji_in_json_error_output(self, repo: pathlib.Path) -> None: result = _log(repo, "--json", "--since", "bad-date") # Must still be parseable JSON data = json.loads(result.output.strip()) assert "❌" not in result.output assert data["status"] == "error" def test_ansi_in_commit_message_is_json_escaped( self, tmp_path: pathlib.Path ) -> None: """ANSI escape in commit message must be JSON-encoded, not echoed raw.""" os.chdir(tmp_path) _invoke(tmp_path, ["init"]) (tmp_path / "malicious.py").write_text("x = 1\n") _invoke(tmp_path, ["code", "add", "malicious.py"]) _invoke(tmp_path, ["commit", "-m", "safe\x1b[31mmalicious\x1b[0m"]) result = _log(tmp_path, "--json") assert "\x1b" not in result.output data = json.loads(result.output.strip()) assert data["status"] == "ok"