"""TDD: Commit and snapshot anchors on merge proposals. When a proposal is created, the server captures the HEAD commit ID of each branch at that moment and stores them as cryptographic anchors, plus the Snapshot (manifest) ID that each commit points to: from_commit_id — sha256: of from_branch HEAD commit at proposal creation time to_commit_id — sha256: of to_branch HEAD commit at proposal creation time from_snapshot_id — sha256: Snapshot (manifest) ID that from_commit_id points to to_snapshot_id — sha256: Snapshot (manifest) ID that to_commit_id points to These are the "FROM STATE / TO STATE" anchors shown in the proposal detail UI (linked there by commit ID, not snapshot ID). musehub#144 review found that #0049 named the anchor columns from/to_snapshot_id but always populated them with a commit ID -- every row ever written held a commit ID, never a real Snapshot ID. Fixed by adding from/to_commit_id (holding what from/to_snapshot_id always actually held) and correcting from/to_snapshot_id to hold the real snapshot_id looked up from the anchor commit. Acceptance criteria ------------------- T1 POST /proposals stores from_commit_id and to_commit_id when both branches have commits; GET returns both as fromCommitId / toCommitId. T1b Snapshot lookup: when the anchor commit has a real snapshot_id, GET returns it as fromSnapshotId / toSnapshotId. T1c Snapshot lookup miss: when the anchor commit ID doesn't resolve to a real MusehubCommit row (e.g. a fixture/legacy commit), fromSnapshotId / toSnapshotId are null rather than falling back to the commit ID. T2 POST /proposals sets from_commit_id = null when from_branch has no HEAD. T3 POST /proposals sets to_commit_id = null when to_branch has no HEAD. T4 fromCommitId/toCommitId/fromSnapshotId/toSnapshotId are present (possibly null) on every ProposalResponse — the fields are never absent. T5 Existing proposals created before this feature have null anchors — backwards-compatible, no crash on GET. T6 The stored from_commit_id matches the branch's head_commit_id at creation time, not whatever the branch HEAD becomes later. """ from __future__ import annotations from datetime import datetime, timezone import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from musehub.db.musehub_repo_models import MusehubBranch, MusehubCommit from musehub.db.musehub_social_models import MusehubProposal from musehub.core.genesis import compute_branch_id from musehub.types.json_types import StrDict from sqlalchemy import select # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- async def _create_repo(client: AsyncClient, auth_headers: StrDict, name: str) -> str: r = await client.post( "/api/repos", json={"name": name, "owner": "testuser", "initialize": False}, headers=auth_headers, ) assert r.status_code == 201 return str(r.json()["repoId"]) async def _push_branch( db: AsyncSession, repo_id: str, branch_name: str, head_commit_id: str | None = None, ) -> None: """Insert a branch pointing at head_commit_id, without a backing commit row. Used for tests that only care about the commit-ID anchor, not the snapshot lookup (the commit ID is a bare fixture value, not a real MusehubCommit primary key). """ branch = MusehubBranch( branch_id=compute_branch_id(repo_id, branch_name), repo_id=repo_id, name=branch_name, head_commit_id=head_commit_id, ) db.add(branch) await db.commit() async def _push_branch_with_real_commit( db: AsyncSession, repo_id: str, branch_name: str, commit_id: str, snapshot_id: str, ) -> None: """Insert a branch AND a backing MusehubCommit row with a real snapshot_id.""" db.add(MusehubCommit( commit_id=commit_id, branch=branch_name, message="m", author="testuser", timestamp=datetime.now(tz=timezone.utc), snapshot_id=snapshot_id, )) db.add(MusehubBranch( branch_id=compute_branch_id(repo_id, branch_name), repo_id=repo_id, name=branch_name, head_commit_id=commit_id, )) await db.commit() _COMMIT_A = "sha256:" + "a" * 64 _COMMIT_B = "sha256:" + "b" * 64 _SNAPSHOT_A = "sha256:" + "1" * 64 _SNAPSHOT_B = "sha256:" + "2" * 64 # --------------------------------------------------------------------------- # T1 — both branches have commits → commit anchors stored and returned # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_commit_anchors_stored_when_both_branches_have_heads( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: repo_id = await _create_repo(client, auth_headers, "anchor-both-repo") await _push_branch(db_session, repo_id, "feat/anchor", head_commit_id=_COMMIT_A) await _push_branch(db_session, repo_id, "main", head_commit_id=_COMMIT_B) r = await client.post( f"/api/repos/{repo_id}/proposals", json={"title": "Anchor test", "fromBranch": "feat/anchor", "toBranch": "main"}, headers=auth_headers, ) assert r.status_code == 201 body = r.json() assert body["fromCommitId"] == _COMMIT_A assert body["toCommitId"] == _COMMIT_B # Neither fixture commit_id resolves to a real MusehubCommit row here. assert body["fromSnapshotId"] is None assert body["toSnapshotId"] is None # --------------------------------------------------------------------------- # T1b/T1c — snapshot lookup: hit when the anchor commit is real, miss otherwise # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_snapshot_ids_resolved_from_real_anchor_commits( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: repo_id = await _create_repo(client, auth_headers, "anchor-real-commit-repo") await _push_branch_with_real_commit( db_session, repo_id, "feat/real", commit_id=_COMMIT_A, snapshot_id=_SNAPSHOT_A, ) await _push_branch_with_real_commit( db_session, repo_id, "main", commit_id=_COMMIT_B, snapshot_id=_SNAPSHOT_B, ) r = await client.post( f"/api/repos/{repo_id}/proposals", json={"title": "Real anchor test", "fromBranch": "feat/real", "toBranch": "main"}, headers=auth_headers, ) assert r.status_code == 201 body = r.json() assert body["fromCommitId"] == _COMMIT_A assert body["toCommitId"] == _COMMIT_B assert body["fromSnapshotId"] == _SNAPSHOT_A assert body["toSnapshotId"] == _SNAPSHOT_B # --------------------------------------------------------------------------- # T2 — from_branch has no HEAD → fromCommitId is null # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_from_commit_null_when_from_branch_empty( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: repo_id = await _create_repo(client, auth_headers, "anchor-empty-from-repo") await _push_branch(db_session, repo_id, "feat/empty", head_commit_id=None) await _push_branch(db_session, repo_id, "main", head_commit_id=_COMMIT_B) r = await client.post( f"/api/repos/{repo_id}/proposals", json={"title": "Empty from", "fromBranch": "feat/empty", "toBranch": "main"}, headers=auth_headers, ) assert r.status_code == 201 body = r.json() assert body["fromCommitId"] is None assert body["fromSnapshotId"] is None assert body["toCommitId"] == _COMMIT_B # --------------------------------------------------------------------------- # T3 — to_branch has no HEAD → toCommitId is null # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_to_commit_null_when_to_branch_empty( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: repo_id = await _create_repo(client, auth_headers, "anchor-empty-to-repo") await _push_branch(db_session, repo_id, "feat/has-commits", head_commit_id=_COMMIT_A) await _push_branch(db_session, repo_id, "main", head_commit_id=None) r = await client.post( f"/api/repos/{repo_id}/proposals", json={"title": "Empty to", "fromBranch": "feat/has-commits", "toBranch": "main"}, headers=auth_headers, ) assert r.status_code == 201 body = r.json() assert body["fromCommitId"] == _COMMIT_A assert body["toCommitId"] is None assert body["toSnapshotId"] is None # --------------------------------------------------------------------------- # T4 — all four fields always present in ProposalResponse (never absent) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_anchor_fields_always_present_in_response( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: repo_id = await _create_repo(client, auth_headers, "anchor-fields-repo") await _push_branch(db_session, repo_id, "feat/fields", head_commit_id=None) r = await client.post( f"/api/repos/{repo_id}/proposals", json={"title": "Field presence", "fromBranch": "feat/fields", "toBranch": "main"}, headers=auth_headers, ) assert r.status_code == 201 body = r.json() assert "fromCommitId" in body assert "toCommitId" in body assert "fromSnapshotId" in body assert "toSnapshotId" in body # --------------------------------------------------------------------------- # T5 — existing proposals (null anchors) don't crash on GET # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_existing_proposal_with_null_anchors_returns_ok( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: repo_id = await _create_repo(client, auth_headers, "anchor-legacy-repo") await _push_branch(db_session, repo_id, "feat/legacy") # Create via API (will have anchors), then NULL them out to simulate legacy r = await client.post( f"/api/repos/{repo_id}/proposals", json={"title": "Legacy proposal", "fromBranch": "feat/legacy", "toBranch": "main"}, headers=auth_headers, ) assert r.status_code == 201 proposal_id = r.json()["proposalId"] row = (await db_session.execute( select(MusehubProposal).where(MusehubProposal.proposal_id == proposal_id) )).scalar_one() row.from_commit_id = None row.to_commit_id = None row.from_snapshot_id = None row.to_snapshot_id = None await db_session.commit() get_r = await client.get( f"/api/repos/{repo_id}/proposals/{proposal_id}", headers=auth_headers, ) assert get_r.status_code == 200 body = get_r.json() assert body["fromCommitId"] is None assert body["toCommitId"] is None assert body["fromSnapshotId"] is None assert body["toSnapshotId"] is None # --------------------------------------------------------------------------- # T6 — anchors are frozen at creation time, not updated when branch moves # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_commit_anchors_frozen_at_creation_time( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: repo_id = await _create_repo(client, auth_headers, "anchor-frozen-repo") await _push_branch(db_session, repo_id, "feat/frozen", head_commit_id=_COMMIT_A) await _push_branch(db_session, repo_id, "main", head_commit_id=_COMMIT_B) r = await client.post( f"/api/repos/{repo_id}/proposals", json={"title": "Frozen anchor", "fromBranch": "feat/frozen", "toBranch": "main"}, headers=auth_headers, ) assert r.status_code == 201 proposal_id = r.json()["proposalId"] # Advance the branch HEAD after proposal creation _COMMIT_NEW = "sha256:" + "c" * 64 branch_row = (await db_session.execute( select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == "feat/frozen", ) )).scalar_one() branch_row.head_commit_id = _COMMIT_NEW await db_session.commit() get_r = await client.get( f"/api/repos/{repo_id}/proposals/{proposal_id}", headers=auth_headers, ) assert get_r.status_code == 200 body = get_r.json() # Anchor must still reflect the HEAD at creation time assert body["fromCommitId"] == _COMMIT_A