test_proposal_snapshot_anchors.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """TDD: Snapshot anchors on merge proposals. |
| 2 | |
| 3 | When a proposal is created, the server captures the HEAD commit ID of each |
| 4 | branch at that moment and stores them as cryptographic anchors: |
| 5 | |
| 6 | from_snapshot_id — sha256:<hex> of from_branch HEAD at proposal creation time |
| 7 | to_snapshot_id — sha256:<hex> of to_branch HEAD at proposal creation time |
| 8 | |
| 9 | These are the "FROM STATE / TO STATE" anchors shown in the proposal detail UI. |
| 10 | They are nullable — a branch with no commits has no HEAD, so the anchor is null. |
| 11 | |
| 12 | Acceptance criteria |
| 13 | ------------------- |
| 14 | T1 POST /proposals stores from_snapshot_id and to_snapshot_id when both |
| 15 | branches have commits; GET returns both as fromSnapshotId / toSnapshotId. |
| 16 | T2 POST /proposals sets from_snapshot_id = null when from_branch has no HEAD. |
| 17 | T3 POST /proposals sets to_snapshot_id = null when to_branch has no HEAD. |
| 18 | T4 fromSnapshotId and toSnapshotId are present (possibly null) on every |
| 19 | ProposalResponse — the fields are never absent. |
| 20 | T5 Existing proposals created before this feature have null anchors — |
| 21 | backwards-compatible, no crash on GET. |
| 22 | T6 The stored from_snapshot_id matches the branch's head_commit_id at |
| 23 | creation time, not whatever the branch HEAD becomes later. |
| 24 | """ |
| 25 | from __future__ import annotations |
| 26 | |
| 27 | import pytest |
| 28 | from httpx import AsyncClient |
| 29 | from sqlalchemy.ext.asyncio import AsyncSession |
| 30 | |
| 31 | from musehub.db.musehub_repo_models import MusehubBranch |
| 32 | from musehub.db.musehub_social_models import MusehubProposal |
| 33 | from musehub.core.genesis import compute_branch_id |
| 34 | from musehub.types.json_types import StrDict |
| 35 | from sqlalchemy import select |
| 36 | |
| 37 | |
| 38 | # --------------------------------------------------------------------------- |
| 39 | # Helpers |
| 40 | # --------------------------------------------------------------------------- |
| 41 | |
| 42 | async def _create_repo(client: AsyncClient, auth_headers: StrDict, name: str) -> str: |
| 43 | r = await client.post( |
| 44 | "/api/repos", |
| 45 | json={"name": name, "owner": "testuser", "initialize": False}, |
| 46 | headers=auth_headers, |
| 47 | ) |
| 48 | assert r.status_code == 201 |
| 49 | return str(r.json()["repoId"]) |
| 50 | |
| 51 | |
| 52 | async def _push_branch( |
| 53 | db: AsyncSession, |
| 54 | repo_id: str, |
| 55 | branch_name: str, |
| 56 | head_commit_id: str | None = None, |
| 57 | ) -> None: |
| 58 | branch = MusehubBranch( |
| 59 | branch_id=compute_branch_id(repo_id, branch_name), |
| 60 | repo_id=repo_id, |
| 61 | name=branch_name, |
| 62 | head_commit_id=head_commit_id, |
| 63 | ) |
| 64 | db.add(branch) |
| 65 | await db.commit() |
| 66 | |
| 67 | |
| 68 | _COMMIT_A = "sha256:" + "a" * 64 |
| 69 | _COMMIT_B = "sha256:" + "b" * 64 |
| 70 | |
| 71 | |
| 72 | # --------------------------------------------------------------------------- |
| 73 | # T1 — both branches have commits → anchors stored and returned |
| 74 | # --------------------------------------------------------------------------- |
| 75 | |
| 76 | @pytest.mark.asyncio |
| 77 | async def test_snapshot_anchors_stored_when_both_branches_have_heads( |
| 78 | client: AsyncClient, |
| 79 | auth_headers: StrDict, |
| 80 | db_session: AsyncSession, |
| 81 | ) -> None: |
| 82 | repo_id = await _create_repo(client, auth_headers, "anchor-both-repo") |
| 83 | await _push_branch(db_session, repo_id, "feat/anchor", head_commit_id=_COMMIT_A) |
| 84 | await _push_branch(db_session, repo_id, "main", head_commit_id=_COMMIT_B) |
| 85 | |
| 86 | r = await client.post( |
| 87 | f"/api/repos/{repo_id}/proposals", |
| 88 | json={"title": "Anchor test", "fromBranch": "feat/anchor", "toBranch": "main"}, |
| 89 | headers=auth_headers, |
| 90 | ) |
| 91 | assert r.status_code == 201 |
| 92 | body = r.json() |
| 93 | assert body["fromSnapshotId"] == _COMMIT_A |
| 94 | assert body["toSnapshotId"] == _COMMIT_B |
| 95 | |
| 96 | |
| 97 | # --------------------------------------------------------------------------- |
| 98 | # T2 — from_branch has no HEAD → fromSnapshotId is null |
| 99 | # --------------------------------------------------------------------------- |
| 100 | |
| 101 | @pytest.mark.asyncio |
| 102 | async def test_from_snapshot_null_when_from_branch_empty( |
| 103 | client: AsyncClient, |
| 104 | auth_headers: StrDict, |
| 105 | db_session: AsyncSession, |
| 106 | ) -> None: |
| 107 | repo_id = await _create_repo(client, auth_headers, "anchor-empty-from-repo") |
| 108 | await _push_branch(db_session, repo_id, "feat/empty", head_commit_id=None) |
| 109 | await _push_branch(db_session, repo_id, "main", head_commit_id=_COMMIT_B) |
| 110 | |
| 111 | r = await client.post( |
| 112 | f"/api/repos/{repo_id}/proposals", |
| 113 | json={"title": "Empty from", "fromBranch": "feat/empty", "toBranch": "main"}, |
| 114 | headers=auth_headers, |
| 115 | ) |
| 116 | assert r.status_code == 201 |
| 117 | body = r.json() |
| 118 | assert body["fromSnapshotId"] is None |
| 119 | assert body["toSnapshotId"] == _COMMIT_B |
| 120 | |
| 121 | |
| 122 | # --------------------------------------------------------------------------- |
| 123 | # T3 — to_branch has no HEAD → toSnapshotId is null |
| 124 | # --------------------------------------------------------------------------- |
| 125 | |
| 126 | @pytest.mark.asyncio |
| 127 | async def test_to_snapshot_null_when_to_branch_empty( |
| 128 | client: AsyncClient, |
| 129 | auth_headers: StrDict, |
| 130 | db_session: AsyncSession, |
| 131 | ) -> None: |
| 132 | repo_id = await _create_repo(client, auth_headers, "anchor-empty-to-repo") |
| 133 | await _push_branch(db_session, repo_id, "feat/has-commits", head_commit_id=_COMMIT_A) |
| 134 | await _push_branch(db_session, repo_id, "main", head_commit_id=None) |
| 135 | |
| 136 | r = await client.post( |
| 137 | f"/api/repos/{repo_id}/proposals", |
| 138 | json={"title": "Empty to", "fromBranch": "feat/has-commits", "toBranch": "main"}, |
| 139 | headers=auth_headers, |
| 140 | ) |
| 141 | assert r.status_code == 201 |
| 142 | body = r.json() |
| 143 | assert body["fromSnapshotId"] == _COMMIT_A |
| 144 | assert body["toSnapshotId"] is None |
| 145 | |
| 146 | |
| 147 | # --------------------------------------------------------------------------- |
| 148 | # T4 — both fields always present in ProposalResponse (never absent) |
| 149 | # --------------------------------------------------------------------------- |
| 150 | |
| 151 | @pytest.mark.asyncio |
| 152 | async def test_snapshot_fields_always_present_in_response( |
| 153 | client: AsyncClient, |
| 154 | auth_headers: StrDict, |
| 155 | db_session: AsyncSession, |
| 156 | ) -> None: |
| 157 | repo_id = await _create_repo(client, auth_headers, "anchor-fields-repo") |
| 158 | await _push_branch(db_session, repo_id, "feat/fields", head_commit_id=None) |
| 159 | |
| 160 | r = await client.post( |
| 161 | f"/api/repos/{repo_id}/proposals", |
| 162 | json={"title": "Field presence", "fromBranch": "feat/fields", "toBranch": "main"}, |
| 163 | headers=auth_headers, |
| 164 | ) |
| 165 | assert r.status_code == 201 |
| 166 | body = r.json() |
| 167 | assert "fromSnapshotId" in body |
| 168 | assert "toSnapshotId" in body |
| 169 | |
| 170 | |
| 171 | # --------------------------------------------------------------------------- |
| 172 | # T5 — existing proposals (null anchors) don't crash on GET |
| 173 | # --------------------------------------------------------------------------- |
| 174 | |
| 175 | @pytest.mark.asyncio |
| 176 | async def test_existing_proposal_with_null_anchors_returns_ok( |
| 177 | client: AsyncClient, |
| 178 | auth_headers: StrDict, |
| 179 | db_session: AsyncSession, |
| 180 | ) -> None: |
| 181 | repo_id = await _create_repo(client, auth_headers, "anchor-legacy-repo") |
| 182 | await _push_branch(db_session, repo_id, "feat/legacy") |
| 183 | |
| 184 | # Create via API (will have anchors), then NULL them out to simulate legacy |
| 185 | r = await client.post( |
| 186 | f"/api/repos/{repo_id}/proposals", |
| 187 | json={"title": "Legacy proposal", "fromBranch": "feat/legacy", "toBranch": "main"}, |
| 188 | headers=auth_headers, |
| 189 | ) |
| 190 | assert r.status_code == 201 |
| 191 | proposal_id = r.json()["proposalId"] |
| 192 | |
| 193 | row = (await db_session.execute( |
| 194 | select(MusehubProposal).where(MusehubProposal.proposal_id == proposal_id) |
| 195 | )).scalar_one() |
| 196 | row.from_snapshot_id = None |
| 197 | row.to_snapshot_id = None |
| 198 | await db_session.commit() |
| 199 | |
| 200 | get_r = await client.get( |
| 201 | f"/api/repos/{repo_id}/proposals/{proposal_id}", |
| 202 | headers=auth_headers, |
| 203 | ) |
| 204 | assert get_r.status_code == 200 |
| 205 | body = get_r.json() |
| 206 | assert body["fromSnapshotId"] is None |
| 207 | assert body["toSnapshotId"] is None |
| 208 | |
| 209 | |
| 210 | # --------------------------------------------------------------------------- |
| 211 | # T6 — anchors are frozen at creation time, not updated when branch moves |
| 212 | # --------------------------------------------------------------------------- |
| 213 | |
| 214 | @pytest.mark.asyncio |
| 215 | async def test_snapshot_anchors_frozen_at_creation_time( |
| 216 | client: AsyncClient, |
| 217 | auth_headers: StrDict, |
| 218 | db_session: AsyncSession, |
| 219 | ) -> None: |
| 220 | repo_id = await _create_repo(client, auth_headers, "anchor-frozen-repo") |
| 221 | await _push_branch(db_session, repo_id, "feat/frozen", head_commit_id=_COMMIT_A) |
| 222 | await _push_branch(db_session, repo_id, "main", head_commit_id=_COMMIT_B) |
| 223 | |
| 224 | r = await client.post( |
| 225 | f"/api/repos/{repo_id}/proposals", |
| 226 | json={"title": "Frozen anchor", "fromBranch": "feat/frozen", "toBranch": "main"}, |
| 227 | headers=auth_headers, |
| 228 | ) |
| 229 | assert r.status_code == 201 |
| 230 | proposal_id = r.json()["proposalId"] |
| 231 | |
| 232 | # Advance the branch HEAD after proposal creation |
| 233 | _COMMIT_NEW = "sha256:" + "c" * 64 |
| 234 | branch_row = (await db_session.execute( |
| 235 | select(MusehubBranch).where( |
| 236 | MusehubBranch.repo_id == repo_id, |
| 237 | MusehubBranch.name == "feat/frozen", |
| 238 | ) |
| 239 | )).scalar_one() |
| 240 | branch_row.head_commit_id = _COMMIT_NEW |
| 241 | await db_session.commit() |
| 242 | |
| 243 | get_r = await client.get( |
| 244 | f"/api/repos/{repo_id}/proposals/{proposal_id}", |
| 245 | headers=auth_headers, |
| 246 | ) |
| 247 | assert get_r.status_code == 200 |
| 248 | body = get_r.json() |
| 249 | # Anchor must still reflect the HEAD at creation time |
| 250 | assert body["fromSnapshotId"] == _COMMIT_A |