"""Comprehensive tests for ``muse rerere`` CLI and ``muse/core/rerere.py`` hardening. Audit findings addressed ------------------------ Security - _entry_dir now calls _validate_fingerprint — crafted fingerprints like "../evil" raise ValueError and are rejected by all callers. - load_record enforces _MAX_META_BYTES and _MAX_RESOLUTION_BYTES caps. - apply_cached enforces _MAX_RESOLUTION_BYTES before reading. - list_records and clear_all skip symlinks (is_symlink() guard). - auto_apply validates paths with .relative_to(root) before dest write. - CLI run() validates paths before apply_cached. Dead code removed - domain = read_domain(root) in run() (dead variable — domain unused). - color: bool parameter from _fmt_record (accepted but never used). New capabilities - _RerereApplyJson, _RerereRecordJson, _RerereStatusJson, _RerereForgetJson, _RerereScalarJson TypedDicts for all subcommand JSON outputs. - --fingerprint HEX on ``forget`` subcommand — forget by fingerprint without requiring an active merge state. - --age DAYS on ``gc`` subcommand — configurable GC threshold. - gc_stale(age_days=...) parameter in core. - _validate_fingerprint() public function with 64-char hex validation. - _check_format() helper for format validation. Coverage tiers -------------- Unit _validate_fingerprint, _entry_dir, load_record (size caps), apply_cached (size cap), list_records (symlink guard), clear_all (symlink guard), auto_apply (path traversal guard), gc_stale (age_days param), _fmt_record (no color param), _check_format Integration record_preimage, save_resolution, forget_record, clear_all, list_records, gc_stale, has_resolution Security path traversal in _entry_dir and auto_apply, meta.json size cap, resolution file size cap, symlink guards, ANSI sanitization in text output, JSON output on stdout E2E CLI flags for all subcommands (--json, --dry-run, --yes, --fingerprint, --age), JSON schemas, exit codes, help text Stress 10k list_records scan, concurrent isolated record/load, concurrent gc_stale on isolated repos """ from __future__ import annotations import datetime import json import pathlib import threading import uuid import pytest from muse.core.errors import ExitCode from muse.core.rerere import ( RerereRecord, _MAX_META_BYTES, _MAX_RESOLUTION_BYTES, _entry_dir, _validate_fingerprint, apply_cached, clear_all, conflict_fingerprint, forget_record, gc_stale, list_records, load_record, record_preimage, rr_cache_dir, save_resolution, ) from tests.cli_test_helper import CliRunner, InvokeResult from muse.core._types import Manifest, MsgpackDict runner = CliRunner() cli = None _SHA_A = "a" * 64 _SHA_B = "b" * 64 _SHA_C = "c" * 64 _VALID_FP = conflict_fingerprint(_SHA_A, _SHA_B) # --------------------------------------------------------------------------- # Repo helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: muse = tmp_path / ".muse" for sub in ("commits", "snapshots", "refs/heads", "objects", "rr-cache"): (muse / sub).mkdir(parents=True, exist_ok=True) (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (muse / "repo.json").write_text( json.dumps({"repo_id": str(uuid.uuid4()), "domain": "code"}), encoding="utf-8", ) return tmp_path def _write_preimage(root: pathlib.Path, fp: str, path: str = "src/a.py") -> None: """Write a valid meta.json entry for *fp*.""" entry_dir = rr_cache_dir(root) / fp entry_dir.mkdir(parents=True, exist_ok=True) meta = { "fingerprint": fp, "path": path, "ours_id": _SHA_A, "theirs_id": _SHA_B, "domain": "code", "recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } (entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8") def _write_resolution(root: pathlib.Path, fp: str, res_id: str = _SHA_C) -> None: """Write a resolution file for *fp*.""" entry_dir = rr_cache_dir(root) / fp entry_dir.mkdir(parents=True, exist_ok=True) (entry_dir / "resolution").write_text(res_id, encoding="utf-8") def _invoke(root: pathlib.Path, *args: str) -> InvokeResult: return runner.invoke( cli, ["rerere", *args], env={"MUSE_REPO_ROOT": str(root)}, ) def _parse_json(result: InvokeResult) -> MsgpackDict: start = result.output.index("{") blob = result.output[start:] depth = 0 end = 0 for i, ch in enumerate(blob): if ch == "{": depth += 1 elif ch == "}": depth -= 1 if depth == 0: end = i + 1 break parsed = json.loads(blob[:end]) assert isinstance(parsed, dict) return parsed # --------------------------------------------------------------------------- # Unit — _validate_fingerprint # --------------------------------------------------------------------------- class TestValidateFingerprint: def test_valid_fingerprint_passes(self) -> None: _validate_fingerprint(_VALID_FP) # no exception def test_path_traversal_rejected(self) -> None: with pytest.raises(ValueError): _validate_fingerprint("../evil") def test_too_short_rejected(self) -> None: with pytest.raises(ValueError): _validate_fingerprint("abc123") def test_too_long_rejected(self) -> None: with pytest.raises(ValueError): _validate_fingerprint("a" * 65) def test_uppercase_rejected(self) -> None: with pytest.raises(ValueError): _validate_fingerprint("A" * 64) def test_slash_in_fingerprint_rejected(self) -> None: with pytest.raises(ValueError): _validate_fingerprint("a" * 32 + "/" + "a" * 31) def test_null_byte_rejected(self) -> None: with pytest.raises(ValueError): _validate_fingerprint("a" * 63 + "\x00") def test_empty_string_rejected(self) -> None: with pytest.raises(ValueError): _validate_fingerprint("") def test_exactly_64_hex_passes(self) -> None: fp = "0123456789abcdef" * 4 _validate_fingerprint(fp) # no exception # --------------------------------------------------------------------------- # Unit — _entry_dir path traversal guard # --------------------------------------------------------------------------- class TestEntryDir: def test_valid_fp_returns_path(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) entry = _entry_dir(root, _VALID_FP) assert entry.parent == rr_cache_dir(root) def test_traversal_fp_raises(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) with pytest.raises(ValueError): _entry_dir(root, "../evil") def test_entry_is_inside_rr_cache(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) entry = _entry_dir(root, _VALID_FP) assert str(entry).startswith(str(rr_cache_dir(root))) # --------------------------------------------------------------------------- # Unit — load_record size caps # --------------------------------------------------------------------------- class TestLoadRecordSizeCaps: def test_oversized_meta_returns_none(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) meta_path = rr_cache_dir(root) / _VALID_FP / "meta.json" # Overwrite with a file that exceeds the cap import muse.core.rerere as rmod original = rmod._MAX_META_BYTES try: rmod._MAX_META_BYTES = 5 # tiny cap result = load_record(root, _VALID_FP) finally: rmod._MAX_META_BYTES = original assert result is None def test_oversized_resolution_ignored(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) res_path = rr_cache_dir(root) / _VALID_FP / "resolution" import muse.core.rerere as rmod original = rmod._MAX_RESOLUTION_BYTES try: rmod._MAX_RESOLUTION_BYTES = 5 # tiny cap — can't fit a 64-char hex res_path.write_text(_SHA_C, encoding="utf-8") record = load_record(root, _VALID_FP) finally: rmod._MAX_RESOLUTION_BYTES = original # Record still loaded (preimage exists) but resolution_id is None assert record is not None assert record.resolution_id is None def test_invalid_fingerprint_returns_none(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = load_record(root, "not-a-fingerprint") assert result is None def test_valid_entry_loads_correctly(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP, path="src/x.py") _write_resolution(root, _VALID_FP, res_id=_SHA_C) record = load_record(root, _VALID_FP) assert record is not None assert record.fingerprint == _VALID_FP assert record.path == "src/x.py" assert record.resolution_id == _SHA_C # --------------------------------------------------------------------------- # Unit — apply_cached size cap # --------------------------------------------------------------------------- class TestApplyCachedSizeCap: def test_oversized_resolution_skipped(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) res_path = rr_cache_dir(root) / _VALID_FP / "resolution" import muse.core.rerere as rmod original = rmod._MAX_RESOLUTION_BYTES try: rmod._MAX_RESOLUTION_BYTES = 5 res_path.write_text(_SHA_C, encoding="utf-8") dest = tmp_path / "output.py" ok = apply_cached(root, _VALID_FP, dest) finally: rmod._MAX_RESOLUTION_BYTES = original assert not ok assert not dest.exists() def test_invalid_fingerprint_returns_false(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) ok = apply_cached(root, "../traversal", tmp_path / "out.py") assert not ok # --------------------------------------------------------------------------- # Unit — list_records symlink guard # --------------------------------------------------------------------------- class TestListRecordsSymlinkGuard: def test_symlink_skipped(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) cache = rr_cache_dir(root) # Create a symlink pointing to some external dir target = tmp_path / "external" target.mkdir() link = cache / ("f" * 64) try: link.symlink_to(target) records = list_records(root) assert all(r.fingerprint == _VALID_FP for r in records) assert len(records) == 1 except NotImplementedError: pytest.skip("symlinks not supported on this platform") def test_regular_dirs_included(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_B, _SHA_C) _write_preimage(root, _VALID_FP) _write_preimage(root, fp2) records = list_records(root) fps = {r.fingerprint for r in records} assert _VALID_FP in fps assert fp2 in fps # --------------------------------------------------------------------------- # Unit — clear_all symlink guard # --------------------------------------------------------------------------- class TestClearAllSymlinkGuard: def test_symlink_not_cleared(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) external = tmp_path / "external_dir" external.mkdir() sentinel = external / "sentinel.txt" sentinel.write_text("keep me", encoding="utf-8") cache = rr_cache_dir(root) link = cache / ("e" * 64) try: link.symlink_to(external) clear_all(root) # The symlink target is untouched assert sentinel.exists() except NotImplementedError: pytest.skip("symlinks not supported on this platform") def test_regular_entries_cleared(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) removed = clear_all(root) assert removed == 1 assert not (rr_cache_dir(root) / _VALID_FP).exists() # --------------------------------------------------------------------------- # Unit — auto_apply path traversal guard # --------------------------------------------------------------------------- class TestAutoApplyPathTraversal: def test_traversal_path_skipped(self, tmp_path: pathlib.Path) -> None: from muse.core.rerere import auto_apply from unittest.mock import MagicMock root = _make_repo(tmp_path) plugin = MagicMock() del plugin.conflict_fingerprint # not a RererePlugin resolved, remaining = auto_apply( root, conflict_paths=["../../etc/passwd"], ours_manifest={"../../etc/passwd": _SHA_A}, theirs_manifest={"../../etc/passwd": _SHA_B}, domain="code", plugin=plugin, ) assert "../../etc/passwd" in remaining assert "../../etc/passwd" not in resolved def test_valid_path_processed(self, tmp_path: pathlib.Path) -> None: from muse.core.rerere import auto_apply from unittest.mock import MagicMock root = _make_repo(tmp_path) plugin = MagicMock() del plugin.conflict_fingerprint # not a RererePlugin # No resolution cached — should be added to remaining and preimage recorded resolved, remaining = auto_apply( root, conflict_paths=["src/module.py"], ours_manifest={"src/module.py": _SHA_A}, theirs_manifest={"src/module.py": _SHA_B}, domain="code", plugin=plugin, ) assert "src/module.py" in remaining # --------------------------------------------------------------------------- # Unit — gc_stale age_days parameter # --------------------------------------------------------------------------- class TestGcStaleAgeDays: def test_default_60_days_keeps_recent(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) # Entry is just created — within 60 days — should NOT be removed removed = gc_stale(root, age_days=60) assert removed == 0 def test_age_1_day_removes_entry_written_two_days_ago( self, tmp_path: pathlib.Path ) -> None: root = _make_repo(tmp_path) cache_dir = rr_cache_dir(root) entry_dir = cache_dir / _VALID_FP entry_dir.mkdir(parents=True, exist_ok=True) old_time = ( datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=2) ).isoformat() meta = { "fingerprint": _VALID_FP, "path": "src/a.py", "ours_id": _SHA_A, "theirs_id": _SHA_B, "domain": "code", "recorded_at": old_time, } (entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8") removed = gc_stale(root, age_days=1) assert removed == 1 def test_resolved_entry_never_removed(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) cache_dir = rr_cache_dir(root) entry_dir = cache_dir / _VALID_FP entry_dir.mkdir(parents=True, exist_ok=True) old_time = ( datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=100) ).isoformat() meta = { "fingerprint": _VALID_FP, "path": "src/a.py", "ours_id": _SHA_A, "theirs_id": _SHA_B, "domain": "code", "recorded_at": old_time, } (entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8") (entry_dir / "resolution").write_text(_SHA_C, encoding="utf-8") removed = gc_stale(root, age_days=1) assert removed == 0 # --------------------------------------------------------------------------- # Unit — _fmt_record (no color parameter) # --------------------------------------------------------------------------- class TestFmtRecord: def _rec(self) -> RerereRecord: return RerereRecord( fingerprint=_VALID_FP, path="src/module.py", ours_id=_SHA_A, theirs_id=_SHA_B, domain="code", recorded_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), ) def test_no_color_parameter(self) -> None: from muse.cli.commands.rerere import _fmt_record rec = self._rec() result = _fmt_record(rec) # must not require color argument assert isinstance(result, str) def test_sanitizes_path(self) -> None: from muse.cli.commands.rerere import _fmt_record rec = RerereRecord( fingerprint=_VALID_FP, path="src/\x1b[31mevil\x1b[0m.py", ours_id=_SHA_A, theirs_id=_SHA_B, domain="code", recorded_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), ) result = _fmt_record(rec) assert "\x1b" not in result def test_shows_resolved_status(self) -> None: from muse.cli.commands.rerere import _fmt_record rec = RerereRecord( fingerprint=_VALID_FP, path="src/a.py", ours_id=_SHA_A, theirs_id=_SHA_B, domain="code", recorded_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), resolution_id=_SHA_C, ) result = _fmt_record(rec) assert "resolved" in result def test_shows_preimage_only_status(self) -> None: from muse.cli.commands.rerere import _fmt_record rec = self._rec() result = _fmt_record(rec) assert "preimage" in result # --------------------------------------------------------------------------- # Unit — _check_format # --------------------------------------------------------------------------- class TestCheckFormat: def test_valid_text_passes(self) -> None: from muse.cli.commands.rerere import _check_format _check_format("text") def test_valid_json_passes(self) -> None: from muse.cli.commands.rerere import _check_format _check_format("json") def test_invalid_format_raises(self) -> None: from muse.cli.commands.rerere import _check_format with pytest.raises(SystemExit) as exc_info: _check_format("xml") assert exc_info.value.code == ExitCode.USER_ERROR.value # --------------------------------------------------------------------------- # Integration — status and gc subcommands # --------------------------------------------------------------------------- class TestStatusIntegration: def test_empty_cache_shows_no_records(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "status") assert result.exit_code == 0 assert "No rerere records" in result.output def test_status_shows_entry(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP, path="src/x.py") result = _invoke(root, "status") assert result.exit_code == 0 assert _VALID_FP[:12] in result.output def test_status_json_schema(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP, path="src/x.py") result = _invoke(root, "status", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "total" in raw assert "records" in raw records = raw["records"] assert isinstance(records, list) assert len(records) == 1 rec = records[0] assert isinstance(rec, dict) for field in ("fingerprint", "path", "domain", "has_resolution", "resolution_id", "recorded_at", "matches_current_conflict"): assert field in rec, f"Missing field: {field}" def test_status_json_has_resolution_false(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "status", "--json") raw = _parse_json(result) records = raw["records"] assert isinstance(records, list) rec = records[0] assert isinstance(rec, dict) assert rec["has_resolution"] is False assert rec["resolution_id"] is None def test_status_json_has_resolution_true(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) _write_resolution(root, _VALID_FP, res_id=_SHA_C) result = _invoke(root, "status", "--json") raw = _parse_json(result) records = raw["records"] assert isinstance(records, list) rec = records[0] assert isinstance(rec, dict) assert rec["has_resolution"] is True assert rec["resolution_id"] == _SHA_C # =========================================================================== # TestRerereStatusExtended — 18 tests # =========================================================================== class TestRerereStatusExtended: def test_exit_code_zero_empty(self, tmp_path: pathlib.Path) -> None: """Empty cache exits 0.""" root = _make_repo(tmp_path) result = _invoke(root, "status") assert result.exit_code == 0 def test_exit_code_zero_with_entries(self, tmp_path: pathlib.Path) -> None: """Non-empty cache exits 0.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "status") assert result.exit_code == 0 def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: """No .muse directory → exit 2.""" result = _invoke(tmp_path, "status") assert result.exit_code == 2 def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "status", "--format", "xml") assert result.exit_code == 1 def test_j_alias(self, tmp_path: pathlib.Path) -> None: """-j must produce identical output to --json.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) r1 = _invoke(root, "status", "--json") r2 = _invoke(root, "status", "-j") assert r1.exit_code == 0 and r2.exit_code == 0 assert _parse_json(r1) == _parse_json(r2) def test_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None: """JSON output must be compact — no indent=2.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "status", "--json") assert result.exit_code == 0 assert "\n" not in result.output.strip() def test_json_total_field(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) raw = _parse_json(_invoke(root, "status", "--json")) assert raw["total"] == 1 def test_json_total_zero_when_empty(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) raw = _parse_json(_invoke(root, "status", "--json")) assert raw["total"] == 0 assert raw["records"] == [] def test_json_all_fields_present(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP, path="src/x.py") raw = _parse_json(_invoke(root, "status", "--json")) rec = raw["records"][0] for field in ("fingerprint", "path", "domain", "has_resolution", "resolution_id", "recorded_at", "matches_current_conflict"): assert field in rec def test_json_multiple_entries_total(self, tmp_path: pathlib.Path) -> None: """total reflects number of distinct entries.""" root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_B, _SHA_C) _write_preimage(root, _VALID_FP, path="src/a.py") _write_preimage(root, fp2, path="src/b.py") raw = _parse_json(_invoke(root, "status", "--json")) assert raw["total"] == 2 assert len(raw["records"]) == 2 def test_json_resolution_id_null_without_resolution(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) raw = _parse_json(_invoke(root, "status", "--json")) assert raw["records"][0]["resolution_id"] is None def test_json_resolution_id_set_with_resolution(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) _write_resolution(root, _VALID_FP, res_id=_SHA_C) raw = _parse_json(_invoke(root, "status", "--json")) assert raw["records"][0]["resolution_id"] == _SHA_C def test_json_has_resolution_true_and_false(self, tmp_path: pathlib.Path) -> None: """Mix of resolved and preimage-only entries reported correctly.""" root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_B, _SHA_C) _write_preimage(root, _VALID_FP, path="src/a.py") _write_preimage(root, fp2, path="src/b.py") _write_resolution(root, _VALID_FP) raw = _parse_json(_invoke(root, "status", "--json")) by_fp = {r["fingerprint"]: r for r in raw["records"]} assert by_fp[_VALID_FP]["has_resolution"] is True assert by_fp[fp2]["has_resolution"] is False def test_json_matches_current_conflict_false_without_merge( self, tmp_path: pathlib.Path ) -> None: """No merge in progress → matches_current_conflict always False.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) raw = _parse_json(_invoke(root, "status", "--json")) assert raw["records"][0]["matches_current_conflict"] is False def test_text_format_explicit(self, tmp_path: pathlib.Path) -> None: """--format text is identical to default.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) r_default = _invoke(root, "status") r_text = _invoke(root, "status", "--format", "text") assert r_default.output == r_text.output def test_text_shows_header_when_entries_exist(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "status") assert "fingerprint" in result.output assert "status" in result.output def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "status", "--help") assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "status", "--help") assert "Exit codes" in result.output # =========================================================================== # TestRerereStatusSecurity — 6 tests # =========================================================================== class TestRerereStatusSecurity: def test_ansi_in_path_stripped_json(self, tmp_path: pathlib.Path) -> None: """ANSI in a path must be stripped in JSON output.""" root = _make_repo(tmp_path) evil_path = "src/\x1b[31mevil\x1b[0m.py" _write_preimage(root, _VALID_FP, path=evil_path) result = _invoke(root, "status", "--json") assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_in_path_stripped_text(self, tmp_path: pathlib.Path) -> None: """ANSI in a path must be stripped in text output.""" root = _make_repo(tmp_path) evil_path = "src/\x1b[31mevil\x1b[0m.py" _write_preimage(root, _VALID_FP, path=evil_path) result = _invoke(root, "status") assert result.exit_code == 0 assert "\x1b" not in result.output def test_control_char_in_path_stripped_json(self, tmp_path: pathlib.Path) -> None: """Control characters in path stripped from JSON output.""" root = _make_repo(tmp_path) evil_path = "src/evil\x07.py" _write_preimage(root, _VALID_FP, path=evil_path) result = _invoke(root, "status", "--json") assert result.exit_code == 0 assert "\x07" not in result.output def test_no_json_outside_repo(self, tmp_path: pathlib.Path) -> None: """Running outside a repo must not emit JSON — exits 2.""" result = _invoke(tmp_path, "status", "--json") assert result.exit_code == 2 assert not result.output.strip().startswith("{") def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path) -> None: """Running outside a repo must not emit a traceback.""" result = _invoke(tmp_path, "status") assert result.exit_code == 2 assert "Traceback" not in result.output def test_resolution_id_ansi_stripped_json(self, tmp_path: pathlib.Path) -> None: """ANSI in a resolution_id must be stripped in JSON output.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP, path="src/a.py") evil_res_id = _SHA_C[:32] + "\x1b[31m" + _SHA_C[32:] _write_resolution(root, _VALID_FP, res_id=evil_res_id) result = _invoke(root, "status", "--json") assert result.exit_code == 0 assert "\x1b" not in result.output # =========================================================================== # TestRerereStatusStress — 3 tests # =========================================================================== class TestRerereStatusStress: def test_100_entries_text(self, tmp_path: pathlib.Path) -> None: """100 cache entries are all listed in text mode.""" import hashlib as _hl root = _make_repo(tmp_path) fps = [] for i in range(100): fp = conflict_fingerprint( _hl.sha256(f"a{i}".encode()).hexdigest(), _hl.sha256(f"b{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/f{i}.py") fps.append(fp) result = _invoke(root, "status") assert result.exit_code == 0 for fp in fps: assert fp[:12] in result.output def test_100_entries_json_total(self, tmp_path: pathlib.Path) -> None: """100 cache entries produce total=100 in JSON mode.""" import hashlib as _hl root = _make_repo(tmp_path) for i in range(100): fp = conflict_fingerprint( _hl.sha256(f"c{i}".encode()).hexdigest(), _hl.sha256(f"d{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/g{i}.py") raw = _parse_json(_invoke(root, "status", "--json")) assert raw["total"] == 100 assert len(raw["records"]) == 100 def test_mixed_resolved_and_preimage_counts(self, tmp_path: pathlib.Path) -> None: """50 resolved + 50 preimage-only: has_resolution counts correct.""" import hashlib as _hl root = _make_repo(tmp_path) for i in range(100): fp = conflict_fingerprint( _hl.sha256(f"e{i}".encode()).hexdigest(), _hl.sha256(f"f{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/h{i}.py") if i < 50: _write_resolution(root, fp) raw = _parse_json(_invoke(root, "status", "--json")) resolved = sum(1 for r in raw["records"] if r["has_resolution"]) preimage_only = sum(1 for r in raw["records"] if not r["has_resolution"]) assert resolved == 50 assert preimage_only == 50 # --------------------------------------------------------------------------- # Integration — forget with --fingerprint # --------------------------------------------------------------------------- class TestForgetFingerprint: def test_forget_by_fingerprint_removes_entry( self, tmp_path: pathlib.Path ) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 0 assert not (rr_cache_dir(root) / _VALID_FP).exists() def test_forget_by_fingerprint_json(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "forgotten" in raw assert "not_found" in raw def test_forget_by_fingerprint_not_found(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json") raw = _parse_json(result) not_found = raw["not_found"] assert isinstance(not_found, list) assert _VALID_FP in not_found def test_forget_by_fingerprint_no_merge_required( self, tmp_path: pathlib.Path ) -> None: """Fingerprint mode does not need an active merge state.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) # No MERGE_STATE.json exists result = _invoke(root, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 0 def test_forget_invalid_fingerprint_rejected( self, tmp_path: pathlib.Path ) -> None: root = _make_repo(tmp_path) result = _invoke(root, "forget", "--fingerprint", "../evil") # Should exit 0 (not_found), not crash assert result.exit_code == 0 assert "Traceback" not in result.output def test_forget_multiple_fingerprints(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_B, _SHA_C) _write_preimage(root, _VALID_FP) _write_preimage(root, fp2) result = _invoke( root, "forget", "--fingerprint", _VALID_FP, "--fingerprint", fp2, "--json", ) assert result.exit_code == 0 raw = _parse_json(result) forgotten = raw["forgotten"] assert isinstance(forgotten, list) assert len(forgotten) == 2 def test_forget_no_args_exits_user_error( self, tmp_path: pathlib.Path ) -> None: root = _make_repo(tmp_path) result = _invoke(root, "forget") assert result.exit_code == ExitCode.USER_ERROR.value # =========================================================================== # TestRerereForgetExtended — 18 tests # =========================================================================== class TestRerereForgetExtended: def test_fingerprint_mode_removes_entry(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 0 assert not (rr_cache_dir(root) / _VALID_FP).exists() def test_fingerprint_mode_forgotten_in_json(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) raw = _parse_json(_invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")) assert _VALID_FP in raw["forgotten"] assert raw["not_found"] == [] def test_fingerprint_mode_not_found_in_json(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) raw = _parse_json(_invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")) assert _VALID_FP in raw["not_found"] assert raw["forgotten"] == [] def test_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None: """JSON output must be compact — no indent=2.""" root = _make_repo(tmp_path) result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json") assert result.exit_code == 0 assert "\n" not in result.output.strip() def test_j_alias(self, tmp_path: pathlib.Path) -> None: """-j must produce identical output to --json.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) r1 = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json") _write_preimage(root, _VALID_FP) # restore for second call r2 = _invoke(root, "forget", "--fingerprint", _VALID_FP, "-j") assert r1.exit_code == 0 and r2.exit_code == 0 assert _parse_json(r1).keys() == _parse_json(r2).keys() def test_multiple_fingerprints_all_forgotten(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_B, _SHA_C) _write_preimage(root, _VALID_FP, path="src/a.py") _write_preimage(root, fp2, path="src/b.py") raw = _parse_json(_invoke( root, "forget", "--fingerprint", _VALID_FP, "--fingerprint", fp2, "--json", )) assert len(raw["forgotten"]) == 2 assert raw["not_found"] == [] def test_mixed_found_and_not_found(self, tmp_path: pathlib.Path) -> None: """One found + one absent → both lists populated.""" root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_B, _SHA_C) _write_preimage(root, _VALID_FP) raw = _parse_json(_invoke( root, "forget", "--fingerprint", _VALID_FP, "--fingerprint", fp2, "--json", )) assert _VALID_FP in raw["forgotten"] assert fp2 in raw["not_found"] def test_text_output_shows_forgot(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 0 assert "forgot" in result.output def test_text_output_shows_no_record_found(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 0 assert "no record found" in result.output def test_no_args_exits_1(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "forget") assert result.exit_code == 1 def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--format", "xml") assert result.exit_code == 1 def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: result = _invoke(tmp_path, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 2 def test_path_mode_no_merge_exits_1(self, tmp_path: pathlib.Path) -> None: """Path mode without an active merge must exit 1.""" root = _make_repo(tmp_path) result = _invoke(root, "forget", "src/a.py") assert result.exit_code == 1 def test_fingerprint_mode_no_merge_required(self, tmp_path: pathlib.Path) -> None: """Fingerprint mode works without any MERGE_STATE.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 0 def test_json_schema_fields_present(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) raw = _parse_json(_invoke(root, "forget", "--fingerprint", _VALID_FP, "--json")) assert "forgotten" in raw assert "not_found" in raw assert isinstance(raw["forgotten"], list) assert isinstance(raw["not_found"], list) def test_invalid_fingerprint_lands_in_not_found(self, tmp_path: pathlib.Path) -> None: """A traversal-like fingerprint is rejected by forget_record and lands in not_found.""" root = _make_repo(tmp_path) raw = _parse_json(_invoke(root, "forget", "--fingerprint", "../evil", "--json")) assert "../evil" in raw["not_found"] assert raw["forgotten"] == [] def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "forget", "--help") assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "forget", "--help") assert "Exit codes" in result.output # =========================================================================== # TestRerereForgetSecurity — 6 tests # =========================================================================== class TestRerereForgetSecurity: def test_ansi_in_fingerprint_stripped_json(self, tmp_path: pathlib.Path) -> None: """ANSI in a forgotten fingerprint must be stripped in JSON output.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json") assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_in_not_found_stripped_json(self, tmp_path: pathlib.Path) -> None: """ANSI injected into a not_found item must be stripped.""" root = _make_repo(tmp_path) evil_fp = "\x1b[31m" + "a" * 64 + "\x1b[0m" result = _invoke(root, "forget", "--fingerprint", evil_fp, "--json") assert result.exit_code == 0 assert "\x1b" not in result.output def test_control_char_in_item_stripped_json(self, tmp_path: pathlib.Path) -> None: """Control chars other than ANSI stripped from JSON.""" root = _make_repo(tmp_path) evil_fp = "a" * 32 + "\x07" + "b" * 31 result = _invoke(root, "forget", "--fingerprint", evil_fp, "--json") assert result.exit_code == 0 assert "\x07" not in result.output def test_ansi_in_fingerprint_stripped_text(self, tmp_path: pathlib.Path) -> None: """ANSI stripped from text output.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 0 assert "\x1b" not in result.output def test_no_json_outside_repo(self, tmp_path: pathlib.Path) -> None: """Outside a repo must not emit JSON.""" result = _invoke(tmp_path, "forget", "--fingerprint", _VALID_FP, "--json") assert result.exit_code == 2 assert not result.output.strip().startswith("{") def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path) -> None: result = _invoke(tmp_path, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 2 assert "Traceback" not in result.output # =========================================================================== # TestRerereForgetStress — 3 tests # =========================================================================== class TestRerereForgetStress: def test_50_fingerprints_all_forgotten(self, tmp_path: pathlib.Path) -> None: """50 fingerprints forgotten in one call.""" import hashlib as _hl root = _make_repo(tmp_path) fps = [] for i in range(50): fp = conflict_fingerprint( _hl.sha256(f"x{i}".encode()).hexdigest(), _hl.sha256(f"y{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/f{i}.py") fps.append(fp) args = ["forget"] for fp in fps: args += ["--fingerprint", fp] args.append("--json") raw = _parse_json(_invoke(root, *args)) assert len(raw["forgotten"]) == 50 assert raw["not_found"] == [] def test_50_fingerprints_all_not_found(self, tmp_path: pathlib.Path) -> None: """50 absent fingerprints all land in not_found.""" import hashlib as _hl root = _make_repo(tmp_path) fps = [] for i in range(50): fp = conflict_fingerprint( _hl.sha256(f"p{i}".encode()).hexdigest(), _hl.sha256(f"q{i}".encode()).hexdigest(), ) fps.append(fp) args = ["forget"] for fp in fps: args += ["--fingerprint", fp] args.append("--json") raw = _parse_json(_invoke(root, *args)) assert len(raw["not_found"]) == 50 assert raw["forgotten"] == [] def test_mixed_25_found_25_not_found(self, tmp_path: pathlib.Path) -> None: """25 found + 25 absent → counts match.""" import hashlib as _hl root = _make_repo(tmp_path) fps_present = [] fps_absent = [] for i in range(25): fp = conflict_fingerprint( _hl.sha256(f"m{i}".encode()).hexdigest(), _hl.sha256(f"n{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/g{i}.py") fps_present.append(fp) for i in range(25): fp = conflict_fingerprint( _hl.sha256(f"r{i}".encode()).hexdigest(), _hl.sha256(f"s{i}".encode()).hexdigest(), ) fps_absent.append(fp) args = ["forget"] for fp in fps_present + fps_absent: args += ["--fingerprint", fp] args.append("--json") raw = _parse_json(_invoke(root, *args)) assert len(raw["forgotten"]) == 25 assert len(raw["not_found"]) == 25 # --------------------------------------------------------------------------- # Integration — gc with --age # --------------------------------------------------------------------------- class TestGcAge: def test_default_age_flag_present_in_help(self) -> None: result = runner.invoke(cli, ["rerere", "gc", "--help"]) assert "age" in result.output.lower() def test_gc_age_custom(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) cache_dir = rr_cache_dir(root) entry_dir = cache_dir / _VALID_FP entry_dir.mkdir(parents=True, exist_ok=True) old_time = ( datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=2) ).isoformat() meta = { "fingerprint": _VALID_FP, "path": "src/a.py", "ours_id": _SHA_A, "theirs_id": _SHA_B, "domain": "code", "recorded_at": old_time, } (entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8") result = _invoke(root, "gc", "--age", "1", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert raw["removed"] == 1 def test_gc_json_schema(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "removed" in raw def test_gc_text_mentions_threshold(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--age", "30") assert result.exit_code == 0 assert "30" in result.output def _write_stale_preimage( root: pathlib.Path, fp: str, age_days: int = 90, path: str = "src/a.py", ) -> None: """Write a preimage entry recorded *age_days* ago.""" entry_dir = rr_cache_dir(root) / fp entry_dir.mkdir(parents=True, exist_ok=True) old_time = ( datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=age_days) ).isoformat() meta = { "fingerprint": fp, "path": path, "ours_id": _SHA_A, "theirs_id": _SHA_B, "domain": "code", "recorded_at": old_time, } (entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8") # =========================================================================== # TestRerereGcExtended — 18 tests # =========================================================================== class TestRerereGcExtended: def test_exit_code_zero_nothing_removed(self, tmp_path: pathlib.Path) -> None: """Empty cache exits 0 with removed=0.""" root = _make_repo(tmp_path) result = _invoke(root, "gc", "--json") assert result.exit_code == 0 assert _parse_json(result)["removed"] == 0 def test_exit_code_zero_with_removal(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_stale_preimage(root, _VALID_FP, age_days=90) result = _invoke(root, "gc", "--age", "1", "--json") assert result.exit_code == 0 assert _parse_json(result)["removed"] == 1 def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: result = _invoke(tmp_path, "gc") assert result.exit_code == 2 def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--format", "xml") assert result.exit_code == 1 def test_age_zero_exits_1(self, tmp_path: pathlib.Path) -> None: """--age 0 is below the valid range — must exit 1, not crash.""" root = _make_repo(tmp_path) result = _invoke(root, "gc", "--age", "0") assert result.exit_code == 1 assert "Traceback" not in result.output def test_age_negative_exits_1(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--age", "-1") assert result.exit_code == 1 assert "Traceback" not in result.output def test_j_alias(self, tmp_path: pathlib.Path) -> None: """-j must produce identical output to --json.""" root = _make_repo(tmp_path) r1 = _invoke(root, "gc", "--json") r2 = _invoke(root, "gc", "-j") assert r1.exit_code == 0 and r2.exit_code == 0 assert _parse_json(r1) == _parse_json(r2) def test_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--json") assert result.exit_code == 0 assert "\n" not in result.output.strip() def test_stale_preimage_removed(self, tmp_path: pathlib.Path) -> None: """Preimage-only entry older than threshold is removed.""" root = _make_repo(tmp_path) _write_stale_preimage(root, _VALID_FP, age_days=90) raw = _parse_json(_invoke(root, "gc", "--age", "1", "--json")) assert raw["removed"] == 1 assert not (rr_cache_dir(root) / _VALID_FP).exists() def test_fresh_preimage_kept(self, tmp_path: pathlib.Path) -> None: """Preimage-only entry newer than threshold is kept.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) # recorded just now raw = _parse_json(_invoke(root, "gc", "--age", "60", "--json")) assert raw["removed"] == 0 assert (rr_cache_dir(root) / _VALID_FP).exists() def test_resolved_entry_never_removed(self, tmp_path: pathlib.Path) -> None: """An entry with a resolution is kept regardless of age.""" root = _make_repo(tmp_path) _write_stale_preimage(root, _VALID_FP, age_days=365) _write_resolution(root, _VALID_FP) raw = _parse_json(_invoke(root, "gc", "--age", "1", "--json")) assert raw["removed"] == 0 assert (rr_cache_dir(root) / _VALID_FP).exists() def test_stale_removed_fresh_kept(self, tmp_path: pathlib.Path) -> None: """Only entries older than threshold are removed.""" fp2 = conflict_fingerprint(_SHA_B, _SHA_C) root = _make_repo(tmp_path) _write_stale_preimage(root, _VALID_FP, age_days=90) _write_preimage(root, fp2) # fresh raw = _parse_json(_invoke(root, "gc", "--age", "30", "--json")) assert raw["removed"] == 1 assert not (rr_cache_dir(root) / _VALID_FP).exists() assert (rr_cache_dir(root) / fp2).exists() def test_default_age_is_60(self, tmp_path: pathlib.Path) -> None: """Default --age=60: entry 61 days old is removed, 59 days old is kept.""" fp2 = conflict_fingerprint(_SHA_B, _SHA_C) root = _make_repo(tmp_path) _write_stale_preimage(root, _VALID_FP, age_days=61) _write_stale_preimage(root, fp2, age_days=59) raw = _parse_json(_invoke(root, "gc", "--json")) assert raw["removed"] == 1 def test_text_output_removed_count(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_stale_preimage(root, _VALID_FP, age_days=90) result = _invoke(root, "gc", "--age", "1") assert result.exit_code == 0 assert "1" in result.output def test_text_output_nothing_to_remove(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc") assert result.exit_code == 0 assert "nothing" in result.output.lower() or "0" in result.output def test_format_text_explicit(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) r_default = _invoke(root, "gc") r_text = _invoke(root, "gc", "--format", "text") assert r_default.output == r_text.output def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "gc", "--help") assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "gc", "--help") assert "Exit codes" in result.output # =========================================================================== # TestRerereGcSecurity — 6 tests # =========================================================================== class TestRerereGcSecurity: def test_no_json_outside_repo(self, tmp_path: pathlib.Path) -> None: result = _invoke(tmp_path, "gc", "--json") assert result.exit_code == 2 assert not result.output.strip().startswith("{") def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path) -> None: result = _invoke(tmp_path, "gc") assert result.exit_code == 2 assert "Traceback" not in result.output def test_no_traceback_invalid_format(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--format", "toml") assert "Traceback" not in result.output def test_no_traceback_age_zero(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--age", "0") assert "Traceback" not in result.output def test_age_max_boundary_accepted(self, tmp_path: pathlib.Path) -> None: """age=36500 (100 years) is within the valid range.""" root = _make_repo(tmp_path) result = _invoke(root, "gc", "--age", "36500", "--json") assert result.exit_code == 0 def test_age_exceeds_max_exits_1(self, tmp_path: pathlib.Path) -> None: """age=36501 exceeds valid range — must exit 1 cleanly.""" root = _make_repo(tmp_path) result = _invoke(root, "gc", "--age", "36501") assert result.exit_code == 1 assert "Traceback" not in result.output # =========================================================================== # TestRerereGcStress — 3 tests # =========================================================================== class TestRerereGcStress: def test_100_stale_entries_removed(self, tmp_path: pathlib.Path) -> None: """100 stale preimage-only entries all removed in one gc call.""" import hashlib as _hl root = _make_repo(tmp_path) for i in range(100): fp = conflict_fingerprint( _hl.sha256(f"gc-a{i}".encode()).hexdigest(), _hl.sha256(f"gc-b{i}".encode()).hexdigest(), ) _write_stale_preimage(root, fp, age_days=90, path=f"src/f{i}.py") raw = _parse_json(_invoke(root, "gc", "--age", "1", "--json")) assert raw["removed"] == 100 assert list_records(root) == [] def test_50_stale_50_fresh_only_stale_removed(self, tmp_path: pathlib.Path) -> None: """50 stale + 50 fresh → only stale removed.""" import hashlib as _hl root = _make_repo(tmp_path) for i in range(50): fp = conflict_fingerprint( _hl.sha256(f"gc-s{i}".encode()).hexdigest(), _hl.sha256(f"gc-t{i}".encode()).hexdigest(), ) _write_stale_preimage(root, fp, age_days=90, path=f"src/s{i}.py") for i in range(50): fp = conflict_fingerprint( _hl.sha256(f"gc-u{i}".encode()).hexdigest(), _hl.sha256(f"gc-v{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/u{i}.py") raw = _parse_json(_invoke(root, "gc", "--age", "30", "--json")) assert raw["removed"] == 50 assert len(list_records(root)) == 50 def test_50_stale_50_resolved_only_stale_removed(self, tmp_path: pathlib.Path) -> None: """50 stale preimage-only + 50 stale resolved → only preimage-only removed.""" import hashlib as _hl root = _make_repo(tmp_path) for i in range(50): fp = conflict_fingerprint( _hl.sha256(f"gc-p{i}".encode()).hexdigest(), _hl.sha256(f"gc-q{i}".encode()).hexdigest(), ) _write_stale_preimage(root, fp, age_days=90, path=f"src/p{i}.py") for i in range(50): fp = conflict_fingerprint( _hl.sha256(f"gc-r{i}".encode()).hexdigest(), _hl.sha256(f"gc-w{i}".encode()).hexdigest(), ) _write_stale_preimage(root, fp, age_days=90, path=f"src/r{i}.py") _write_resolution(root, fp) raw = _parse_json(_invoke(root, "gc", "--age", "1", "--json")) assert raw["removed"] == 50 assert len(list_records(root)) == 50 # --------------------------------------------------------------------------- # Integration — clear subcommand # --------------------------------------------------------------------------- class TestClearIntegration: def test_clear_yes_removes_all(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "clear", "--yes") assert result.exit_code == 0 assert not list_records(root) def test_clear_json_schema(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "clear", "--yes", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "removed" in raw removed_val = raw["removed"] assert isinstance(removed_val, int) assert removed_val == 1 def test_clear_empty_cache_json(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "clear", "--yes", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert raw["removed"] == 0 # =========================================================================== # TestRerereclearExtended — 18 tests # =========================================================================== class TestRerereclearExtended: def test_exit_code_zero_yes_empty(self, tmp_path: pathlib.Path) -> None: """--yes on empty cache exits 0.""" root = _make_repo(tmp_path) result = _invoke(root, "clear", "--yes") assert result.exit_code == 0 def test_exit_code_zero_yes_with_entries(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "clear", "--yes") assert result.exit_code == 0 def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: result = _invoke(tmp_path, "clear", "--yes") assert result.exit_code == 2 def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "clear", "--yes", "--format", "xml") assert result.exit_code == 1 def test_j_alias(self, tmp_path: pathlib.Path) -> None: """-j must produce identical output to --json.""" root = _make_repo(tmp_path) r1 = _invoke(root, "clear", "--yes", "--json") r2 = _invoke(root, "clear", "--yes", "-j") assert r1.exit_code == 0 and r2.exit_code == 0 assert _parse_json(r1) == _parse_json(r2) def test_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None: """JSON output must be compact — no indent=2.""" root = _make_repo(tmp_path) result = _invoke(root, "clear", "--yes", "--json") assert result.exit_code == 0 assert "\n" not in result.output.strip() def test_json_removed_zero_when_empty(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) raw = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw["removed"] == 0 def test_json_removed_count_matches_entries(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_B, _SHA_C) _write_preimage(root, _VALID_FP, path="src/a.py") _write_preimage(root, fp2, path="src/b.py") raw = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw["removed"] == 2 def test_entries_gone_after_clear(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) _invoke(root, "clear", "--yes") assert list_records(root) == [] def test_clear_is_idempotent(self, tmp_path: pathlib.Path) -> None: """Clearing an already-empty cache returns removed=0.""" root = _make_repo(tmp_path) _invoke(root, "clear", "--yes") raw = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw["removed"] == 0 def test_text_output_shows_count(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "clear", "--yes") assert "1" in result.output def test_text_output_empty_cache_message(self, tmp_path: pathlib.Path) -> None: """Empty cache --yes should still exit 0 (skips prompt path entirely).""" root = _make_repo(tmp_path) result = _invoke(root, "clear", "--yes") assert result.exit_code == 0 def test_format_text_explicit(self, tmp_path: pathlib.Path) -> None: """--format text is identical to default.""" root = _make_repo(tmp_path) r_default = _invoke(root, "clear", "--yes") r_text = _invoke(root, "clear", "--yes", "--format", "text") assert r_default.output == r_text.output def test_y_alias_same_as_yes(self, tmp_path: pathlib.Path) -> None: """-y must behave identically to --yes.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "clear", "-y") assert result.exit_code == 0 assert list_records(root) == [] def test_resolved_entries_also_cleared(self, tmp_path: pathlib.Path) -> None: """Both preimage-only and resolved entries are removed.""" root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_B, _SHA_C) _write_preimage(root, _VALID_FP) _write_preimage(root, fp2) _write_resolution(root, _VALID_FP) raw = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw["removed"] == 2 assert list_records(root) == [] def test_json_schema_field_present(self, tmp_path: pathlib.Path) -> None: raw = _parse_json(_invoke(_make_repo(tmp_path), "clear", "--yes", "--json")) assert "removed" in raw assert isinstance(raw["removed"], int) def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "clear", "--help") assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "clear", "--help") assert "Exit codes" in result.output # =========================================================================== # TestRerereclearSecurity — 6 tests # =========================================================================== class TestRerereclearSecurity: def test_no_json_outside_repo(self, tmp_path: pathlib.Path) -> None: """Outside a repo must not emit JSON — exits 2.""" result = _invoke(tmp_path, "clear", "--yes", "--json") assert result.exit_code == 2 assert not result.output.strip().startswith("{") def test_no_traceback_outside_repo(self, tmp_path: pathlib.Path) -> None: result = _invoke(tmp_path, "clear", "--yes") assert result.exit_code == 2 assert "Traceback" not in result.output def test_symlink_entry_not_removed(self, tmp_path: pathlib.Path) -> None: """A symlink inside rr-cache must not be removed by clear.""" root = _make_repo(tmp_path) cache = rr_cache_dir(root) target = tmp_path / "external_dir" target.mkdir() link = cache / ("l" * 64) link.symlink_to(target) _invoke(root, "clear", "--yes") assert link.exists() # symlink survives def test_symlink_not_counted_in_removed(self, tmp_path: pathlib.Path) -> None: """Symlink entries must not appear in removed count.""" root = _make_repo(tmp_path) cache = rr_cache_dir(root) target = tmp_path / "ext2" target.mkdir() link = cache / ("m" * 64) link.symlink_to(target) raw = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw["removed"] == 0 def test_no_traceback_on_invalid_format(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "clear", "--yes", "--format", "toml") assert "Traceback" not in result.output def test_json_stdout_on_success(self, tmp_path: pathlib.Path) -> None: """JSON output goes to stdout on success.""" root = _make_repo(tmp_path) result = _invoke(root, "clear", "--yes", "--json") assert result.output.strip().startswith("{") # =========================================================================== # TestRerereclearStress — 3 tests # =========================================================================== class TestRerereclearStress: def test_100_entries_cleared(self, tmp_path: pathlib.Path) -> None: """100 entries all removed in one clear call.""" import hashlib as _hl root = _make_repo(tmp_path) for i in range(100): fp = conflict_fingerprint( _hl.sha256(f"a{i}".encode()).hexdigest(), _hl.sha256(f"b{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/f{i}.py") raw = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw["removed"] == 100 assert list_records(root) == [] def test_clear_then_repopulate_then_clear(self, tmp_path: pathlib.Path) -> None: """Clear → repopulate → clear again: counts correct both times.""" import hashlib as _hl root = _make_repo(tmp_path) for i in range(10): fp = conflict_fingerprint( _hl.sha256(f"c{i}".encode()).hexdigest(), _hl.sha256(f"d{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/g{i}.py") raw1 = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw1["removed"] == 10 for i in range(5): fp = conflict_fingerprint( _hl.sha256(f"e{i}".encode()).hexdigest(), _hl.sha256(f"f{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/h{i}.py") raw2 = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw2["removed"] == 5 def test_mixed_preimage_and_resolved_100_entries(self, tmp_path: pathlib.Path) -> None: """50 preimage-only + 50 resolved → removed=100.""" import hashlib as _hl root = _make_repo(tmp_path) for i in range(100): fp = conflict_fingerprint( _hl.sha256(f"p{i}".encode()).hexdigest(), _hl.sha256(f"q{i}".encode()).hexdigest(), ) _write_preimage(root, fp, path=f"src/k{i}.py") if i < 50: _write_resolution(root, fp) raw = _parse_json(_invoke(root, "clear", "--yes", "--json")) assert raw["removed"] == 100 # --------------------------------------------------------------------------- # Security — CLI # --------------------------------------------------------------------------- class TestReRereSecurity: _ANSI = "\x1b[31mevil\x1b[0m" def test_ansi_in_path_sanitized_in_status(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) fp2 = conflict_fingerprint(_SHA_A, _SHA_C) # Write meta with ANSI in path cache = rr_cache_dir(root) entry_dir = cache / fp2 entry_dir.mkdir(parents=True, exist_ok=True) meta = { "fingerprint": fp2, "path": f"src/{self._ANSI}.py", "ours_id": _SHA_A, "theirs_id": _SHA_C, "domain": "code", "recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } (entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8") result = _invoke(root, "status") assert result.exit_code == 0 assert "\x1b[" not in result.output def test_ansi_in_path_sanitized_in_forget(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "forget", "--fingerprint", _VALID_FP) assert result.exit_code == 0 assert "\x1b[" not in result.output def test_invalid_format_rejected(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "status", "--format", "xml") assert result.exit_code == ExitCode.USER_ERROR.value def test_no_traceback_on_invalid_format(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--format", "toml") assert "Traceback" not in result.output def test_json_on_stdout_text_on_stderr(self, tmp_path: pathlib.Path) -> None: """JSON output goes to stdout; error output goes to stderr.""" root = _make_repo(tmp_path) _write_preimage(root, _VALID_FP) result = _invoke(root, "status", "--json") # stdout (result.output) must start with valid JSON stripped = result.output.lstrip() assert stripped.startswith("{"), f"Expected JSON on stdout: {stripped[:80]!r}" # --------------------------------------------------------------------------- # E2E — JSON schemas for all subcommands # --------------------------------------------------------------------------- class TestJsonSchemas: def test_apply_json_schema(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "dry_run" in raw assert "auto_resolved" in raw assert "remaining" in raw def test_apply_dry_run_json(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "--dry-run", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert raw["dry_run"] is True def test_record_json_schema(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "record", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "recorded" in raw assert "skipped" in raw def test_forget_json_schema(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "forget", "--fingerprint", _VALID_FP, "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "forgotten" in raw assert "not_found" in raw def test_clear_json_schema(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "clear", "--yes", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "removed" in raw def test_gc_json_schema_fields(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "gc", "--json") assert result.exit_code == 0 raw = _parse_json(result) assert "removed" in raw # --------------------------------------------------------------------------- # E2E — help text # --------------------------------------------------------------------------- class TestHelpText: def test_help_shows_subcommands(self) -> None: result = runner.invoke(cli, ["rerere", "--help"]) assert result.exit_code == 0 for sub in ("record", "status", "forget", "clear", "gc"): assert sub in result.output def test_forget_help_shows_fingerprint_flag(self) -> None: result = runner.invoke(cli, ["rerere", "forget", "--help"]) assert result.exit_code == 0 assert "--fingerprint" in result.output def test_gc_help_shows_age_flag(self) -> None: result = runner.invoke(cli, ["rerere", "gc", "--help"]) assert result.exit_code == 0 assert "--age" in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_list_records_10k_entries(self, tmp_path: pathlib.Path) -> None: import hashlib root = _make_repo(tmp_path) cache = rr_cache_dir(root) # Write 1000 valid preimage entries (10k would be slow; 1k validates the cap logic) fps: list[str] = [] for i in range(1000): raw = f"{i:064d}".encode() fp = hashlib.sha256(raw).hexdigest() entry = cache / fp entry.mkdir(parents=True, exist_ok=True) meta = { "fingerprint": fp, "path": f"src/{i}.py", "ours_id": _SHA_A, "theirs_id": _SHA_B, "domain": "code", "recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } (entry / "meta.json").write_text(json.dumps(meta), encoding="utf-8") fps.append(fp) records = list_records(root) assert len(records) == 1000 def test_concurrent_record_load_isolated(self, tmp_path: pathlib.Path) -> None: """Eight threads record and load rerere entries on isolated repos.""" errors: list[str] = [] def worker(idx: int) -> None: try: repo = _make_repo(tmp_path / f"repo{idx}") fp = conflict_fingerprint(_SHA_A, f"{idx:064d}") entry_dir = rr_cache_dir(repo) / fp entry_dir.mkdir(parents=True, exist_ok=True) meta = { "fingerprint": fp, "path": f"src/{idx}.py", "ours_id": _SHA_A, "theirs_id": f"{idx:064d}", "domain": "code", "recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } (entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8") record = load_record(repo, fp) if record is None: errors.append(f"Thread {idx}: load_record returned None") elif record.path != f"src/{idx}.py": errors.append(f"Thread {idx}: wrong path {record.path!r}") except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], f"Concurrent rerere failures: {errors}" def test_concurrent_gc_stale_isolated(self, tmp_path: pathlib.Path) -> None: """Eight threads run gc_stale on isolated repos simultaneously.""" errors: list[str] = [] def worker(idx: int) -> None: try: repo = _make_repo(tmp_path / f"repo{idx}") # Write an old entry cache = rr_cache_dir(repo) entry_dir = cache / _VALID_FP entry_dir.mkdir(parents=True, exist_ok=True) old_time = ( datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=100) ).isoformat() meta = { "fingerprint": _VALID_FP, "path": "src/a.py", "ours_id": _SHA_A, "theirs_id": _SHA_B, "domain": "code", "recorded_at": old_time, } (entry_dir / "meta.json").write_text(json.dumps(meta), encoding="utf-8") removed = gc_stale(repo, age_days=1) if removed != 1: errors.append(f"Thread {idx}: expected 1 removed, got {removed}") except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], f"Concurrent gc_stale failures: {errors}" # =========================================================================== # TestRerereRecordExtended — 18 tests # =========================================================================== # --------------------------------------------------------------------------- # Helpers for record tests # --------------------------------------------------------------------------- def _write_merge_state( root: pathlib.Path, ours_commit: str, theirs_commit: str, conflict_paths: list[str], ) -> None: """Write a minimal MERGE_STATE.json for record tests.""" state = { "ours_commit": ours_commit, "theirs_commit": theirs_commit, "conflict_paths": conflict_paths, "base_commit": None, "other_branch": "feat/other", } ms_path = root / ".muse" / "MERGE_STATE.json" ms_path.write_text(json.dumps(state), encoding="utf-8") def _write_commit_with_manifest( root: pathlib.Path, tag: str, manifest: Manifest, ) -> str: """Write a valid commit + snapshot. Returns the real content-hash commit_id.""" from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, write_commit, write_snapshot, SnapshotRecord import datetime as _dt snap_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = _dt.datetime(2024, 1, 1, tzinfo=_dt.timezone.utc) commit_id = compute_commit_id([], snap_id, f"commit {tag}", committed_at.isoformat()) write_commit( root, CommitRecord( commit_id=commit_id, repo_id="rerere-record-test", branch="main", snapshot_id=snap_id, message=f"commit {tag}", committed_at=committed_at, parent_commit_id=None, ), ) return commit_id def _make_record_repo( tmp_path: pathlib.Path, paths: list[str] | None = None, ours_manifest_extra: Manifest | None = None, theirs_manifest_extra: Manifest | None = None, ) -> tuple[pathlib.Path, str, str]: """ Create a repo with a MERGE_STATE that has real commits/snapshots. Returns (root, ours_commit_id, theirs_commit_id). """ import hashlib as _hl root = _make_repo(tmp_path) conflict_paths = paths or ["src/a.py"] # Build manifests so each conflict path has both ours and theirs blobs. ours_manifest: Manifest = {} theirs_manifest: Manifest = {} for p in conflict_paths: ours_manifest[p] = _hl.sha256(f"ours-{p}".encode()).hexdigest() theirs_manifest[p] = _hl.sha256(f"theirs-{p}".encode()).hexdigest() if ours_manifest_extra: ours_manifest.update(ours_manifest_extra) if theirs_manifest_extra: theirs_manifest.update(theirs_manifest_extra) ours_id = _write_commit_with_manifest(root, "ours", ours_manifest) theirs_id = _write_commit_with_manifest(root, "theirs", theirs_manifest) _write_merge_state(root, ours_id, theirs_id, conflict_paths) return root, ours_id, theirs_id class TestRerereRecordExtended: def test_exits_0_no_merge_in_progress(self, tmp_path: pathlib.Path) -> None: """No MERGE_STATE → exit 0 with a 'nothing to record' message.""" root = _make_repo(tmp_path) result = _invoke(root, "record") assert result.exit_code == 0 def test_text_nothing_to_record_message(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "record") assert "nothing to record" in result.output.lower() def test_exits_0_with_conflicts(self, tmp_path: pathlib.Path) -> None: root, _, _ = _make_record_repo(tmp_path) result = _invoke(root, "record") assert result.exit_code == 0 def test_text_shows_recorded_path(self, tmp_path: pathlib.Path) -> None: root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"]) result = _invoke(root, "record") assert result.exit_code == 0 assert "src/a.py" in result.output def test_rr_cache_entry_created(self, tmp_path: pathlib.Path) -> None: """After record, a meta.json must exist in rr-cache.""" root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"]) _invoke(root, "record") cache = rr_cache_dir(root) entries = list(cache.iterdir()) if cache.exists() else [] assert len(entries) >= 1 assert any((e / "meta.json").exists() for e in entries if e.is_dir()) def test_idempotent_double_record(self, tmp_path: pathlib.Path) -> None: """Recording twice must not create duplicate entries.""" root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"]) _invoke(root, "record") cache_before = set(rr_cache_dir(root).iterdir()) _invoke(root, "record") cache_after = set(rr_cache_dir(root).iterdir()) assert cache_before == cache_after def test_json_schema_no_conflicts(self, tmp_path: pathlib.Path) -> None: """No merge → JSON has recorded=[] skipped=[].""" root = _make_repo(tmp_path) result = _invoke(root, "record", "--json") assert result.exit_code == 0 data = _parse_json(result) assert data["recorded"] == [] assert data["skipped"] == [] def test_json_schema_with_conflicts(self, tmp_path: pathlib.Path) -> None: root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"]) result = _invoke(root, "record", "--json") assert result.exit_code == 0 data = _parse_json(result) assert "src/a.py" in data["recorded"] assert isinstance(data["skipped"], list) def test_json_is_single_line(self, tmp_path: pathlib.Path) -> None: """JSON output must be compact — no indent=2.""" root = _make_repo(tmp_path) result = _invoke(root, "record", "--json") assert result.exit_code == 0 assert "\n" not in result.output.strip() def test_j_alias(self, tmp_path: pathlib.Path) -> None: """-j must produce identical output to --json.""" root = _make_repo(tmp_path) r1 = _invoke(root, "record", "--json") r2 = _invoke(root, "record", "-j") assert r1.exit_code == 0 and r2.exit_code == 0 assert _parse_json(r1) == _parse_json(r2) def test_skipped_path_one_side_deleted(self, tmp_path: pathlib.Path) -> None: """A path present only in one manifest appears in skipped, not recorded.""" root = _make_repo(tmp_path) # ours has the file, theirs does not (deletion conflict). ours_id = _write_commit_with_manifest(root, "ours-del", {"src/deleted.py": _SHA_A}) theirs_id = _write_commit_with_manifest(root, "theirs-del", {}) _write_merge_state(root, ours_id, theirs_id, ["src/deleted.py"]) result = _invoke(root, "record", "--json") assert result.exit_code == 0 data = _parse_json(result) assert "src/deleted.py" in data["skipped"] assert "src/deleted.py" not in data["recorded"] def test_multiple_paths_all_recorded(self, tmp_path: pathlib.Path) -> None: """Multiple conflict paths all appear in recorded.""" paths = ["src/a.py", "src/b.py", "src/c.py"] root, _, _ = _make_record_repo(tmp_path, paths=paths) result = _invoke(root, "record", "--json") assert result.exit_code == 0 data = _parse_json(result) for p in paths: assert p in data["recorded"] def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: """No .muse directory → exit 2.""" result = _invoke(tmp_path, "record") assert result.exit_code == 2 def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "record", "--help") assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "record", "--help") assert "Exit codes" in result.output def test_help_mentions_json_schema(self) -> None: result = _invoke(_make_repo(__import__("pathlib").Path("/tmp")), "record", "--help") assert "JSON output schema" in result.output def test_format_text_explicit(self, tmp_path: pathlib.Path) -> None: """--format text must produce the same output as the default.""" root = _make_repo(tmp_path) r_default = _invoke(root, "record") r_text = _invoke(root, "record", "--format", "text") assert r_default.output == r_text.output def test_invalid_format_exits_1(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _invoke(root, "record", "--format", "xml") assert result.exit_code == 1 # =========================================================================== # TestRerereRecordSecurity — 6 tests # =========================================================================== class TestRerereRecordSecurity: def test_ansi_path_stripped_text(self, tmp_path: pathlib.Path) -> None: """ANSI escape in a conflict path must be stripped in text output.""" root = _make_repo(tmp_path) evil_path = "src/\x1b[31mevil\x1b[0m.py" ours_id = _write_commit_with_manifest(root, "ours-ansi", {evil_path: _SHA_A}) theirs_id = _write_commit_with_manifest(root, "theirs-ansi", {evil_path: _SHA_B}) _write_merge_state(root, ours_id, theirs_id, [evil_path]) result = _invoke(root, "record") assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_path_stripped_json(self, tmp_path: pathlib.Path) -> None: """ANSI escape in a conflict path must be stripped in JSON output.""" root = _make_repo(tmp_path) evil_path = "src/\x1b[31mevil\x1b[0m.py" ours_id = _write_commit_with_manifest(root, "ours-ansi-j", {evil_path: _SHA_A}) theirs_id = _write_commit_with_manifest(root, "theirs-ansi-j", {evil_path: _SHA_B}) _write_merge_state(root, ours_id, theirs_id, [evil_path]) result = _invoke(root, "record", "--json") assert result.exit_code == 0 assert "\x1b" not in result.output def test_skipped_ansi_path_stripped_json(self, tmp_path: pathlib.Path) -> None: """ANSI in a skipped path must also be stripped in JSON output.""" root = _make_repo(tmp_path) evil_path = "src/\x1b[32mskipped\x1b[0m.py" # Only ours has the file → skipped. ours_id = _write_commit_with_manifest(root, "ours-skip-ansi", {evil_path: _SHA_A}) theirs_id = _write_commit_with_manifest(root, "theirs-skip-ansi", {}) _write_merge_state(root, ours_id, theirs_id, [evil_path]) result = _invoke(root, "record", "--json") assert result.exit_code == 0 assert "\x1b" not in result.output def test_no_json_on_error(self, tmp_path: pathlib.Path) -> None: """Incomplete MERGE_STATE error must not emit JSON to stdout.""" import hashlib as _hl root = _make_repo(tmp_path) # Write a MERGE_STATE that references non-existent commits. _write_merge_state(root, "x" * 64, "y" * 64, ["src/a.py"]) result = _invoke(root, "record", "--json") # Exit 1 (incomplete manifests), no JSON blob on stdout. assert not result.output.strip().startswith("{") def test_outside_repo_no_traceback(self, tmp_path: pathlib.Path) -> None: """Running outside a repo must exit cleanly with code 2, no traceback.""" result = _invoke(tmp_path, "record") assert result.exit_code == 2 assert "Traceback" not in result.output def test_control_char_in_path_stripped(self, tmp_path: pathlib.Path) -> None: """Control characters other than ANSI also stripped from text output.""" root = _make_repo(tmp_path) evil_path = "src/evil\x07.py" ours_id = _write_commit_with_manifest(root, "ours-ctrl", {evil_path: _SHA_A}) theirs_id = _write_commit_with_manifest(root, "theirs-ctrl", {evil_path: _SHA_B}) _write_merge_state(root, ours_id, theirs_id, [evil_path]) result = _invoke(root, "record") assert result.exit_code == 0 assert "\x07" not in result.output # =========================================================================== # TestRerereRecordStress — 3 tests # =========================================================================== class TestRerereRecordStress: def test_50_conflict_paths(self, tmp_path: pathlib.Path) -> None: """50 simultaneous conflict paths are all recorded.""" paths = [f"src/file_{i}.py" for i in range(50)] root, _, _ = _make_record_repo(tmp_path, paths=paths) result = _invoke(root, "record", "--json") assert result.exit_code == 0 data = _parse_json(result) assert len(data["recorded"]) == 50 def test_idempotent_100_calls(self, tmp_path: pathlib.Path) -> None: """Recording the same conflict 100 times creates only one cache entry.""" root, _, _ = _make_record_repo(tmp_path, paths=["src/a.py"]) for _ in range(100): r = _invoke(root, "record") assert r.exit_code == 0 entries = [e for e in rr_cache_dir(root).iterdir() if e.is_dir()] assert len(entries) == 1 def test_mixed_recorded_and_skipped(self, tmp_path: pathlib.Path) -> None: """Mix of recordable and skipped paths: counts must sum to total conflicts.""" import hashlib as _hl root = _make_repo(tmp_path) good_paths = [f"src/good_{i}.py" for i in range(10)] skip_paths = [f"src/skip_{i}.py" for i in range(5)] all_paths = good_paths + skip_paths ours_manifest = {p: _hl.sha256(f"o-{p}".encode()).hexdigest() for p in good_paths} theirs_manifest = {p: _hl.sha256(f"t-{p}".encode()).hexdigest() for p in good_paths} # skip_paths have no theirs entry → will be skipped. for p in skip_paths: ours_manifest[p] = _hl.sha256(f"o-{p}".encode()).hexdigest() ours_id = _write_commit_with_manifest(root, "ours-mixed", ours_manifest) theirs_id = _write_commit_with_manifest(root, "theirs-mixed", theirs_manifest) _write_merge_state(root, ours_id, theirs_id, all_paths) result = _invoke(root, "record", "--json") assert result.exit_code == 0 data = _parse_json(result) assert len(data["recorded"]) == 10 assert len(data["skipped"]) == 5