"""Tests for ``muse code migrate`` — commit-id-v2 DAG replay. Coverage tiers -------------- Unit: - v1-style commit fixture helper produces a file with mismatched ID - dry-run reports old→new mapping and makes zero writes - bare base64 signature is normalised to ``ed25519:…`` prefix - object path migration: legacy paths moved under ``sha256/`` subdir Integration: - single root commit is rewritten with v2 formula - linear chain: parent IDs cascade correctly through the DAG - merge commit: both parents resolved through id_map - all branch heads updated to new commit IDs - idempotent: running twice produces the same store state - preflight aborts when MERGE_STATE is present - old records carrying ``format_version`` or ``branch`` key still migrate """ from __future__ import annotations import datetime import json import pathlib import msgpack import pytest from tests.cli_test_helper import CliRunner from muse.core._types import MsgpackDict, blob_id, split_id from muse.core.paths import merge_state_path as _merge_state_path from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import ( CommitRecord, SnapshotRecord, commit_path, get_all_branch_heads, read_commit, write_commit, write_snapshot, ) cli = None runner = CliRunner() _REPO_ID = "test-repo-v2-migrate" _AUTHOR = "gabriel" _AT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) _AT_ISO = _AT.isoformat() # --------------------------------------------------------------------------- # Repo bootstrap helpers # --------------------------------------------------------------------------- def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path: """Create a minimal .muse layout with no commits.""" muse = tmp_path / ".muse" for sub in ("commits/sha256", "snapshots/sha256", "objects/sha256", "refs/heads"): (muse / sub).mkdir(parents=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 tmp_path type _Environ = dict[str, str] def _env(repo: pathlib.Path) -> _Environ: return {"MUSE_REPO_ROOT": str(repo)} def _snap(repo: pathlib.Path, tag: str = "s") -> str: manifest: MsgpackDict = {f"file_{tag}.py": f"sha256:{'a' * 64}"} sid = compute_snapshot_id(manifest) write_snapshot(repo, SnapshotRecord( snapshot_id=sid, manifest=manifest, created_at=_AT, )) return sid def _v2_commit( repo: pathlib.Path, tag: str, sid: str, branch: str = "main", parent: str | None = None, author: str = _AUTHOR, signer_public_key: str = "", ) -> str: """Write a proper v2 commit (using compute_commit_id with all fields).""" parent_ids = [parent] if parent else [] cid = compute_commit_id( parent_ids=parent_ids, snapshot_id=sid, message=tag, committed_at_iso=_AT_ISO, repo_id=_REPO_ID, author=author, signer_public_key=signer_public_key, ) write_commit(repo, CommitRecord( commit_id=cid, repo_id=_REPO_ID, created_on_branch=branch, snapshot_id=sid, message=tag, committed_at=_AT, author=author, parent_commit_id=parent, signer_public_key=signer_public_key, )) _set_ref(repo, branch, cid) return cid def _v1_commit_id( parent_ids: list[str], snapshot_id: str, message: str, committed_at_iso: str, ) -> str: """Compute a v1 commit ID (old formula — no repo_id/author/signer).""" _SEP = "\x00" parts = [ _SEP.join(sorted(split_id(p)[1] for p in parent_ids)), split_id(snapshot_id)[1], message, committed_at_iso, ] payload = _SEP.join(parts).encode() return blob_id(payload) def _write_v1_commit_raw( repo: pathlib.Path, tag: str, sid: str, branch: str = "main", parent: str | None = None, author: str = _AUTHOR, signature: str = "", signer_public_key: str = "", extra: MsgpackDict | None = None, ) -> str: """Write a v1-style commit directly to disk, bypassing write_commit. The stored commit_id is computed with the OLD v1 formula so it will NOT match the v2 formula. migrate must detect this mismatch and rewrite. """ parent_ids = [parent] if parent else [] cid = _v1_commit_id(parent_ids, sid, tag, _AT_ISO) record: MsgpackDict = { "commit_id": cid, "repo_id": _REPO_ID, "created_on_branch": branch, "snapshot_id": sid, "message": tag, "committed_at": _AT_ISO, "parent_commit_id": parent, "parent2_commit_id": None, "author": author, "metadata": {}, "structured_delta": None, "sem_ver_bump": "none", "breaking_changes": [], "agent_id": "", "model_id": "", "toolchain_id": "", "prompt_hash": "", "signature": signature, "signer_public_key": signer_public_key, "signer_key_id": "", "reviewed_by": [], "test_runs": 0, "labels": [], "status": "", "notes": [], "score": None, } if extra: record.update(extra) algo, hex_id = split_id(cid) dest = repo / ".muse" / "commits" / algo / f"{hex_id}.msgpack" dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(msgpack.packb(record, use_bin_type=True)) return cid def _set_ref(repo: pathlib.Path, branch: str, commit_id: str) -> None: ref = repo / ".muse" / "refs" / "heads" / branch ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(commit_id, encoding="utf-8") def _invoke(args: list[str], repo: pathlib.Path) -> MsgpackDict: if "--json" not in args: args = args + ["--json"] result = runner.invoke(cli, args, env=_env(repo)) assert result.exit_code == 0, ( f"muse {' '.join(args)} failed (exit {result.exit_code}):\n{result.output}" ) return json.loads(result.output) # --------------------------------------------------------------------------- # Fixture validity sanity-check # --------------------------------------------------------------------------- class TestFixture: def test_v1_commit_id_differs_from_v2(self, tmp_path: pathlib.Path) -> None: """v1 formula must produce a different ID than v2 for the same inputs.""" repo = _init_repo(tmp_path) sid = _snap(repo) v1_id = _v1_commit_id([], sid, "root", _AT_ISO) v2_id = compute_commit_id( parent_ids=[], snapshot_id=sid, message="root", committed_at_iso=_AT_ISO, repo_id=_REPO_ID, author=_AUTHOR, signer_public_key="", ) assert v1_id != v2_id, "v1 and v2 IDs must differ when author/repo_id are non-empty" def test_v1_raw_write_is_unreadable_by_read_commit(self, tmp_path: pathlib.Path) -> None: """read_commit must return None for a v1 commit (ID mismatch).""" repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) assert read_commit(repo, old_id) is None # --------------------------------------------------------------------------- # Preflight # --------------------------------------------------------------------------- class TestPreflight: def test_aborts_when_merge_state_exists(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) _write_v1_commit_raw(repo, "root", sid) _merge_state_path(repo).write_text("{}", encoding="utf-8") result = runner.invoke(cli, ["code", "migrate", "--json"], env=_env(repo)) assert result.exit_code != 0 out = json.loads(result.output) assert "merge" in out.get("error", "").lower() or "merge" in str(out).lower() def test_aborts_when_rebase_in_progress(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) (repo / ".muse" / "rebase-merge").mkdir(parents=True) result = runner.invoke(cli, ["code", "migrate", "--json"], env=_env(repo)) assert result.exit_code != 0 out = json.loads(result.output) assert "rebase" in out.get("error", "").lower() or "rebase" in str(out).lower() # --------------------------------------------------------------------------- # Dry-run # --------------------------------------------------------------------------- class TestDryRun: def test_dry_run_makes_no_writes(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) before = list((repo / ".muse" / "commits").rglob("*.msgpack")) runner.invoke(cli, ["code", "migrate", "--dry-run", "--json"], env=_env(repo)) after = list((repo / ".muse" / "commits").rglob("*.msgpack")) assert set(str(p) for p in before) == set(str(p) for p in after) assert (repo / ".muse" / "refs" / "heads" / "main").read_text() == old_id def test_dry_run_reports_old_to_new_mapping(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) result = runner.invoke( cli, ["code", "migrate", "--dry-run", "--json"], env=_env(repo) ) assert result.exit_code == 0 out = json.loads(result.output) mapping = out.get("id_map", {}) assert old_id in mapping new_id = mapping[old_id] expected = compute_commit_id( parent_ids=[], snapshot_id=sid, message="root", committed_at_iso=_AT_ISO, repo_id=_REPO_ID, author=_AUTHOR, signer_public_key="", ) assert new_id == expected def test_dry_run_reports_summary_counts(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) result = runner.invoke( cli, ["code", "migrate", "--dry-run", "--json"], env=_env(repo) ) assert result.exit_code == 0 out = json.loads(result.output) assert out.get("commits_rewritten", -1) == 1 assert "blobs_migrated" in out assert "dry_run" in out and out["dry_run"] is True # --------------------------------------------------------------------------- # Single root commit # --------------------------------------------------------------------------- class TestSingleRootCommit: def test_rewrites_v1_commit_with_v2_id(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) expected_new_id = compute_commit_id( parent_ids=[], snapshot_id=sid, message="root", committed_at_iso=_AT_ISO, repo_id=_REPO_ID, author=_AUTHOR, signer_public_key="", ) rec = read_commit(repo, expected_new_id) assert rec is not None assert rec.commit_id == expected_new_id def test_old_commit_file_deleted(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) old_path = commit_path(repo, old_id) assert not old_path.exists() def test_already_v2_commit_unchanged(self, tmp_path: pathlib.Path) -> None: """A commit already using v2 ID must not be rewritten (old_id == new_id).""" repo = _init_repo(tmp_path) sid = _snap(repo) v2_id = _v2_commit(repo, "root", sid) _invoke(["code", "migrate"], repo) rec = read_commit(repo, v2_id) assert rec is not None assert rec.commit_id == v2_id # --------------------------------------------------------------------------- # Linear chain — parent ID cascading # --------------------------------------------------------------------------- class TestLinearChain: def _build_chain(self, repo: pathlib.Path, length: int = 3) -> list[str]: old_ids: list[str] = [] sid = _snap(repo, "s0") parent = None for i in range(length): if i > 0: sid = _snap(repo, f"s{i}") old_id = _write_v1_commit_raw(repo, f"c{i}", sid, parent=parent) old_ids.append(old_id) parent = old_id _set_ref(repo, "main", old_ids[-1]) return old_ids def test_chain_all_readable_after_migrate(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) old_ids = self._build_chain(repo, 3) _invoke(["code", "migrate"], repo) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() head_rec = read_commit(repo, new_head) assert head_rec is not None parent_rec = read_commit(repo, head_rec.parent_commit_id) assert parent_rec is not None root_rec = read_commit(repo, parent_rec.parent_commit_id) assert root_rec is not None assert root_rec.parent_commit_id is None def test_chain_old_ids_all_deleted(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) old_ids = self._build_chain(repo, 3) _invoke(["code", "migrate"], repo) for old_id in old_ids: assert not commit_path(repo, old_id).exists() def test_chain_parent_ids_consistent(self, tmp_path: pathlib.Path) -> None: """After migrate, each commit's parent_commit_id must match the new ID of the previous commit in the chain.""" repo = _init_repo(tmp_path) self._build_chain(repo, 3) _invoke(["code", "migrate"], repo) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() tip = read_commit(repo, new_head) mid = read_commit(repo, tip.parent_commit_id) root = read_commit(repo, mid.parent_commit_id) assert root.parent_commit_id is None assert mid.parent_commit_id == root.commit_id assert tip.parent_commit_id == mid.commit_id # --------------------------------------------------------------------------- # Merge commit # --------------------------------------------------------------------------- class TestMergeCommit: def test_merge_commit_both_parents_cascaded(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) root_id = _write_v1_commit_raw(repo, "root", sid) sid_a = _snap(repo, "a") a_id = _write_v1_commit_raw(repo, "feat-a", sid_a, branch="feat/a", parent=root_id) sid_b = _snap(repo, "b") b_id = _write_v1_commit_raw(repo, "feat-b", sid_b, branch="feat/b", parent=root_id) sid_m = _snap(repo, "m") new_root_id = _v1_commit_id([], sid, "root", _AT_ISO) new_a_id = _v1_commit_id([root_id], sid_a, "feat-a", _AT_ISO) new_b_id = _v1_commit_id([root_id], sid_b, "feat-b", _AT_ISO) merge_id_raw = _v1_commit_id([a_id, b_id], sid_m, "merge", _AT_ISO) merge_record = { "commit_id": merge_id_raw, "repo_id": _REPO_ID, "created_on_branch": "main", "snapshot_id": sid_m, "message": "merge", "committed_at": _AT_ISO, "parent_commit_id": a_id, "parent2_commit_id": b_id, "author": _AUTHOR, "metadata": {}, "structured_delta": None, "sem_ver_bump": "none", "breaking_changes": [], "agent_id": "", "model_id": "", "toolchain_id": "", "prompt_hash": "", "signature": "", "signer_public_key": "", "signer_key_id": "", "reviewed_by": [], "test_runs": 0, "labels": [], "status": "", "notes": [], "score": None, } algo, hex_id = split_id(merge_id_raw) dest = repo / ".muse" / "commits" / algo / f"{hex_id}.msgpack" dest.write_bytes(msgpack.packb(merge_record, use_bin_type=True)) _set_ref(repo, "main", merge_id_raw) _set_ref(repo, "feat/a", a_id) _set_ref(repo, "feat/b", b_id) _invoke(["code", "migrate"], repo) new_main = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() merge_rec = read_commit(repo, new_main) assert merge_rec is not None assert merge_rec.parent_commit_id is not None assert merge_rec.parent2_commit_id is not None p1 = read_commit(repo, merge_rec.parent_commit_id) p2 = read_commit(repo, merge_rec.parent2_commit_id) assert p1 is not None assert p2 is not None def test_merge_commit_old_file_deleted(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) root_id = _write_v1_commit_raw(repo, "root", sid) sid2 = _snap(repo, "s2") c2_id = _write_v1_commit_raw(repo, "c2", sid2, parent=root_id) sid3 = _snap(repo, "s3") c3_id = _write_v1_commit_raw(repo, "c3", sid3, parent=root_id) sid_m = _snap(repo, "sm") merge_id = _v1_commit_id([c2_id, c3_id], sid_m, "merge", _AT_ISO) rec = { "commit_id": merge_id, "repo_id": _REPO_ID, "created_on_branch": "main", "snapshot_id": sid_m, "message": "merge", "committed_at": _AT_ISO, "parent_commit_id": c2_id, "parent2_commit_id": c3_id, "author": _AUTHOR, "metadata": {}, "structured_delta": None, "sem_ver_bump": "none", "breaking_changes": [], "agent_id": "", "model_id": "", "toolchain_id": "", "prompt_hash": "", "signature": "", "signer_public_key": "", "signer_key_id": "", "reviewed_by": [], "test_runs": 0, "labels": [], "status": "", "notes": [], "score": None, } dest = repo / ".muse" / "commits" / "sha256" / f"{split_id(merge_id)[1]}.msgpack" dest.write_bytes(msgpack.packb(rec, use_bin_type=True)) _set_ref(repo, "main", merge_id) _invoke(["code", "migrate"], repo) assert not commit_path(repo, merge_id).exists() # --------------------------------------------------------------------------- # Branch heads # --------------------------------------------------------------------------- class TestBranchHeads: def test_all_branch_heads_updated(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) root_id = _write_v1_commit_raw(repo, "root", sid, branch="main") _set_ref(repo, "main", root_id) sid2 = _snap(repo, "s2") feat_id = _write_v1_commit_raw(repo, "feat", sid2, branch="feat/x", parent=root_id) _set_ref(repo, "feat/x", feat_id) _invoke(["code", "migrate"], repo) heads = get_all_branch_heads(repo) for branch, cid in heads.items(): rec = read_commit(repo, cid) assert rec is not None, f"Branch {branch!r} head {cid[:12]} unreadable after migrate" def test_branch_heads_point_to_v2_ids(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() assert new_head != old_id expected = compute_commit_id( parent_ids=[], snapshot_id=sid, message="root", committed_at_iso=_AT_ISO, repo_id=_REPO_ID, author=_AUTHOR, signer_public_key="", ) assert new_head == expected # --------------------------------------------------------------------------- # Signature normalisation # --------------------------------------------------------------------------- class TestSignatureNormalisation: def _fake_pubkey(self) -> str: import base64 return "ed25519:" + base64.urlsafe_b64encode(b"\x01" * 32).rstrip(b"=").decode() def _fake_sig_bytes(self) -> bytes: return b"\x02" * 64 def test_bare_base64_sig_normalised(self, tmp_path: pathlib.Path) -> None: import base64 repo = _init_repo(tmp_path) sid = _snap(repo) bare_sig = base64.urlsafe_b64encode(self._fake_sig_bytes()).rstrip(b"=").decode() pubkey = self._fake_pubkey() old_id = _write_v1_commit_raw( repo, "signed", sid, signature=bare_sig, signer_public_key=pubkey, ) _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() rec = read_commit(repo, new_head) assert rec is not None assert rec.signature.startswith("ed25519:"), ( f"Expected 'ed25519:' prefix on signature, got: {rec.signature!r}" ) def test_already_prefixed_sig_unchanged(self, tmp_path: pathlib.Path) -> None: import base64 repo = _init_repo(tmp_path) sid = _snap(repo) prefixed_sig = "ed25519:" + base64.urlsafe_b64encode(self._fake_sig_bytes()).rstrip(b"=").decode() pubkey = self._fake_pubkey() old_id = _write_v1_commit_raw( repo, "signed", sid, signature=prefixed_sig, signer_public_key=pubkey, ) _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() rec = read_commit(repo, new_head) assert rec is not None assert rec.signature == prefixed_sig def test_empty_sig_not_modified(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "unsigned", sid, signature="") _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() rec = read_commit(repo, new_head) assert rec is not None assert rec.signature == "" # --------------------------------------------------------------------------- # Idempotency # --------------------------------------------------------------------------- class TestIdempotent: def test_running_twice_same_state(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) head_after_first = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() files_after_first = set( p.name for p in (repo / ".muse" / "commits").rglob("*.msgpack") ) _invoke(["code", "migrate"], repo) head_after_second = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() files_after_second = set( p.name for p in (repo / ".muse" / "commits").rglob("*.msgpack") ) assert head_after_first == head_after_second assert files_after_first == files_after_second # --------------------------------------------------------------------------- # Object path migration (Part B) # --------------------------------------------------------------------------- class TestObjectPathMigration: def _write_legacy_blob(self, repo: pathlib.Path, content: bytes) -> str: """Write a blob at the legacy path (no sha256/ subdir).""" oid = blob_id(content) hex_id = split_id(oid)[1] prefix, rest = hex_id[:2], hex_id[2:] legacy = repo / ".muse" / "objects" / prefix / rest legacy.parent.mkdir(parents=True, exist_ok=True) legacy.write_bytes(content) return oid def test_legacy_blob_moved_to_canonical_path(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) content = b"hello blob" oid = self._write_legacy_blob(repo, content) _invoke(["code", "migrate"], repo) hex_id = split_id(oid)[1] prefix, rest = hex_id[:2], hex_id[2:] canonical = repo / ".muse" / "objects" / "sha256" / prefix / rest legacy = repo / ".muse" / "objects" / prefix / rest assert canonical.exists(), "Blob must exist at canonical algo-prefixed path" assert canonical.read_bytes() == content assert not legacy.exists(), "Legacy blob path must be deleted after migration" def test_legacy_dir_removed_when_empty(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) content = b"only blob in shard" oid = self._write_legacy_blob(repo, content) hex_id = split_id(oid)[1] prefix = hex_id[:2] legacy_dir = repo / ".muse" / "objects" / prefix _invoke(["code", "migrate"], repo) assert not legacy_dir.exists(), ( "Empty legacy shard directory must be removed after migration" ) def test_blob_already_at_canonical_not_duplicated(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) content = b"already canonical" oid = blob_id(content) hex_id = split_id(oid)[1] prefix, rest = hex_id[:2], hex_id[2:] canonical = repo / ".muse" / "objects" / "sha256" / prefix / rest canonical.parent.mkdir(parents=True, exist_ok=True) canonical.write_bytes(content) _invoke(["code", "migrate"], repo) assert canonical.exists() assert canonical.read_bytes() == content def test_dry_run_reports_blobs_to_migrate(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) self._write_legacy_blob(repo, b"blob1") self._write_legacy_blob(repo, b"blob2") result = runner.invoke( cli, ["code", "migrate", "--dry-run", "--json"], env=_env(repo) ) assert result.exit_code == 0 out = json.loads(result.output) assert out.get("blobs_migrated", -1) == 2 # --------------------------------------------------------------------------- # Legacy field handling # --------------------------------------------------------------------------- class TestLegacyFields: def test_format_version_in_raw_dict_ignored(self, tmp_path: pathlib.Path) -> None: """Old records with format_version key must still migrate successfully.""" repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw( repo, "root", sid, extra={"format_version": 7} ) _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() rec = read_commit(repo, new_head) assert rec is not None assert not hasattr(rec, "format_version") def test_old_branch_key_in_raw_dict_handled(self, tmp_path: pathlib.Path) -> None: """Records stored with ``branch`` instead of ``created_on_branch`` migrate.""" repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _v1_commit_id([], sid, "root", _AT_ISO) record = { "commit_id": old_id, "repo_id": _REPO_ID, "branch": "main", "snapshot_id": sid, "message": "root", "committed_at": _AT_ISO, "parent_commit_id": None, "parent2_commit_id": None, "author": _AUTHOR, "metadata": {}, "structured_delta": None, "sem_ver_bump": "none", "breaking_changes": [], "agent_id": "", "model_id": "", "toolchain_id": "", "prompt_hash": "", "signature": "", "signer_public_key": "", "signer_key_id": "", "reviewed_by": [], "test_runs": 0, "labels": [], "status": "", "notes": [], "score": None, } dest = repo / ".muse" / "commits" / "sha256" / f"{split_id(old_id)[1]}.msgpack" dest.write_bytes(msgpack.packb(record, use_bin_type=True)) _set_ref(repo, "main", old_id) _invoke(["code", "migrate"], repo) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() rec = read_commit(repo, new_head) assert rec is not None assert rec.created_on_branch == "main" # --------------------------------------------------------------------------- # JSON output contract # --------------------------------------------------------------------------- class TestJsonOutput: def test_json_output_has_required_keys(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) out = _invoke(["code", "migrate"], repo) for key in ("commits_rewritten", "commits_signed", "blobs_migrated", "id_map", "dry_run"): assert key in out, f"Missing key {key!r} in JSON output" def test_json_dry_run_flag_is_true(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) _set_ref(repo, "main", _write_v1_commit_raw(repo, "root", sid)) result = runner.invoke( cli, ["code", "migrate", "--dry-run", "--json"], env=_env(repo) ) out = json.loads(result.output) assert out["dry_run"] is True def test_json_live_run_dry_run_is_false(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) _set_ref(repo, "main", _write_v1_commit_raw(repo, "root", sid)) out = _invoke(["code", "migrate"], repo) assert out["dry_run"] is False def test_json_commits_signed_zero_when_unsigned(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) sid = _snap(repo) _set_ref(repo, "main", _write_v1_commit_raw(repo, "root", sid)) out = _invoke(["code", "migrate"], repo) assert out["commits_signed"] == 0 # --------------------------------------------------------------------------- # Re-signing (progressive enhancement — full provenance path) # --------------------------------------------------------------------------- def _generate_test_key() -> "Ed25519PrivateKey": from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey return Ed25519PrivateKey.generate() class TestResigning: def test_signed_migrate_produces_valid_signatures(self, tmp_path: pathlib.Path) -> None: from muse.core.migrate import migrate as _migrate from muse.core.provenance import verify_commit_ed25519, provenance_payload from muse.core.provenance import encode_public_key from muse.core._types import decode_pubkey repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) private_key = _generate_test_key() result = _migrate(repo, dry_run=False, private_key=private_key) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() rec = read_commit(repo, new_head) assert rec is not None assert rec.signature.startswith("ed25519:"), "signature must have ed25519: prefix" payload = provenance_payload( rec.commit_id, author=rec.author, agent_id=rec.agent_id, model_id=rec.model_id, toolchain_id=rec.toolchain_id, prompt_hash=rec.prompt_hash, committed_at=rec.committed_at.isoformat(), ) _, pub_bytes = decode_pubkey(rec.signer_public_key) assert verify_commit_ed25519(payload, rec.signature, pub_bytes), ( "Re-signed commit must verify against stored signer_public_key" ) def test_signed_migrate_all_commits_signed(self, tmp_path: pathlib.Path) -> None: from muse.core.migrate import migrate as _migrate repo = _init_repo(tmp_path) sid = _snap(repo) root_id = _write_v1_commit_raw(repo, "root", sid) sid2 = _snap(repo, "s2") child_id = _write_v1_commit_raw(repo, "child", sid2, parent=root_id) _set_ref(repo, "main", child_id) private_key = _generate_test_key() result = _migrate(repo, dry_run=False, private_key=private_key) assert result.commits_signed == 2, ( f"Expected 2 commits signed, got {result.commits_signed}" ) def test_signed_migrate_signer_public_key_bound_in_commit_id( self, tmp_path: pathlib.Path ) -> None: from muse.core.migrate import migrate as _migrate from muse.core.provenance import encode_public_key repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) private_key = _generate_test_key() result = _migrate(repo, dry_run=False, private_key=private_key) new_head = (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() rec = read_commit(repo, new_head) assert rec is not None _, expected_pubkey = encode_public_key(private_key) expected_id = compute_commit_id( parent_ids=[], snapshot_id=rec.snapshot_id, message=rec.message, committed_at_iso=rec.committed_at.isoformat(), repo_id=_REPO_ID, author=rec.author, signer_public_key=expected_pubkey, ) assert rec.commit_id == expected_id, ( "Commit ID must be computed with signer_public_key bound in" ) def test_dry_run_skips_signing(self, tmp_path: pathlib.Path) -> None: from muse.core.migrate import migrate as _migrate repo = _init_repo(tmp_path) sid = _snap(repo) old_id = _write_v1_commit_raw(repo, "root", sid) _set_ref(repo, "main", old_id) private_key = _generate_test_key() result = _migrate(repo, dry_run=True, private_key=private_key) assert result.commits_signed == 0, "dry-run must not sign anything" assert (repo / ".muse" / "refs" / "heads" / "main").read_text().strip() == old_id