"""TDD: 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: from_snapshot_id — sha256: of from_branch HEAD at proposal creation time to_snapshot_id — sha256: of to_branch HEAD at proposal creation time These are the "FROM STATE / TO STATE" anchors shown in the proposal detail UI. They are nullable — a branch with no commits has no HEAD, so the anchor is null. Acceptance criteria ------------------- T1 POST /proposals stores from_snapshot_id and to_snapshot_id when both branches have commits; GET returns both as fromSnapshotId / toSnapshotId. T2 POST /proposals sets from_snapshot_id = null when from_branch has no HEAD. T3 POST /proposals sets to_snapshot_id = null when to_branch has no HEAD. T4 fromSnapshotId and 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_snapshot_id matches the branch's head_commit_id at creation time, not whatever the branch HEAD becomes later. """ from __future__ import annotations import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from musehub.db.musehub_repo_models import MusehubBranch 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: 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() _COMMIT_A = "sha256:" + "a" * 64 _COMMIT_B = "sha256:" + "b" * 64 # --------------------------------------------------------------------------- # T1 — both branches have commits → anchors stored and returned # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_snapshot_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["fromSnapshotId"] == _COMMIT_A assert body["toSnapshotId"] == _COMMIT_B # --------------------------------------------------------------------------- # T2 — from_branch has no HEAD → fromSnapshotId is null # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_from_snapshot_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["fromSnapshotId"] is None assert body["toSnapshotId"] == _COMMIT_B # --------------------------------------------------------------------------- # T3 — to_branch has no HEAD → toSnapshotId is null # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_to_snapshot_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["fromSnapshotId"] == _COMMIT_A assert body["toSnapshotId"] is None # --------------------------------------------------------------------------- # T4 — both fields always present in ProposalResponse (never absent) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_snapshot_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 "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_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["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_snapshot_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["fromSnapshotId"] == _COMMIT_A