"""Hardening tests for ``muse rebase`` and ``muse/core/rebase.py``. Covers: - Security: symlink guard on REBASE_STATE.json (load, save, clear) - Security: size cap on REBASE_STATE.json - Error routing: all user-visible errors go to stderr - JSON schema validation: every outcome shape (_RebaseResultJson, _RebaseStatusJson, _RebaseDryRunJson) - --dry-run: preview without side effects - --status: progress reporting for active and inactive rebase - --max-commits: cap enforcement - Integration: full lifecycle — init → rebase → abort / continue - Stress: 50-commit chain, repeated collect_commits_to_replay calls """ from __future__ import annotations import datetime import hashlib import json import pathlib import threading from typing import TypedDict import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.object_store import write_object from muse.core.rebase import ( RebaseState, RebaseProgress, _MAX_STATE_BYTES, _REBASE_STATE_FILE, clear_rebase_state, collect_commits_to_replay, get_rebase_progress, load_rebase_state, save_rebase_state, ) from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot from muse.core._types import Manifest runner = CliRunner() _REPO_ID = "rebase-harden-test" # --------------------------------------------------------------------------- # TypedDicts for JSON schema validation # --------------------------------------------------------------------------- class _ResultJson(TypedDict): status: str branch: str new_head: str | None onto: str squash: bool replayed: int conflicts: list[str] class _StatusJson(TypedDict): active: bool original_branch: str original_head: str onto: str total: int done: int remaining: int squash: bool class _DryRunCommitJson(TypedDict): commit_id: str message: str class _DryRunJson(TypedDict): branch: str onto: str commits: list[_DryRunCommitJson] count: int squash: bool # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _sha(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _init_repo(path: pathlib.Path) -> pathlib.Path: muse = path / ".muse" for d in ("commits", "snapshots", "objects", "refs/heads"): (muse / d).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": _REPO_ID, "domain": "code"}), encoding="utf-8" ) return path _counter = 0 _counter_lock = threading.Lock() def _make_commit( root: pathlib.Path, parent_id: str | None = None, content: bytes = b"data", branch: str = "main", ) -> str: global _counter with _counter_lock: _counter += 1 c_val = _counter c = content + str(c_val).encode() obj_id = _sha(c) write_object(root, obj_id, c) manifest = {f"f_{c_val}.txt": obj_id} snap_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = datetime.datetime.now(datetime.timezone.utc) parent_ids = [parent_id] if parent_id else [] commit_id = compute_commit_id(parent_ids, snap_id, f"commit {c_val}", committed_at.isoformat()) write_commit(root, CommitRecord( commit_id=commit_id, repo_id=_REPO_ID, branch=branch, snapshot_id=snap_id, message=f"commit {c_val}", committed_at=committed_at, parent_commit_id=parent_id, )) (root / ".muse" / "refs" / "heads" / branch).write_text(commit_id, encoding="utf-8") return commit_id def _env(repo: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(repo)} def _json_blob(output: str) -> str: """Extract the first JSON object from mixed CLI output.""" for line in output.splitlines(): line = line.strip() if line.startswith("{") or line.startswith("["): return line return output.strip() def _parse_result(output: str) -> _ResultJson: raw = json.loads(_json_blob(output)) assert isinstance(raw, dict), f"Expected dict, got: {raw!r}" status_val = raw["status"] branch_val = raw["branch"] new_head_val = raw.get("new_head") onto_val = raw["onto"] squash_val = raw["squash"] replayed_val = raw["replayed"] conflicts_val = raw["conflicts"] assert isinstance(status_val, str) assert isinstance(branch_val, str) assert new_head_val is None or isinstance(new_head_val, str) assert isinstance(onto_val, str) assert isinstance(squash_val, bool) assert isinstance(replayed_val, int) assert isinstance(conflicts_val, list) return _ResultJson( status=status_val, branch=branch_val, new_head=new_head_val, onto=onto_val, squash=squash_val, replayed=replayed_val, conflicts=conflicts_val, ) def _parse_status(output: str) -> _StatusJson: raw = json.loads(_json_blob(output)) assert isinstance(raw, dict) active_val = raw["active"] ob_val = raw["original_branch"] oh_val = raw["original_head"] onto_val = raw["onto"] total_val = raw["total"] done_val = raw["done"] remaining_val = raw["remaining"] squash_val = raw["squash"] assert isinstance(active_val, bool) assert isinstance(ob_val, str) assert isinstance(oh_val, str) assert isinstance(onto_val, str) assert isinstance(total_val, int) assert isinstance(done_val, int) assert isinstance(remaining_val, int) assert isinstance(squash_val, bool) return _StatusJson( active=active_val, original_branch=ob_val, original_head=oh_val, onto=onto_val, total=total_val, done=done_val, remaining=remaining_val, squash=squash_val, ) def _parse_dry_run(output: str) -> _DryRunJson: raw = json.loads(_json_blob(output)) assert isinstance(raw, dict) branch_val = raw["branch"] onto_val = raw["onto"] commits_raw = raw["commits"] count_val = raw["count"] squash_val = raw["squash"] assert isinstance(branch_val, str) assert isinstance(onto_val, str) assert isinstance(commits_raw, list) assert isinstance(count_val, int) assert isinstance(squash_val, bool) commits: list[_DryRunCommitJson] = [] for entry in commits_raw: assert isinstance(entry, dict) cid = entry["commit_id"] msg = entry["message"] assert isinstance(cid, str) assert isinstance(msg, str) commits.append(_DryRunCommitJson(commit_id=cid, message=msg)) return _DryRunJson( branch=branch_val, onto=onto_val, commits=commits, count=count_val, squash=squash_val, ) def _invoke(args: list[str], repo: pathlib.Path) -> InvokeResult: return runner.invoke(None, args, env=_env(repo)) # --------------------------------------------------------------------------- # Unit: Security — symlink guard on load_rebase_state # --------------------------------------------------------------------------- def test_load_rebase_state_symlink_is_rejected(tmp_path: pathlib.Path) -> None: """A symlink planted at REBASE_STATE.json must be silently ignored.""" _init_repo(tmp_path) state_path = tmp_path / _REBASE_STATE_FILE target = tmp_path / "sensitive.json" target.write_text( json.dumps({ "original_branch": "main", "original_head": "a" * 64, "onto": "b" * 64, "remaining": [], "completed": [], "squash": False, }), encoding="utf-8", ) state_path.symlink_to(target) result = load_rebase_state(tmp_path) assert result is None, "Symlinked state file should be rejected" def test_save_rebase_state_symlink_is_rejected(tmp_path: pathlib.Path) -> None: """Writing state to a symlink must raise OSError.""" _init_repo(tmp_path) state_path = tmp_path / _REBASE_STATE_FILE target = tmp_path / "victim.json" target.write_text("{}", encoding="utf-8") state_path.symlink_to(target) state = RebaseState( original_branch="main", original_head="a" * 64, onto="b" * 64, remaining=[], completed=[], squash=False, ) with pytest.raises(OSError, match="symlink"): save_rebase_state(tmp_path, state) # Victim file must be untouched. assert target.read_text(encoding="utf-8") == "{}" def test_clear_rebase_state_symlink_is_rejected(tmp_path: pathlib.Path) -> None: """clear_rebase_state must not unlink a symlink.""" _init_repo(tmp_path) state_path = tmp_path / _REBASE_STATE_FILE target = tmp_path / "do_not_delete.json" target.write_text("important", encoding="utf-8") state_path.symlink_to(target) clear_rebase_state(tmp_path) # must not raise assert target.exists(), "Target file must not be deleted through symlink" assert state_path.exists(), "Symlink itself should remain" # --------------------------------------------------------------------------- # Unit: Security — size cap on load_rebase_state # --------------------------------------------------------------------------- def test_load_rebase_state_size_cap_rejected(tmp_path: pathlib.Path) -> None: """A state file exceeding _MAX_STATE_BYTES must be rejected.""" _init_repo(tmp_path) state_path = tmp_path / _REBASE_STATE_FILE # Write more than the cap directly. state_path.write_bytes(b"x" * (_MAX_STATE_BYTES + 1)) result = load_rebase_state(tmp_path) assert result is None, "Oversized state file should be rejected" def test_load_rebase_state_exactly_at_cap_is_rejected(tmp_path: pathlib.Path) -> None: """A state file at exactly _MAX_STATE_BYTES (but invalid JSON) is rejected.""" _init_repo(tmp_path) state_path = tmp_path / _REBASE_STATE_FILE state_path.write_bytes(b"y" * _MAX_STATE_BYTES) result = load_rebase_state(tmp_path) # Content is invalid JSON, so also None — but the size check fires first. assert result is None # --------------------------------------------------------------------------- # Unit: get_rebase_progress # --------------------------------------------------------------------------- def test_get_rebase_progress_inactive(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) p = get_rebase_progress(tmp_path) assert isinstance(p, dict) assert p["active"] is False assert p["total"] == 0 assert p["done"] == 0 assert p["remaining"] == 0 def test_get_rebase_progress_active(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) state = RebaseState( original_branch="feat/x", original_head="a" * 64, onto="b" * 64, remaining=["c" * 64, "d" * 64], completed=["e" * 64], squash=True, ) save_rebase_state(tmp_path, state) p = get_rebase_progress(tmp_path) assert p["active"] is True assert p["original_branch"] == "feat/x" assert p["total"] == 3 # 1 done + 2 remaining assert p["done"] == 1 assert p["remaining"] == 2 assert p["squash"] is True # --------------------------------------------------------------------------- # Error routing: errors go to stderr # --------------------------------------------------------------------------- def test_abort_no_state_error_to_stderr(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["rebase", "--abort"], tmp_path) assert result.exit_code != 0 assert "No rebase" in result.output or "No rebase" in (result.stderr or "") def test_continue_no_state_error_to_stderr(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["rebase", "--continue"], tmp_path) assert result.exit_code != 0 assert "Nothing to continue" in result.output or "No rebase" in result.output def test_rebase_in_progress_error_to_stderr(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path) state = RebaseState( original_branch="main", original_head=base, onto=base, remaining=[], completed=[], squash=False, ) save_rebase_state(tmp_path, state) result = _invoke(["rebase", "main"], tmp_path) assert result.exit_code != 0 # Error should mention --continue or --abort assert "--continue" in result.output or "--abort" in result.output def test_no_upstream_error_to_stderr(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path) result = _invoke(["rebase"], tmp_path) assert result.exit_code != 0 def test_unknown_upstream_error(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path) result = _invoke(["rebase", "nonexistent-branch"], tmp_path) assert result.exit_code != 0 assert "not found" in result.output.lower() # --------------------------------------------------------------------------- # JSON schema: --status # --------------------------------------------------------------------------- def test_status_json_inactive(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["rebase", "--status", "--json"], tmp_path) assert result.exit_code == 0 parsed = _parse_status(result.output) assert parsed["active"] is False assert parsed["total"] == 0 assert parsed["done"] == 0 assert parsed["remaining"] == 0 def test_status_json_active(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) state = RebaseState( original_branch="feat/x", original_head="a" * 64, onto="b" * 64, remaining=["c" * 64, "d" * 64, "f" * 64], completed=["e" * 64, "g" * 64], squash=False, ) save_rebase_state(tmp_path, state) result = _invoke(["rebase", "--status", "--json"], tmp_path) assert result.exit_code == 0 parsed = _parse_status(result.output) assert parsed["active"] is True assert parsed["original_branch"] == "feat/x" assert parsed["total"] == 5 assert parsed["done"] == 2 assert parsed["remaining"] == 3 assert parsed["squash"] is False def test_status_text_inactive(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["rebase", "--status"], tmp_path) assert result.exit_code == 0 assert "No rebase" in result.output def test_status_text_active(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) state = RebaseState( original_branch="feat/y", original_head="a" * 64, onto="b" * 64, remaining=["c" * 64], completed=["d" * 64], squash=True, ) save_rebase_state(tmp_path, state) result = _invoke(["rebase", "--status"], tmp_path) assert result.exit_code == 0 assert "feat/y" in result.output assert "1/2" in result.output # --------------------------------------------------------------------------- # JSON schema: --abort # --------------------------------------------------------------------------- def test_abort_json_schema(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path) tip = _make_commit(tmp_path, parent_id=base) state = RebaseState( original_branch="main", original_head=base, onto=base, remaining=[tip], completed=[], squash=False, ) save_rebase_state(tmp_path, state) result = _invoke(["rebase", "--abort", "--json"], tmp_path) assert result.exit_code == 0, result.output parsed = _parse_result(result.output) assert parsed["status"] == "aborted" assert parsed["branch"] == "main" assert parsed["new_head"] == base assert parsed["squash"] is False assert parsed["conflicts"] == [] # --------------------------------------------------------------------------- # JSON schema: already up to date # --------------------------------------------------------------------------- def test_already_up_to_date_json(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) cid = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( cid, encoding="utf-8" ) result = _invoke(["rebase", "--json", "upstream"], tmp_path) assert result.exit_code == 0, result.output parsed = _parse_result(result.output) assert parsed["status"] == "up_to_date" assert parsed["replayed"] == 0 assert parsed["conflicts"] == [] # --------------------------------------------------------------------------- # JSON schema: --dry-run # --------------------------------------------------------------------------- def test_dry_run_json_schema(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( base, encoding="utf-8" ) c1 = _make_commit(tmp_path, parent_id=base) c2 = _make_commit(tmp_path, parent_id=c1) result = _invoke(["rebase", "--dry-run", "--json", "upstream"], tmp_path) assert result.exit_code == 0, result.output parsed = _parse_dry_run(result.output) assert parsed["branch"] == "main" assert parsed["onto"] == base assert parsed["count"] == 2 assert parsed["squash"] is False assert len(parsed["commits"]) == 2 assert parsed["commits"][0]["commit_id"] == c1 assert parsed["commits"][1]["commit_id"] == c2 def test_dry_run_no_side_effects(tmp_path: pathlib.Path) -> None: """--dry-run must not write REBASE_STATE.json or modify branch refs.""" _init_repo(tmp_path) base = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( base, encoding="utf-8" ) c1 = _make_commit(tmp_path, parent_id=base) original_head = (tmp_path / ".muse" / "refs" / "heads" / "main").read_text( encoding="utf-8" ).strip() result = _invoke(["rebase", "--dry-run", "upstream"], tmp_path) assert result.exit_code == 0 # State file must not exist after a dry-run. assert not (tmp_path / _REBASE_STATE_FILE).exists() # Branch ref must not change. new_head = (tmp_path / ".muse" / "refs" / "heads" / "main").read_text( encoding="utf-8" ).strip() assert new_head == original_head # Output should describe the plan. assert c1[:12] in result.output def test_dry_run_text_output(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( base, encoding="utf-8" ) c1 = _make_commit(tmp_path, parent_id=base) result = _invoke(["rebase", "--dry-run", "upstream"], tmp_path) assert result.exit_code == 0 assert "Would rebase" in result.output assert c1[:12] in result.output # --------------------------------------------------------------------------- # JSON schema: completed rebase (normal) # --------------------------------------------------------------------------- def test_completed_rebase_json_schema(tmp_path: pathlib.Path) -> None: """muse rebase --json upstream emits a completed result on success.""" _init_repo(tmp_path) base = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( base, encoding="utf-8" ) c1 = _make_commit(tmp_path, parent_id=base) result = _invoke(["rebase", "--json", "upstream"], tmp_path) assert result.exit_code == 0, result.output parsed = _parse_result(result.output) assert parsed["status"] == "completed" assert parsed["branch"] == "main" assert parsed["squash"] is False assert parsed["replayed"] == 1 assert parsed["conflicts"] == [] assert isinstance(parsed["new_head"], str) and len(parsed["new_head"]) == 64 # --------------------------------------------------------------------------- # --max-commits: cap enforcement # --------------------------------------------------------------------------- def test_max_commits_cap(tmp_path: pathlib.Path) -> None: """--max-commits 2 on a 5-commit chain should only replay 2 commits.""" _init_repo(tmp_path) base = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( base, encoding="utf-8" ) prev = base for _ in range(5): prev = _make_commit(tmp_path, parent_id=prev) result = _invoke( ["rebase", "--dry-run", "--json", "--max-commits", "2", "upstream"], tmp_path ) assert result.exit_code == 0, result.output parsed = _parse_dry_run(result.output) # collect_commits_to_replay caps at 2. assert parsed["count"] <= 2 # --------------------------------------------------------------------------- # collect_commits_to_replay: max_commits parameter # --------------------------------------------------------------------------- def test_collect_commits_max_cap(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path) prev = base for i in range(10): prev = _make_commit(tmp_path, parent_id=prev, content=f"cap{i}".encode()) result = collect_commits_to_replay(tmp_path, stop_at=base, tip=prev, max_commits=5) assert len(result) <= 5 # --------------------------------------------------------------------------- # Integration: abort clears state and restores HEAD # --------------------------------------------------------------------------- def test_full_abort_lifecycle(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path) tip = _make_commit(tmp_path, parent_id=base) original_head = (tmp_path / ".muse" / "refs" / "heads" / "main").read_text( encoding="utf-8" ).strip() state = RebaseState( original_branch="main", original_head=base, onto=base, remaining=[tip], completed=[], squash=False, ) save_rebase_state(tmp_path, state) result = _invoke(["rebase", "--abort"], tmp_path) assert result.exit_code == 0 assert "aborted" in result.output.lower() # State cleared. assert load_rebase_state(tmp_path) is None # HEAD restored to original_head (base), not tip. restored = (tmp_path / ".muse" / "refs" / "heads" / "main").read_text( encoding="utf-8" ).strip() assert restored == base # --------------------------------------------------------------------------- # Integration: --dry-run squash flag propagated # --------------------------------------------------------------------------- def test_dry_run_squash_flag(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( base, encoding="utf-8" ) _make_commit(tmp_path, parent_id=base) result = _invoke( ["rebase", "--dry-run", "--squash", "--json", "upstream"], tmp_path ) assert result.exit_code == 0, result.output parsed = _parse_dry_run(result.output) assert parsed["squash"] is True # --------------------------------------------------------------------------- # Integration: already-up-to-date text output # --------------------------------------------------------------------------- def test_already_up_to_date_text(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) cid = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( cid, encoding="utf-8" ) result = _invoke(["rebase", "upstream"], tmp_path) assert result.exit_code == 0 assert "up to date" in result.output.lower() # --------------------------------------------------------------------------- # Integration: --status reflects paused rebase # --------------------------------------------------------------------------- def test_status_after_state_saved(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) state = RebaseState( original_branch="feat/z", original_head="a" * 64, onto="b" * 64, remaining=["c" * 64, "d" * 64], completed=["e" * 64], squash=False, ) save_rebase_state(tmp_path, state) result = _invoke(["rebase", "--status", "--json"], tmp_path) assert result.exit_code == 0 parsed = _parse_status(result.output) assert parsed["active"] is True assert parsed["remaining"] == 2 assert parsed["done"] == 1 # --------------------------------------------------------------------------- # E2E: human-readable text for all outcomes # --------------------------------------------------------------------------- def test_help_output(tmp_path: pathlib.Path) -> None: result = runner.invoke(None, ["rebase", "--help"]) assert result.exit_code == 0 assert "--abort" in result.output assert "--continue" in result.output assert "--dry-run" in result.output assert "--status" in result.output assert "--max-commits" in result.output assert "--json" in result.output assert "--squash" in result.output def test_abort_text_output(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path) state = RebaseState( original_branch="main", original_head=base, onto=base, remaining=[], completed=[], squash=False, ) save_rebase_state(tmp_path, state) result = _invoke(["rebase", "--abort"], tmp_path) assert result.exit_code == 0 assert "aborted" in result.output.lower() assert base[:12] in result.output def test_up_to_date_text(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) cid = _make_commit(tmp_path) (tmp_path / ".muse" / "refs" / "heads" / "up").write_text(cid, encoding="utf-8") result = _invoke(["rebase", "up"], tmp_path) assert result.exit_code == 0 assert "up to date" in result.output.lower() # --------------------------------------------------------------------------- # Stress: 50-commit chain dry-run # --------------------------------------------------------------------------- def test_stress_50_commit_dry_run(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path, content=b"stress-base") (tmp_path / ".muse" / "refs" / "heads" / "upstream").write_text( base, encoding="utf-8" ) prev = base commit_ids: list[str] = [] for i in range(50): prev = _make_commit(tmp_path, parent_id=prev, content=f"s{i}".encode()) commit_ids.append(prev) result = _invoke(["rebase", "--dry-run", "--json", "upstream"], tmp_path) assert result.exit_code == 0, result.output parsed = _parse_dry_run(result.output) assert parsed["count"] == 50 assert len(parsed["commits"]) == 50 # Oldest first. assert parsed["commits"][0]["commit_id"] == commit_ids[0] assert parsed["commits"][-1]["commit_id"] == commit_ids[-1] def test_stress_collect_50_commits(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path, content=b"stress-base-2") prev = base ids: list[str] = [] for i in range(50): prev = _make_commit(tmp_path, parent_id=prev, content=f"t{i}".encode()) ids.append(prev) result = collect_commits_to_replay(tmp_path, stop_at=base, tip=prev) assert len(result) == 50 assert result[0].commit_id == ids[0] assert result[-1].commit_id == ids[-1] def test_stress_status_many_commits(tmp_path: pathlib.Path) -> None: """get_rebase_progress is fast even with a 1000-element state.""" _init_repo(tmp_path) state = RebaseState( original_branch="main", original_head="a" * 64, onto="b" * 64, remaining=["c" * 64] * 500, completed=["d" * 64] * 500, squash=False, ) save_rebase_state(tmp_path, state) p = get_rebase_progress(tmp_path) assert p["total"] == 1000 assert p["done"] == 500 assert p["remaining"] == 500 def test_stress_concurrent_status_reads(tmp_path: pathlib.Path) -> None: """Multiple threads calling get_rebase_progress concurrently must not crash.""" _init_repo(tmp_path) state = RebaseState( original_branch="main", original_head="a" * 64, onto="b" * 64, remaining=["c" * 64] * 10, completed=["d" * 64] * 5, squash=False, ) save_rebase_state(tmp_path, state) errors: list[str] = [] def _read() -> None: try: p = get_rebase_progress(tmp_path) assert p["active"] is True except Exception as exc: # noqa: BLE001 errors.append(str(exc)) threads = [threading.Thread(target=_read) for _ in range(20)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent status failures: {errors}"