gabriel / musehub public
test_musehub_ui_stash.py python
557 lines 17.5 KB
Raw
c0f0b481 release: merge dev → main (#5) Gabriel Cardona <cgcardona@gmail.com> 5d ago
1 """Tests for MuseHub stash UI endpoints.
2
3 Covers GET /{owner}/{repo_slug}/stash:
4 - test_stash_list_page_auth_required — unauthenticated GET → 401
5 - test_stash_list_page_returns_200_with_token — authenticated GET → 200 HTML
6 - test_stash_list_page_shows_ref_labels — HTML includes stash@{0} refs
7 - test_stash_list_page_action_buttons_present — Apply / Pop / Drop buttons present
8 - test_stash_list_page_drop_confirm_present — Drop form has hx-confirm attribute
9 - test_stash_list_page_json_response — ?format=json returns JSON with stashes key
10 - test_stash_list_page_json_fields — JSON stash items have required fields
11 - test_stash_list_page_empty_stash — empty stash returns 200 with 0 total
12 - test_stash_list_unknown_repo_404 — unknown owner/slug → 404
13 - test_stash_list_isolates_by_user — only caller's stash is shown
14 - test_stash_list_pagination_query_params — page/page_size accepted without error
15
16 Covers POST /{owner}/{repo_slug}/stash/{stash_ref}/apply:
17 - test_stash_apply_auth_required — unauthenticated POST → 401
18 - test_stash_apply_redirects_to_stash_list — authenticated POST → 303 redirect
19 - test_stash_apply_preserves_stash_entry — stash entry still exists after apply
20
21 Covers POST /{owner}/{repo_slug}/stash/{stash_ref}/pop:
22 - test_stash_pop_auth_required — unauthenticated POST → 401
23 - test_stash_pop_redirects_to_stash_list — authenticated POST → 303 redirect
24 - test_stash_pop_deletes_stash_entry — stash entry removed after pop
25
26 Covers POST /{owner}/{repo_slug}/stash/{stash_ref}/drop:
27 - test_stash_drop_auth_required — unauthenticated POST → 401
28 - test_stash_drop_redirects_to_stash_list — authenticated POST → 303 redirect
29 - test_stash_drop_deletes_stash_entry — stash entry removed after drop
30 - test_stash_drop_wrong_user_404 — another user's stash → 404
31 """
32 from __future__ import annotations
33
34 import uuid
35
36 import pytest
37 from httpx import AsyncClient
38 from sqlalchemy import text
39 from sqlalchemy.ext.asyncio import AsyncSession
40
41 from musehub.db.musehub_models import MusehubRepo
42 from musehub.db.musehub_stash_models import MusehubStash, MusehubStashEntry
43
44 _OWNER = "artist"
45 _SLUG = "album-one"
46 _USER_ID = "550e8400-e29b-41d4-a716-446655440000" # matches test_user fixture
47
48
49 # ---------------------------------------------------------------------------
50 # Helpers
51 # ---------------------------------------------------------------------------
52
53
54 async def _make_repo(db: AsyncSession, owner: str = _OWNER, slug: str = _SLUG) -> str:
55 """Seed a public repo and return its repo_id string."""
56 repo = MusehubRepo(
57 name=slug,
58 owner=owner,
59 slug=slug,
60 visibility="public",
61 owner_user_id=_USER_ID,
62 )
63 db.add(repo)
64 await db.commit()
65 await db.refresh(repo)
66 return str(repo.repo_id)
67
68
69 async def _make_stash(
70 db: AsyncSession,
71 repo_id: str,
72 *,
73 user_id: str = _USER_ID,
74 branch: str = "main",
75 message: str | None = "WIP: brass arrangement",
76 num_entries: int = 2,
77 ) -> MusehubStash:
78 """Seed a stash entry with ``num_entries`` file entries and return it."""
79 stash = MusehubStash(
80 repo_id=repo_id,
81 user_id=user_id,
82 branch=branch,
83 message=message,
84 )
85 db.add(stash)
86 await db.flush()
87
88 for i in range(num_entries):
89 entry = MusehubStashEntry(
90 stash_id=stash.id,
91 path=f"tracks/track_{i}.mid",
92 object_id=f"sha256:{'a' * 64}",
93 position=i,
94 )
95 db.add(entry)
96
97 await db.commit()
98 await db.refresh(stash)
99 return stash
100
101
102 # ---------------------------------------------------------------------------
103 # GET — stash list page
104 # ---------------------------------------------------------------------------
105
106
107 @pytest.mark.anyio
108 async def test_stash_list_page_auth_required(
109 client: AsyncClient,
110 db_session: AsyncSession,
111 ) -> None:
112 """Unauthenticated GET returns 401 — stash is always private."""
113 await _make_repo(db_session)
114 response = await client.get(f"/{_OWNER}/{_SLUG}/stash")
115 assert response.status_code == 401
116
117
118 @pytest.mark.anyio
119 async def test_stash_list_page_returns_200_with_token(
120 client: AsyncClient,
121 db_session: AsyncSession,
122 auth_headers: dict[str, str],
123 test_user: object,
124 ) -> None:
125 """Authenticated GET returns 200 HTML."""
126 await _make_repo(db_session)
127 response = await client.get(
128 f"/{_OWNER}/{_SLUG}/stash",
129 headers=auth_headers,
130 )
131 assert response.status_code == 200
132 assert "text/html" in response.headers["content-type"]
133
134
135 @pytest.mark.anyio
136 async def test_stash_list_page_shows_ref_labels(
137 client: AsyncClient,
138 db_session: AsyncSession,
139 auth_headers: dict[str, str],
140 test_user: object,
141 ) -> None:
142 """HTML page contains stash@{0} ref label for the first stash entry."""
143 repo_id = await _make_repo(db_session)
144 await _make_stash(db_session, repo_id)
145 response = await client.get(
146 f"/{_OWNER}/{_SLUG}/stash",
147 headers=auth_headers,
148 )
149 assert response.status_code == 200
150 assert "stash@{0}" in response.text
151
152
153 @pytest.mark.anyio
154 async def test_stash_list_page_action_buttons_present(
155 client: AsyncClient,
156 db_session: AsyncSession,
157 auth_headers: dict[str, str],
158 test_user: object,
159 ) -> None:
160 """HTML page exposes Apply, Pop, and Drop action buttons."""
161 repo_id = await _make_repo(db_session)
162 await _make_stash(db_session, repo_id)
163 response = await client.get(
164 f"/{_OWNER}/{_SLUG}/stash",
165 headers=auth_headers,
166 )
167 assert response.status_code == 200
168 body = response.text
169 assert "Apply" in body
170 assert "Pop" in body
171 assert "Drop" in body
172
173
174 @pytest.mark.anyio
175 async def test_stash_list_page_drop_confirm_present(
176 client: AsyncClient,
177 db_session: AsyncSession,
178 auth_headers: dict[str, str],
179 test_user: object,
180 ) -> None:
181 """Drop form uses hx-confirm attribute for HTMX-native confirmation dialog."""
182 repo_id = await _make_repo(db_session)
183 await _make_stash(db_session, repo_id)
184 response = await client.get(
185 f"/{_OWNER}/{_SLUG}/stash",
186 headers=auth_headers,
187 )
188 assert response.status_code == 200
189 assert "hx-confirm" in response.text
190
191
192 @pytest.mark.anyio
193 async def test_stash_list_page_json_response(
194 client: AsyncClient,
195 db_session: AsyncSession,
196 auth_headers: dict[str, str],
197 test_user: object,
198 ) -> None:
199 """?format=json returns JSON with HTTP 200."""
200 repo_id = await _make_repo(db_session)
201 await _make_stash(db_session, repo_id)
202 headers = {**auth_headers, "Content-Type": "application/json"}
203 response = await client.get(
204 f"/{_OWNER}/{_SLUG}/stash?format=json",
205 headers=headers,
206 )
207 assert response.status_code == 200
208 assert response.headers["content-type"].startswith("application/json")
209
210
211 @pytest.mark.anyio
212 async def test_stash_list_page_json_fields(
213 client: AsyncClient,
214 db_session: AsyncSession,
215 auth_headers: dict[str, str],
216 test_user: object,
217 ) -> None:
218 """JSON stash items include ref, branch, message, createdAt, entryCount."""
219 repo_id = await _make_repo(db_session)
220 await _make_stash(db_session, repo_id, branch="feat/bass", message="WIP brass", num_entries=3)
221 response = await client.get(
222 f"/{_OWNER}/{_SLUG}/stash?format=json",
223 headers=auth_headers,
224 )
225 assert response.status_code == 200
226 data = response.json()
227 assert "stashes" in data
228 assert "total" in data
229 assert data["total"] == 1
230 item = data["stashes"][0]
231 assert item["ref"] == "stash@{0}"
232 assert item["branch"] == "feat/bass"
233 assert item["message"] == "WIP brass"
234 assert item["entryCount"] == 3
235 assert "createdAt" in item
236
237
238 @pytest.mark.anyio
239 async def test_stash_list_page_empty_stash(
240 client: AsyncClient,
241 db_session: AsyncSession,
242 auth_headers: dict[str, str],
243 test_user: object,
244 ) -> None:
245 """Stash list page with no entries returns 200 and total=0."""
246 await _make_repo(db_session)
247 response = await client.get(
248 f"/{_OWNER}/{_SLUG}/stash?format=json",
249 headers=auth_headers,
250 )
251 assert response.status_code == 200
252 data = response.json()
253 assert data["total"] == 0
254 assert data["stashes"] == []
255
256
257 @pytest.mark.anyio
258 async def test_stash_list_unknown_repo_404(
259 client: AsyncClient,
260 db_session: AsyncSession,
261 auth_headers: dict[str, str],
262 test_user: object,
263 ) -> None:
264 """Unknown owner/slug returns 404."""
265 response = await client.get(
266 "/nobody/nonexistent/stash",
267 headers=auth_headers,
268 )
269 assert response.status_code == 404
270
271
272 @pytest.mark.anyio
273 async def test_stash_list_isolates_by_user(
274 client: AsyncClient,
275 db_session: AsyncSession,
276 auth_headers: dict[str, str],
277 test_user: object,
278 ) -> None:
279 """Stash list only returns entries owned by the authenticated user."""
280 repo_id = await _make_repo(db_session)
281 other_user_id = str(uuid.uuid4())
282 # Create a stash for a different (non-existent for FK purposes) user using raw SQL
283 # to avoid FK violation — we only care about the scoping logic, not DB integrity here.
284 await db_session.execute(
285 text(
286 "INSERT INTO musehub_stash (id, repo_id, user_id, branch, message, is_applied, created_at) "
287 "VALUES (:id, :repo_id, :user_id, :branch, :message, false, datetime('now'))"
288 ),
289 {
290 "id": str(uuid.uuid4()),
291 "repo_id": repo_id,
292 "user_id": other_user_id,
293 "branch": "main",
294 "message": "someone else's stash",
295 },
296 )
297 await db_session.commit()
298
299 response = await client.get(
300 f"/{_OWNER}/{_SLUG}/stash?format=json",
301 headers=auth_headers,
302 )
303 assert response.status_code == 200
304 data = response.json()
305 # Our test user has no stash entries — the other user's stash is invisible
306 assert data["total"] == 0
307
308
309 @pytest.mark.anyio
310 async def test_stash_list_pagination_query_params(
311 client: AsyncClient,
312 db_session: AsyncSession,
313 auth_headers: dict[str, str],
314 test_user: object,
315 ) -> None:
316 """page and page_size query params are accepted without error."""
317 await _make_repo(db_session)
318 response = await client.get(
319 f"/{_OWNER}/{_SLUG}/stash?page=1&page_size=10",
320 headers=auth_headers,
321 )
322 assert response.status_code == 200
323
324
325 # ---------------------------------------------------------------------------
326 # POST — apply stash
327 # ---------------------------------------------------------------------------
328
329
330 @pytest.mark.anyio
331 async def test_stash_apply_auth_required(
332 client: AsyncClient,
333 db_session: AsyncSession,
334 ) -> None:
335 """Unauthenticated POST to /apply returns 401."""
336 repo_id = await _make_repo(db_session)
337 stash = await _make_stash(db_session, repo_id)
338 response = await client.post(
339 f"/{_OWNER}/{_SLUG}/stash/{stash.id}/apply",
340 follow_redirects=False,
341 )
342 assert response.status_code == 401
343
344
345 @pytest.mark.anyio
346 async def test_stash_apply_redirects_to_stash_list(
347 client: AsyncClient,
348 db_session: AsyncSession,
349 auth_headers: dict[str, str],
350 test_user: object,
351 ) -> None:
352 """Authenticated POST to /apply returns 303 redirect to the stash list."""
353 repo_id = await _make_repo(db_session)
354 stash = await _make_stash(db_session, repo_id)
355 response = await client.post(
356 f"/{_OWNER}/{_SLUG}/stash/{stash.id}/apply",
357 headers=auth_headers,
358 follow_redirects=False,
359 )
360 assert response.status_code == 303
361 assert response.headers["location"] == f"/{_OWNER}/{_SLUG}/stash"
362
363
364 @pytest.mark.anyio
365 async def test_stash_apply_preserves_stash_entry(
366 client: AsyncClient,
367 db_session: AsyncSession,
368 auth_headers: dict[str, str],
369 test_user: object,
370 ) -> None:
371 """Apply does NOT delete the stash entry — it stays on the stack."""
372 repo_id = await _make_repo(db_session)
373 stash = await _make_stash(db_session, repo_id)
374 stash_id = stash.id
375
376 await client.post(
377 f"/{_OWNER}/{_SLUG}/stash/{stash_id}/apply",
378 headers=auth_headers,
379 follow_redirects=False,
380 )
381
382 # Verify the stash entry still exists
383 result = await db_session.execute(
384 text("SELECT id FROM musehub_stash WHERE id = :id"),
385 {"id": stash_id},
386 )
387 row = result.mappings().first()
388 assert row is not None, "Stash entry should NOT be deleted after apply"
389
390
391 # ---------------------------------------------------------------------------
392 # POST — pop stash
393 # ---------------------------------------------------------------------------
394
395
396 @pytest.mark.anyio
397 async def test_stash_pop_auth_required(
398 client: AsyncClient,
399 db_session: AsyncSession,
400 ) -> None:
401 """Unauthenticated POST to /pop returns 401."""
402 repo_id = await _make_repo(db_session)
403 stash = await _make_stash(db_session, repo_id)
404 response = await client.post(
405 f"/{_OWNER}/{_SLUG}/stash/{stash.id}/pop",
406 follow_redirects=False,
407 )
408 assert response.status_code == 401
409
410
411 @pytest.mark.anyio
412 async def test_stash_pop_redirects_to_stash_list(
413 client: AsyncClient,
414 db_session: AsyncSession,
415 auth_headers: dict[str, str],
416 test_user: object,
417 ) -> None:
418 """Authenticated POST to /pop returns 303 redirect to the stash list."""
419 repo_id = await _make_repo(db_session)
420 stash = await _make_stash(db_session, repo_id)
421 response = await client.post(
422 f"/{_OWNER}/{_SLUG}/stash/{stash.id}/pop",
423 headers=auth_headers,
424 follow_redirects=False,
425 )
426 assert response.status_code == 303
427 assert response.headers["location"] == f"/{_OWNER}/{_SLUG}/stash"
428
429
430 @pytest.mark.anyio
431 async def test_stash_pop_deletes_stash_entry(
432 client: AsyncClient,
433 db_session: AsyncSession,
434 auth_headers: dict[str, str],
435 test_user: object,
436 ) -> None:
437 """Pop deletes the stash entry from the stack."""
438 repo_id = await _make_repo(db_session)
439 stash = await _make_stash(db_session, repo_id)
440 stash_id = stash.id
441
442 await client.post(
443 f"/{_OWNER}/{_SLUG}/stash/{stash_id}/pop",
444 headers=auth_headers,
445 follow_redirects=False,
446 )
447
448 # Verify the stash entry is gone
449 db_session.expire_all()
450 result = await db_session.execute(
451 text("SELECT id FROM musehub_stash WHERE id = :id"),
452 {"id": stash_id},
453 )
454 row = result.mappings().first()
455 assert row is None, "Stash entry SHOULD be deleted after pop"
456
457
458 # ---------------------------------------------------------------------------
459 # POST — drop stash
460 # ---------------------------------------------------------------------------
461
462
463 @pytest.mark.anyio
464 async def test_stash_drop_auth_required(
465 client: AsyncClient,
466 db_session: AsyncSession,
467 ) -> None:
468 """Unauthenticated POST to /drop returns 401."""
469 repo_id = await _make_repo(db_session)
470 stash = await _make_stash(db_session, repo_id)
471 response = await client.post(
472 f"/{_OWNER}/{_SLUG}/stash/{stash.id}/drop",
473 follow_redirects=False,
474 )
475 assert response.status_code == 401
476
477
478 @pytest.mark.anyio
479 async def test_stash_drop_redirects_to_stash_list(
480 client: AsyncClient,
481 db_session: AsyncSession,
482 auth_headers: dict[str, str],
483 test_user: object,
484 ) -> None:
485 """Authenticated POST to /drop returns 303 redirect to the stash list."""
486 repo_id = await _make_repo(db_session)
487 stash = await _make_stash(db_session, repo_id)
488 response = await client.post(
489 f"/{_OWNER}/{_SLUG}/stash/{stash.id}/drop",
490 headers=auth_headers,
491 follow_redirects=False,
492 )
493 assert response.status_code == 303
494 assert response.headers["location"] == f"/{_OWNER}/{_SLUG}/stash"
495
496
497 @pytest.mark.anyio
498 async def test_stash_drop_deletes_stash_entry(
499 client: AsyncClient,
500 db_session: AsyncSession,
501 auth_headers: dict[str, str],
502 test_user: object,
503 ) -> None:
504 """Drop permanently deletes the stash entry without applying it."""
505 repo_id = await _make_repo(db_session)
506 stash = await _make_stash(db_session, repo_id)
507 stash_id = stash.id
508
509 await client.post(
510 f"/{_OWNER}/{_SLUG}/stash/{stash_id}/drop",
511 headers=auth_headers,
512 follow_redirects=False,
513 )
514
515 db_session.expire_all()
516 result = await db_session.execute(
517 text("SELECT id FROM musehub_stash WHERE id = :id"),
518 {"id": stash_id},
519 )
520 row = result.mappings().first()
521 assert row is None, "Stash entry SHOULD be deleted after drop"
522
523
524 @pytest.mark.anyio
525 async def test_stash_drop_wrong_user_404(
526 client: AsyncClient,
527 db_session: AsyncSession,
528 auth_headers: dict[str, str],
529 test_user: object,
530 ) -> None:
531 """Attempting to drop another user's stash returns 404."""
532 repo_id = await _make_repo(db_session)
533 other_stash_id = str(uuid.uuid4())
534 other_user_id = str(uuid.uuid4())
535
536 # Insert a stash owned by a different user using raw SQL
537 await db_session.execute(
538 text(
539 "INSERT INTO musehub_stash (id, repo_id, user_id, branch, message, is_applied, created_at) "
540 "VALUES (:id, :repo_id, :user_id, :branch, :message, false, datetime('now'))"
541 ),
542 {
543 "id": other_stash_id,
544 "repo_id": repo_id,
545 "user_id": other_user_id,
546 "branch": "main",
547 "message": "other user stash",
548 },
549 )
550 await db_session.commit()
551
552 response = await client.post(
553 f"/{_OWNER}/{_SLUG}/stash/{other_stash_id}/drop",
554 headers=auth_headers,
555 follow_redirects=False,
556 )
557 assert response.status_code == 404