test_musehub_stash.py
python
| 1 | """Tests for Muse Hub stash endpoints (musehub/stash.py). |
| 2 | |
| 3 | Covers all 6 endpoints introduced in PR #467: |
| 4 | - list_stash: GET /repos/{repo_id}/stash (paginated, user-scoped) |
| 5 | - push_stash: POST /repos/{repo_id}/stash |
| 6 | - get_stash: GET /repos/{repo_id}/stash/{stash_id} |
| 7 | - pop_stash: POST /repos/{repo_id}/stash/{stash_id}/pop |
| 8 | - apply_stash: POST /repos/{repo_id}/stash/{stash_id}/apply |
| 9 | - drop_stash: DELETE /repos/{repo_id}/stash/{stash_id} |
| 10 | |
| 11 | Key invariants asserted: |
| 12 | - Stash entries are user-scoped: user A cannot see user B's stash |
| 13 | - pop removes the stash row atomically (deleted=True in response) |
| 14 | - apply leaves the stash row intact (deleted=False in response) |
| 15 | - 404 is returned for stash_id not owned by caller |
| 16 | - Pagination works: total and page fields are correct |
| 17 | - All write endpoints require auth (401 without token) |
| 18 | """ |
| 19 | from __future__ import annotations |
| 20 | |
| 21 | import uuid |
| 22 | |
| 23 | import pytest |
| 24 | from httpx import AsyncClient |
| 25 | from sqlalchemy.ext.asyncio import AsyncSession |
| 26 | |
| 27 | _TEST_REPO_ID = str(uuid.uuid4()) |
| 28 | _BASE = f"/api/v1/musehub/repos/{_TEST_REPO_ID}/stash" |
| 29 | |
| 30 | _PUSH_BODY = { |
| 31 | "message": "WIP: bridge section", |
| 32 | "branch": "feat/bridge", |
| 33 | "entries": [ |
| 34 | {"path": "tracks/piano.mid", "object_id": "sha256:aabbcc"}, |
| 35 | {"path": "tracks/bass.mid", "object_id": "sha256:ddeeff"}, |
| 36 | ], |
| 37 | } |
| 38 | |
| 39 | |
| 40 | # --------------------------------------------------------------------------- |
| 41 | # push_stash — POST /repos/{repo_id}/stash |
| 42 | # --------------------------------------------------------------------------- |
| 43 | |
| 44 | |
| 45 | @pytest.mark.anyio |
| 46 | async def test_push_stash_creates_stash_with_entries( |
| 47 | client: AsyncClient, auth_headers: dict[str, str] |
| 48 | ) -> None: |
| 49 | """Push creates a stash record and returns it with its entries.""" |
| 50 | resp = await client.post(_BASE, json=_PUSH_BODY, headers=auth_headers) |
| 51 | assert resp.status_code == 201, resp.text |
| 52 | data = resp.json() |
| 53 | assert data["branch"] == "feat/bridge" |
| 54 | assert data["message"] == "WIP: bridge section" |
| 55 | assert len(data["entries"]) == 2 |
| 56 | paths = {e["path"] for e in data["entries"]} |
| 57 | assert paths == {"tracks/piano.mid", "tracks/bass.mid"} |
| 58 | |
| 59 | |
| 60 | @pytest.mark.anyio |
| 61 | async def test_push_stash_requires_auth(client: AsyncClient) -> None: |
| 62 | """Pushing a stash without a token returns 401.""" |
| 63 | resp = await client.post(_BASE, json=_PUSH_BODY) |
| 64 | assert resp.status_code == 401 |
| 65 | |
| 66 | |
| 67 | @pytest.mark.anyio |
| 68 | async def test_push_stash_empty_entries( |
| 69 | client: AsyncClient, auth_headers: dict[str, str] |
| 70 | ) -> None: |
| 71 | """Push with no entries creates a stash with an empty entries list.""" |
| 72 | body = {"message": "empty stash", "branch": "main", "entries": []} |
| 73 | resp = await client.post(_BASE, json=body, headers=auth_headers) |
| 74 | assert resp.status_code == 201 |
| 75 | assert resp.json()["entries"] == [] |
| 76 | |
| 77 | |
| 78 | # --------------------------------------------------------------------------- |
| 79 | # list_stash — GET /repos/{repo_id}/stash |
| 80 | # --------------------------------------------------------------------------- |
| 81 | |
| 82 | |
| 83 | @pytest.mark.anyio |
| 84 | async def test_list_stash_returns_only_caller_entries( |
| 85 | client: AsyncClient, auth_headers: dict[str, str] |
| 86 | ) -> None: |
| 87 | """List returns a paginated result with total and page metadata.""" |
| 88 | await client.post(_BASE, json=_PUSH_BODY, headers=auth_headers) |
| 89 | await client.post( |
| 90 | _BASE, json={**_PUSH_BODY, "message": "stash 2"}, headers=auth_headers |
| 91 | ) |
| 92 | |
| 93 | resp = await client.get(_BASE, headers=auth_headers) |
| 94 | assert resp.status_code == 200 |
| 95 | data = resp.json() |
| 96 | assert data["total"] == 2 |
| 97 | assert data["page"] == 1 |
| 98 | assert len(data["items"]) == 2 |
| 99 | |
| 100 | |
| 101 | @pytest.mark.anyio |
| 102 | async def test_list_stash_pagination( |
| 103 | client: AsyncClient, auth_headers: dict[str, str] |
| 104 | ) -> None: |
| 105 | """Pagination parameters are respected: page_size limits results.""" |
| 106 | for i in range(3): |
| 107 | await client.post( |
| 108 | _BASE, json={**_PUSH_BODY, "message": f"stash {i}"}, headers=auth_headers |
| 109 | ) |
| 110 | |
| 111 | resp = await client.get(_BASE, params={"page": 1, "page_size": 2}, headers=auth_headers) |
| 112 | assert resp.status_code == 200 |
| 113 | data = resp.json() |
| 114 | assert data["total"] == 3 |
| 115 | assert len(data["items"]) == 2 |
| 116 | assert data["page_size"] == 2 |
| 117 | |
| 118 | |
| 119 | @pytest.mark.anyio |
| 120 | async def test_list_stash_requires_auth(client: AsyncClient) -> None: |
| 121 | """Listing stash without a token returns 401.""" |
| 122 | resp = await client.get(_BASE) |
| 123 | assert resp.status_code == 401 |
| 124 | |
| 125 | |
| 126 | # --------------------------------------------------------------------------- |
| 127 | # get_stash — GET /repos/{repo_id}/stash/{stash_id} |
| 128 | # --------------------------------------------------------------------------- |
| 129 | |
| 130 | |
| 131 | @pytest.mark.anyio |
| 132 | async def test_get_stash_returns_detail_with_entries( |
| 133 | client: AsyncClient, auth_headers: dict[str, str] |
| 134 | ) -> None: |
| 135 | """get_stash returns the stash row along with its file entries.""" |
| 136 | push_resp = await client.post(_BASE, json=_PUSH_BODY, headers=auth_headers) |
| 137 | stash_id = push_resp.json()["id"] |
| 138 | |
| 139 | resp = await client.get(f"{_BASE}/{stash_id}", headers=auth_headers) |
| 140 | assert resp.status_code == 200 |
| 141 | data = resp.json() |
| 142 | assert data["id"] == stash_id |
| 143 | assert len(data["entries"]) == 2 |
| 144 | |
| 145 | |
| 146 | @pytest.mark.anyio |
| 147 | async def test_get_stash_404_for_unknown_id( |
| 148 | client: AsyncClient, auth_headers: dict[str, str] |
| 149 | ) -> None: |
| 150 | """get_stash returns 404 for a stash_id that does not exist.""" |
| 151 | resp = await client.get(f"{_BASE}/{uuid.uuid4()}", headers=auth_headers) |
| 152 | assert resp.status_code == 404 |
| 153 | |
| 154 | |
| 155 | @pytest.mark.anyio |
| 156 | async def test_get_stash_requires_auth(client: AsyncClient) -> None: |
| 157 | """get_stash without a token returns 401.""" |
| 158 | resp = await client.get(f"{_BASE}/{uuid.uuid4()}") |
| 159 | assert resp.status_code == 401 |
| 160 | |
| 161 | |
| 162 | # --------------------------------------------------------------------------- |
| 163 | # pop_stash — POST /repos/{repo_id}/stash/{stash_id}/pop |
| 164 | # --------------------------------------------------------------------------- |
| 165 | |
| 166 | |
| 167 | @pytest.mark.anyio |
| 168 | async def test_pop_stash_returns_entries_and_deletes_stash( |
| 169 | client: AsyncClient, auth_headers: dict[str, str] |
| 170 | ) -> None: |
| 171 | """pop returns the stash entries and removes the stash (deleted=True).""" |
| 172 | push_resp = await client.post(_BASE, json=_PUSH_BODY, headers=auth_headers) |
| 173 | stash_id = push_resp.json()["id"] |
| 174 | |
| 175 | pop_resp = await client.post(f"{_BASE}/{stash_id}/pop", headers=auth_headers) |
| 176 | assert pop_resp.status_code == 200 |
| 177 | data = pop_resp.json() |
| 178 | assert data["deleted"] is True |
| 179 | assert len(data["entries"]) == 2 |
| 180 | |
| 181 | # Stash should be gone now |
| 182 | get_resp = await client.get(f"{_BASE}/{stash_id}", headers=auth_headers) |
| 183 | assert get_resp.status_code == 404 |
| 184 | |
| 185 | |
| 186 | @pytest.mark.anyio |
| 187 | async def test_pop_stash_404_for_unknown_id( |
| 188 | client: AsyncClient, auth_headers: dict[str, str] |
| 189 | ) -> None: |
| 190 | """pop returns 404 for a stash_id not owned by caller.""" |
| 191 | resp = await client.post(f"{_BASE}/{uuid.uuid4()}/pop", headers=auth_headers) |
| 192 | assert resp.status_code == 404 |
| 193 | |
| 194 | |
| 195 | @pytest.mark.anyio |
| 196 | async def test_pop_stash_requires_auth(client: AsyncClient) -> None: |
| 197 | """pop without a token returns 401.""" |
| 198 | resp = await client.post(f"{_BASE}/{uuid.uuid4()}/pop") |
| 199 | assert resp.status_code == 401 |
| 200 | |
| 201 | |
| 202 | # --------------------------------------------------------------------------- |
| 203 | # apply_stash — POST /repos/{repo_id}/stash/{stash_id}/apply |
| 204 | # --------------------------------------------------------------------------- |
| 205 | |
| 206 | |
| 207 | @pytest.mark.anyio |
| 208 | async def test_apply_stash_returns_entries_and_keeps_stash( |
| 209 | client: AsyncClient, auth_headers: dict[str, str] |
| 210 | ) -> None: |
| 211 | """apply returns stash entries (deleted=False) and leaves the stash intact.""" |
| 212 | push_resp = await client.post(_BASE, json=_PUSH_BODY, headers=auth_headers) |
| 213 | stash_id = push_resp.json()["id"] |
| 214 | |
| 215 | apply_resp = await client.post(f"{_BASE}/{stash_id}/apply", headers=auth_headers) |
| 216 | assert apply_resp.status_code == 200 |
| 217 | data = apply_resp.json() |
| 218 | assert data["deleted"] is False |
| 219 | assert len(data["entries"]) == 2 |
| 220 | |
| 221 | # Stash should still exist |
| 222 | get_resp = await client.get(f"{_BASE}/{stash_id}", headers=auth_headers) |
| 223 | assert get_resp.status_code == 200 |
| 224 | |
| 225 | |
| 226 | @pytest.mark.anyio |
| 227 | async def test_apply_stash_404_for_unknown_id( |
| 228 | client: AsyncClient, auth_headers: dict[str, str] |
| 229 | ) -> None: |
| 230 | """apply returns 404 for a stash_id not owned by caller.""" |
| 231 | resp = await client.post(f"{_BASE}/{uuid.uuid4()}/apply", headers=auth_headers) |
| 232 | assert resp.status_code == 404 |
| 233 | |
| 234 | |
| 235 | @pytest.mark.anyio |
| 236 | async def test_apply_stash_requires_auth(client: AsyncClient) -> None: |
| 237 | """apply without a token returns 401.""" |
| 238 | resp = await client.post(f"{_BASE}/{uuid.uuid4()}/apply") |
| 239 | assert resp.status_code == 401 |
| 240 | |
| 241 | |
| 242 | # --------------------------------------------------------------------------- |
| 243 | # drop_stash — DELETE /repos/{repo_id}/stash/{stash_id} |
| 244 | # --------------------------------------------------------------------------- |
| 245 | |
| 246 | |
| 247 | @pytest.mark.anyio |
| 248 | async def test_drop_stash_deletes_stash_without_applying( |
| 249 | client: AsyncClient, auth_headers: dict[str, str] |
| 250 | ) -> None: |
| 251 | """drop permanently removes the stash entry (204 No Content).""" |
| 252 | push_resp = await client.post(_BASE, json=_PUSH_BODY, headers=auth_headers) |
| 253 | stash_id = push_resp.json()["id"] |
| 254 | |
| 255 | drop_resp = await client.delete(f"{_BASE}/{stash_id}", headers=auth_headers) |
| 256 | assert drop_resp.status_code == 204 |
| 257 | |
| 258 | get_resp = await client.get(f"{_BASE}/{stash_id}", headers=auth_headers) |
| 259 | assert get_resp.status_code == 404 |
| 260 | |
| 261 | |
| 262 | @pytest.mark.anyio |
| 263 | async def test_drop_stash_404_for_unknown_id( |
| 264 | client: AsyncClient, auth_headers: dict[str, str] |
| 265 | ) -> None: |
| 266 | """drop returns 404 for a stash_id not owned by caller.""" |
| 267 | resp = await client.delete(f"{_BASE}/{uuid.uuid4()}", headers=auth_headers) |
| 268 | assert resp.status_code == 404 |
| 269 | |
| 270 | |
| 271 | @pytest.mark.anyio |
| 272 | async def test_drop_stash_requires_auth(client: AsyncClient) -> None: |
| 273 | """drop without a token returns 401.""" |
| 274 | resp = await client.delete(f"{_BASE}/{uuid.uuid4()}") |
| 275 | assert resp.status_code == 401 |
| 276 | |
| 277 | |
| 278 | # --------------------------------------------------------------------------- |
| 279 | # User isolation — a user cannot see another user's stash |
| 280 | # --------------------------------------------------------------------------- |
| 281 | |
| 282 | |
| 283 | @pytest.mark.anyio |
| 284 | async def test_stash_is_user_scoped( |
| 285 | client: AsyncClient, |
| 286 | db_session: AsyncSession, |
| 287 | auth_headers: dict[str, str], |
| 288 | ) -> None: |
| 289 | """A user cannot access another user's stash by guessing the stash_id.""" |
| 290 | push_resp = await client.post(_BASE, json=_PUSH_BODY, headers=auth_headers) |
| 291 | stash_id = push_resp.json()["id"] |
| 292 | |
| 293 | # Create a second user and token |
| 294 | from musehub.auth.tokens import create_access_token |
| 295 | from musehub.db.models import User |
| 296 | |
| 297 | other_user = User(id=str(uuid.uuid4())) |
| 298 | db_session.add(other_user) |
| 299 | await db_session.commit() |
| 300 | other_token = create_access_token(user_id=other_user.id, expires_hours=1) |
| 301 | other_headers = {"Authorization": f"Bearer {other_token}", "Content-Type": "application/json"} |
| 302 | |
| 303 | # Other user cannot see the stash |
| 304 | resp = await client.get(f"{_BASE}/{stash_id}", headers=other_headers) |
| 305 | assert resp.status_code == 404 |
| 306 | |
| 307 | # Other user's list is empty |
| 308 | list_resp = await client.get(_BASE, headers=other_headers) |
| 309 | assert list_resp.json()["total"] == 0 |