"""Hardening tests for ``muse for-each-ref`` — agent supercharge series. Tests added in this pass ------------------------ - ``duration_ms`` present and valid in JSON output - ``exit_code`` present and zero in JSON output - ``current_branch`` present — which branch HEAD points to - JSON is compact (single line) - ``commit_id`` and ``snapshot_id`` carry sha256: prefix - Data integrity: duration_ms non-negative float, exit_code int, current_branch matches HEAD, count == len(refs) - Performance: 100-branch repo round-trip under 5 s, duration_ms plausible - Security: error output to stderr, no traceback """ from __future__ import annotations from collections.abc import Mapping import datetime import json import pathlib import time import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.ids import hash_commit, hash_snapshot from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot from muse.core.paths import muse_dir, ref_path runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path, head_branch: str = "main") -> pathlib.Path: dot_muse = muse_dir(path) for sub in ("commits", "snapshots", "objects", "refs/heads"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "HEAD").write_text(f"ref: refs/heads/{head_branch}\n") (dot_muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "domain": "code"}) ) return path def _commit( repo: pathlib.Path, msg: str, branch: str = "main", parent: str | None = None, ts: datetime.datetime | None = None, author: str = "gabriel", ) -> str: ts = ts or datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) sid = hash_snapshot({}) write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest={}, created_at=ts)) parent_ids = [parent] if parent else [] cid = hash_commit( parent_ids=parent_ids, snapshot_id=sid, message=msg, committed_at_iso=ts.isoformat(), author=author, ) write_commit(repo, CommitRecord( commit_id=cid, branch=branch, snapshot_id=sid, message=msg, committed_at=ts, author=author, parent_commit_id=parent, parent2_commit_id=None, )) branch_ref = ref_path(repo, branch) branch_ref.parent.mkdir(parents=True, exist_ok=True) branch_ref.write_text(cid) return cid def _fer(repo: pathlib.Path, *args: str) -> InvokeResult: return runner.invoke(None, ["for-each-ref", "--json", *args], env={"MUSE_REPO_ROOT": str(repo)}) def _json(r: InvokeResult) -> Mapping[str, object]: return json.loads(r.output) # --------------------------------------------------------------------------- # duration_ms # --------------------------------------------------------------------------- class TestElapsedSeconds: def test_present_in_full_output(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert "duration_ms" in _json(_fer(tmp_path)) def test_present_with_no_commits(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert "duration_ms" in _json(_fer(tmp_path, "--no-commits")) def test_present_on_empty_repo(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) assert "duration_ms" in _json(_fer(tmp_path)) def test_is_float(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert isinstance(_json(_fer(tmp_path))["duration_ms"], float) def test_non_negative(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert _json(_fer(tmp_path))["duration_ms"] >= 0.0 def test_six_decimal_places(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") v = _json(_fer(tmp_path))["duration_ms"] assert v == round(v, 6) def test_present_with_pattern_filter(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") data = _json(_fer(tmp_path, "--pattern", "refs/heads/main")) assert "duration_ms" in data def test_present_with_count_limit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for b in ["a", "b", "c"]: _commit(tmp_path, f"c-{b}", b) assert "duration_ms" in _json(_fer(tmp_path, "--count", "2")) # --------------------------------------------------------------------------- # exit_code # --------------------------------------------------------------------------- class TestExitCode: def test_present_in_full_output(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert "exit_code" in _json(_fer(tmp_path)) def test_zero_on_success(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert _json(_fer(tmp_path))["exit_code"] == 0 def test_zero_on_empty_repo(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) assert _json(_fer(tmp_path))["exit_code"] == 0 def test_zero_with_no_commits(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert _json(_fer(tmp_path, "--no-commits"))["exit_code"] == 0 def test_is_int_not_bool(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert type(_json(_fer(tmp_path))["exit_code"]) is int # --------------------------------------------------------------------------- # current_branch # --------------------------------------------------------------------------- class TestCurrentBranch: def test_present_in_output(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, head_branch="main") _commit(tmp_path, "c1", "main") assert "current_branch" in _json(_fer(tmp_path)) def test_matches_head_branch(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, head_branch="dev") _commit(tmp_path, "c1", "dev") assert _json(_fer(tmp_path))["current_branch"] == "dev" def test_main_by_default(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, head_branch="main") _commit(tmp_path, "c1", "main") assert _json(_fer(tmp_path))["current_branch"] == "main" def test_present_with_no_commits_flag(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, head_branch="main") _commit(tmp_path, "c1", "main") assert "current_branch" in _json(_fer(tmp_path, "--no-commits")) def test_present_on_empty_repo(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, head_branch="main") data = _json(_fer(tmp_path)) assert "current_branch" in data def test_feature_branch_reflected(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, head_branch="feat/my-thing") _commit(tmp_path, "c1", "feat/my-thing") assert _json(_fer(tmp_path))["current_branch"] == "feat/my-thing" # --------------------------------------------------------------------------- # Compact JSON # --------------------------------------------------------------------------- class TestCompactJson: def test_output_is_single_line(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") r = _fer(tmp_path) assert len(r.output.strip().splitlines()) == 1 def test_no_commits_is_single_line(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") r = _fer(tmp_path, "--no-commits") assert len(r.output.strip().splitlines()) == 1 def test_empty_repo_is_single_line(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _fer(tmp_path) assert len(r.output.strip().splitlines()) == 1 # --------------------------------------------------------------------------- # sha256: prefix on IDs # --------------------------------------------------------------------------- class TestSha256Prefix: def test_commit_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") ref = _json(_fer(tmp_path))["refs"][0] assert ref["commit_id"].startswith("sha256:") def test_commit_id_full_length(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") ref = _json(_fer(tmp_path))["refs"][0] # sha256: (7) + 64 hex chars = 71 assert len(ref["commit_id"]) == 71 def test_snapshot_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") ref = _json(_fer(tmp_path))["refs"][0] assert ref["snapshot_id"].startswith("sha256:") def test_no_commits_commit_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") ref = _json(_fer(tmp_path, "--no-commits"))["refs"][0] assert ref["commit_id"].startswith("sha256:") # --------------------------------------------------------------------------- # Data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_count_equals_len_refs(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for b in ["a", "b", "c", "d"]: _commit(tmp_path, f"c-{b}", b) data = _json(_fer(tmp_path)) assert data["count"] == len(data["refs"]) def test_count_equals_len_refs_after_pattern(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for b in ["feat/x", "feat/y", "main"]: _commit(tmp_path, f"c-{b}", b) data = _json(_fer(tmp_path, "--pattern", "refs/heads/feat/*")) assert data["count"] == len(data["refs"]) def test_count_equals_len_refs_after_count_limit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for b in ["a", "b", "c", "d", "e"]: _commit(tmp_path, f"c-{b}", b) data = _json(_fer(tmp_path, "--count", "3")) assert data["count"] == len(data["refs"]) assert data["count"] == 3 def test_all_refs_have_branch_and_ref_fields(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for b in ["main", "dev", "feat/x"]: _commit(tmp_path, f"c-{b}", b) data = _json(_fer(tmp_path)) for ref in data["refs"]: assert "branch" in ref assert "ref" in ref assert ref["ref"] == f"refs/heads/{ref['branch']}" def test_committed_at_is_iso8601(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") ref = _json(_fer(tmp_path))["refs"][0] # Must parse as a datetime without raising import datetime datetime.datetime.fromisoformat(ref["committed_at"]) # --------------------------------------------------------------------------- # Performance # --------------------------------------------------------------------------- class TestPerformance: def test_duration_ms_plausible(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _commit(tmp_path, "c1") assert _json(_fer(tmp_path))["duration_ms"] < 10.0 def test_100_branch_repo_under_5s(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for i in range(100): _commit(tmp_path, f"c-{i}", f"branch-{i:03d}") t0 = time.monotonic() r = _fer(tmp_path) assert r.exit_code == 0 assert time.monotonic() - t0 < 5.0 assert _json(r)["count"] == 100 def test_no_commits_faster_than_full(self, tmp_path: pathlib.Path) -> None: """--no-commits duration_ms <= full duration_ms (with some slack).""" _init_repo(tmp_path) for i in range(50): _commit(tmp_path, f"c-{i}", f"b-{i:03d}") full = _json(_fer(tmp_path))["duration_ms"] fast = _json(_fer(tmp_path, "--no-commits"))["duration_ms"] # fast path must not be 10x slower than full (loose bound; CI noise) assert fast < full * 10 + 1.0