gabriel / musehub public
test_collaborators_section27.py python
947 lines 32.9 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Section 27 — Collaborators & Permissions: 7-layer test suite.
2
3 Covers:
4 - musehub/api/routes/musehub/collaborators.py (CRUD + permission logic)
5 - musehub/db/musehub_collaborator_models.py (ORM model)
6 - repos.py _guard_admin / check_collaborator_access (permission gating)
7
8 Endpoints:
9 GET /api/repos/{repo_id}/collaborators
10 POST /api/repos/{repo_id}/collaborators
11 PUT /api/repos/{repo_id}/collaborators/{handle}/permission
12 DELETE /api/repos/{repo_id}/collaborators/{handle}
13 GET /api/repos/{repo_id}/collaborators/{username}/permission
14
15 Layer map
16 ---------
17 1. Unit — Permission enum, _PERMISSION_RANK, _has_permission, _orm_to_response
18 2. Integration — DB-level collaborator CRUD via session
19 3. E2E — HTTP client against full app
20 4. Stress — 50 collaborators, concurrent list calls
21 5. Data Integrity — permission stored correctly, invited_by set, unique constraint
22 6. Security — auth required, non-admin blocked, owner un-removable
23 7. Performance — timing budgets
24 """
25 from __future__ import annotations
26
27 import asyncio
28 import time
29 import uuid
30 from datetime import datetime, timezone
31
32 import pytest
33 from httpx import AsyncClient
34 from sqlalchemy import select
35 from sqlalchemy.ext.asyncio import AsyncSession
36
37 from musehub.muse_contracts.json_types import StrDict
38 from musehub.api.routes.musehub.collaborators import (
39 Permission,
40 _PERMISSION_RANK,
41 _has_permission,
42 _orm_to_response,
43 )
44 from musehub.db.musehub_collaborator_models import MusehubCollaborator
45 from musehub.db.musehub_models import MusehubRepo
46
47
48 # ---------------------------------------------------------------------------
49 # Fixtures / helpers
50 # ---------------------------------------------------------------------------
51
52 _TEST_HANDLE = "testuser" # matches auth_headers fixture's token.handle
53
54
55 def _uid() -> str:
56 return str(uuid.uuid4())
57
58
59 async def _db_repo(
60 session: AsyncSession,
61 owner: str = _TEST_HANDLE,
62 *,
63 visibility: str = "private",
64 ) -> MusehubRepo:
65 slug = f"repo-{_uid()[:8]}"
66 repo = MusehubRepo(
67 repo_id=_uid(),
68 name=slug,
69 slug=slug,
70 owner=owner,
71 owner_user_id=owner,
72 visibility=visibility,
73 )
74 session.add(repo)
75 await session.flush()
76 return repo
77
78
79 async def _db_collab(
80 session: AsyncSession,
81 repo_id: str,
82 handle: str,
83 *,
84 permission: str = "write",
85 invited_by: str | None = None,
86 accepted: bool = False,
87 ) -> MusehubCollaborator:
88 c = MusehubCollaborator(
89 id=_uid(),
90 repo_id=repo_id,
91 identity_handle=handle,
92 permission=permission,
93 invited_by_handle=invited_by,
94 accepted_at=datetime.now(timezone.utc) if accepted else None,
95 )
96 session.add(c)
97 await session.flush()
98 return c
99
100
101 async def _api_repo(
102 client: AsyncClient,
103 auth_headers: StrDict,
104 *,
105 visibility: str = "private",
106 ) -> str:
107 r = await client.post(
108 "/api/repos",
109 json={"name": f"collab-{_uid()[:8]}", "owner": _TEST_HANDLE, "visibility": visibility},
110 headers=auth_headers,
111 )
112 assert r.status_code == 201, r.text
113 return r.json()["repoId"]
114
115
116 # ===========================================================================
117 # Layer 1 — Unit
118 # ===========================================================================
119
120
121 class TestUnitPermissionEnum:
122 def test_values(self) -> None:
123 assert Permission.read == "read"
124 assert Permission.write == "write"
125 assert Permission.admin == "admin"
126 assert Permission.owner == "owner"
127
128 def test_four_levels(self) -> None:
129 assert len(list(Permission)) == 4
130
131
132 class TestUnitPermissionRank:
133 def test_read_is_lowest(self) -> None:
134 assert _PERMISSION_RANK["read"] < _PERMISSION_RANK["write"]
135
136 def test_write_lt_admin(self) -> None:
137 assert _PERMISSION_RANK["write"] < _PERMISSION_RANK["admin"]
138
139 def test_admin_lt_owner(self) -> None:
140 assert _PERMISSION_RANK["admin"] < _PERMISSION_RANK["owner"]
141
142 def test_all_levels_covered(self) -> None:
143 for p in Permission:
144 assert p.value in _PERMISSION_RANK
145
146
147 class TestUnitHasPermission:
148 def test_exact_match(self) -> None:
149 assert _has_permission("write", Permission.write) is True
150
151 def test_higher_grants_lower(self) -> None:
152 assert _has_permission("admin", Permission.write) is True
153 assert _has_permission("owner", Permission.read) is True
154
155 def test_lower_denied_higher(self) -> None:
156 assert _has_permission("read", Permission.write) is False
157 assert _has_permission("write", Permission.admin) is False
158
159 def test_unknown_permission_denied(self) -> None:
160 assert _has_permission("", Permission.read) is False
161 assert _has_permission("superuser", Permission.read) is False
162
163 def test_read_satisfies_read(self) -> None:
164 assert _has_permission("read", Permission.read) is True
165
166 def test_owner_satisfies_admin(self) -> None:
167 assert _has_permission("owner", Permission.admin) is True
168
169
170 class TestUnitOrmToResponse:
171 @pytest.mark.anyio
172 async def test_fields_mapped_correctly(self, db_session: AsyncSession) -> None:
173 repo = await _db_repo(db_session)
174 collab = await _db_collab(
175 db_session, repo.repo_id, "alice",
176 permission="write", invited_by="bob"
177 )
178 resp = _orm_to_response(collab)
179 assert resp.handle == "alice"
180 assert resp.permission == "write"
181 assert resp.invited_by == "bob"
182 assert resp.repo_id == repo.repo_id
183 assert resp.collaborator_id == collab.id
184
185 @pytest.mark.anyio
186 async def test_invited_by_none_when_null(self, db_session: AsyncSession) -> None:
187 repo = await _db_repo(db_session)
188 collab = await _db_collab(db_session, repo.repo_id, "carol", invited_by=None)
189 resp = _orm_to_response(collab)
190 assert resp.invited_by is None
191
192
193 # ===========================================================================
194 # Layer 2 — Integration (DB-level)
195 # ===========================================================================
196
197
198 class TestIntegrationCollaboratorDB:
199 @pytest.mark.anyio
200 async def test_insert_and_query(self, db_session: AsyncSession) -> None:
201 repo = await _db_repo(db_session)
202 collab = await _db_collab(db_session, repo.repo_id, "alice", permission="admin")
203 await db_session.flush()
204
205 result = await db_session.execute(
206 select(MusehubCollaborator).where(
207 MusehubCollaborator.repo_id == repo.repo_id
208 )
209 )
210 rows = result.scalars().all()
211 assert len(rows) == 1
212 assert rows[0].identity_handle == "alice"
213 assert rows[0].permission == "admin"
214
215 @pytest.mark.anyio
216 async def test_unique_constraint_on_repo_handle(
217 self, db_session: AsyncSession
218 ) -> None:
219 from sqlalchemy.exc import IntegrityError
220
221 repo = await _db_repo(db_session)
222 await _db_collab(db_session, repo.repo_id, "alice")
223 await db_session.flush()
224
225 dup = MusehubCollaborator(
226 id=_uid(),
227 repo_id=repo.repo_id,
228 identity_handle="alice",
229 permission="read",
230 )
231 db_session.add(dup)
232 with pytest.raises(IntegrityError):
233 await db_session.flush()
234
235 @pytest.mark.anyio
236 async def test_delete_collaborator_directly(self, db_session: AsyncSession) -> None:
237 # Verify that a collaborator can be deleted explicitly and is gone afterwards.
238 repo = await _db_repo(db_session)
239 collab = await _db_collab(db_session, repo.repo_id, "alice")
240 await db_session.commit()
241
242 await db_session.delete(collab)
243 await db_session.commit()
244
245 result = await db_session.execute(
246 select(MusehubCollaborator).where(
247 MusehubCollaborator.repo_id == repo.repo_id
248 )
249 )
250 assert result.scalars().first() is None
251
252 @pytest.mark.anyio
253 async def test_accepted_at_null_by_default(self, db_session: AsyncSession) -> None:
254 repo = await _db_repo(db_session)
255 collab = await _db_collab(db_session, repo.repo_id, "dave")
256 assert collab.accepted_at is None
257
258 @pytest.mark.anyio
259 async def test_permission_default_write(self, db_session: AsyncSession) -> None:
260 repo = await _db_repo(db_session)
261 collab = MusehubCollaborator(
262 id=_uid(),
263 repo_id=repo.repo_id,
264 identity_handle="eve",
265 )
266 db_session.add(collab)
267 await db_session.flush()
268 assert collab.permission == "write"
269
270
271 # ===========================================================================
272 # Layer 3 — E2E
273 # ===========================================================================
274
275
276 class TestE2EListCollaborators:
277 @pytest.mark.anyio
278 async def test_list_returns_200(
279 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
280 ) -> None:
281 repo_id = await _api_repo(client, auth_headers)
282 await _db_collab(db_session, repo_id, "alice")
283 await db_session.commit()
284
285 r = await client.get(f"/api/repos/{repo_id}/collaborators", headers=auth_headers)
286 assert r.status_code == 200
287 body = r.json()
288 assert "collaborators" in body
289 assert "total" in body
290 assert body["total"] == 1
291
292 @pytest.mark.anyio
293 async def test_list_requires_auth(
294 self, client: AsyncClient, db_session: AsyncSession
295 ) -> None:
296 repo = await _db_repo(db_session)
297 await db_session.commit()
298
299 r = await client.get(f"/api/repos/{repo.repo_id}/collaborators")
300 assert r.status_code == 401
301
302 @pytest.mark.anyio
303 async def test_list_unknown_repo_404(
304 self, client: AsyncClient, auth_headers: StrDict
305 ) -> None:
306 r = await client.get("/api/repos/no-such-repo/collaborators", headers=auth_headers)
307 assert r.status_code == 404
308
309 @pytest.mark.anyio
310 async def test_list_empty_repo(
311 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
312 ) -> None:
313 repo_id = await _api_repo(client, auth_headers)
314 await db_session.commit()
315
316 r = await client.get(f"/api/repos/{repo_id}/collaborators", headers=auth_headers)
317 assert r.status_code == 200
318 assert r.json()["total"] == 0
319
320
321 class TestE2EInviteCollaborator:
322 @pytest.mark.anyio
323 async def test_owner_can_invite_201(
324 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
325 ) -> None:
326 repo_id = await _api_repo(client, auth_headers)
327 await db_session.commit()
328
329 r = await client.post(
330 f"/api/repos/{repo_id}/collaborators",
331 json={"handle": "alice", "permission": "write"},
332 headers=auth_headers,
333 )
334 assert r.status_code == 201
335 body = r.json()
336 assert body["handle"] == "alice"
337 assert body["permission"] == "write"
338 assert body["invitedBy"] == _TEST_HANDLE
339
340 @pytest.mark.anyio
341 async def test_invite_requires_auth(
342 self, client: AsyncClient, db_session: AsyncSession
343 ) -> None:
344 repo = await _db_repo(db_session)
345 await db_session.commit()
346
347 r = await client.post(
348 f"/api/repos/{repo.repo_id}/collaborators",
349 json={"handle": "bob", "permission": "read"},
350 )
351 assert r.status_code == 401
352
353 @pytest.mark.anyio
354 async def test_non_admin_gets_403(
355 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
356 ) -> None:
357 """testuser is not owner (alice owns repo) and has only 'write' — gets 403."""
358 repo = await _db_repo(db_session, owner="alice")
359 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="write")
360 await db_session.commit()
361
362 r = await client.post(
363 f"/api/repos/{repo.repo_id}/collaborators",
364 json={"handle": "bob", "permission": "read"},
365 headers=auth_headers,
366 )
367 assert r.status_code == 403
368
369 @pytest.mark.anyio
370 async def test_admin_collab_can_invite(
371 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
372 ) -> None:
373 """testuser has admin permission → can invite."""
374 repo = await _db_repo(db_session, owner="alice")
375 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="admin")
376 await db_session.commit()
377
378 r = await client.post(
379 f"/api/repos/{repo.repo_id}/collaborators",
380 json={"handle": "bob", "permission": "read"},
381 headers=auth_headers,
382 )
383 assert r.status_code == 201
384
385 @pytest.mark.anyio
386 async def test_duplicate_invite_409(
387 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
388 ) -> None:
389 repo_id = await _api_repo(client, auth_headers)
390 await db_session.commit()
391
392 body = {"handle": "alice", "permission": "write"}
393 r1 = await client.post(
394 f"/api/repos/{repo_id}/collaborators", json=body, headers=auth_headers
395 )
396 assert r1.status_code == 201
397
398 r2 = await client.post(
399 f"/api/repos/{repo_id}/collaborators", json=body, headers=auth_headers
400 )
401 assert r2.status_code == 409
402 assert "already a collaborator" in r2.json()["detail"]
403
404 @pytest.mark.anyio
405 async def test_invite_unknown_repo_404(
406 self, client: AsyncClient, auth_headers: StrDict
407 ) -> None:
408 r = await client.post(
409 "/api/repos/no-such-repo/collaborators",
410 json={"handle": "alice", "permission": "write"},
411 headers=auth_headers,
412 )
413 assert r.status_code == 404
414
415 @pytest.mark.anyio
416 async def test_default_permission_write(
417 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
418 ) -> None:
419 repo_id = await _api_repo(client, auth_headers)
420 await db_session.commit()
421
422 r = await client.post(
423 f"/api/repos/{repo_id}/collaborators",
424 json={"handle": "alice"}, # no permission field → defaults to write
425 headers=auth_headers,
426 )
427 assert r.status_code == 201
428 assert r.json()["permission"] == "write"
429
430
431 class TestE2EUpdatePermission:
432 @pytest.mark.anyio
433 async def test_owner_can_update_200(
434 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
435 ) -> None:
436 repo_id = await _api_repo(client, auth_headers)
437 await _db_collab(db_session, repo_id, "alice", permission="read")
438 await db_session.commit()
439
440 r = await client.put(
441 f"/api/repos/{repo_id}/collaborators/alice/permission",
442 json={"permission": "admin"},
443 headers=auth_headers,
444 )
445 assert r.status_code == 200
446 assert r.json()["permission"] == "admin"
447
448 @pytest.mark.anyio
449 async def test_update_requires_auth(
450 self, client: AsyncClient, db_session: AsyncSession
451 ) -> None:
452 repo = await _db_repo(db_session)
453 await _db_collab(db_session, repo.repo_id, "alice")
454 await db_session.commit()
455
456 r = await client.put(
457 f"/api/repos/{repo.repo_id}/collaborators/alice/permission",
458 json={"permission": "admin"},
459 )
460 assert r.status_code == 401
461
462 @pytest.mark.anyio
463 async def test_non_admin_gets_403(
464 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
465 ) -> None:
466 repo = await _db_repo(db_session, owner="alice")
467 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="write")
468 await _db_collab(db_session, repo.repo_id, "bob", permission="read")
469 await db_session.commit()
470
471 r = await client.put(
472 f"/api/repos/{repo.repo_id}/collaborators/bob/permission",
473 json={"permission": "admin"},
474 headers=auth_headers,
475 )
476 assert r.status_code == 403
477
478 @pytest.mark.anyio
479 async def test_update_owner_permission_403(
480 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
481 ) -> None:
482 """Cannot change owner's permission via this endpoint."""
483 repo = await _db_repo(db_session, owner="alice")
484 # testuser has admin permission
485 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="admin")
486 # alice has 'owner' permission in collaborators table
487 await _db_collab(db_session, repo.repo_id, "alice", permission="owner")
488 await db_session.commit()
489
490 r = await client.put(
491 f"/api/repos/{repo.repo_id}/collaborators/alice/permission",
492 json={"permission": "write"},
493 headers=auth_headers,
494 )
495 assert r.status_code == 403
496 assert "Owner permission" in r.json()["detail"]
497
498 @pytest.mark.anyio
499 async def test_update_nonexistent_collab_404(
500 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
501 ) -> None:
502 repo_id = await _api_repo(client, auth_headers)
503 await db_session.commit()
504
505 r = await client.put(
506 f"/api/repos/{repo_id}/collaborators/nobody/permission",
507 json={"permission": "read"},
508 headers=auth_headers,
509 )
510 assert r.status_code == 404
511
512
513 class TestE2ERemoveCollaborator:
514 @pytest.mark.anyio
515 async def test_owner_can_remove_204(
516 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
517 ) -> None:
518 repo_id = await _api_repo(client, auth_headers)
519 await _db_collab(db_session, repo_id, "alice")
520 await db_session.commit()
521
522 r = await client.delete(
523 f"/api/repos/{repo_id}/collaborators/alice", headers=auth_headers
524 )
525 assert r.status_code == 204
526
527 @pytest.mark.anyio
528 async def test_remove_requires_auth(
529 self, client: AsyncClient, db_session: AsyncSession
530 ) -> None:
531 repo = await _db_repo(db_session)
532 await _db_collab(db_session, repo.repo_id, "alice")
533 await db_session.commit()
534
535 r = await client.delete(
536 f"/api/repos/{repo.repo_id}/collaborators/alice"
537 )
538 assert r.status_code == 401
539
540 @pytest.mark.anyio
541 async def test_non_admin_gets_403(
542 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
543 ) -> None:
544 repo = await _db_repo(db_session, owner="alice")
545 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="write")
546 await _db_collab(db_session, repo.repo_id, "bob", permission="read")
547 await db_session.commit()
548
549 r = await client.delete(
550 f"/api/repos/{repo.repo_id}/collaborators/bob", headers=auth_headers
551 )
552 assert r.status_code == 403
553
554 @pytest.mark.anyio
555 async def test_remove_owner_403(
556 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
557 ) -> None:
558 """Owner-permission collaborator cannot be removed."""
559 repo = await _db_repo(db_session, owner="alice")
560 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="admin")
561 await _db_collab(db_session, repo.repo_id, "alice", permission="owner")
562 await db_session.commit()
563
564 r = await client.delete(
565 f"/api/repos/{repo.repo_id}/collaborators/alice", headers=auth_headers
566 )
567 assert r.status_code == 403
568 assert "Owner cannot be removed" in r.json()["detail"]
569
570 @pytest.mark.anyio
571 async def test_remove_nonexistent_404(
572 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
573 ) -> None:
574 repo_id = await _api_repo(client, auth_headers)
575 await db_session.commit()
576
577 r = await client.delete(
578 f"/api/repos/{repo_id}/collaborators/nobody", headers=auth_headers
579 )
580 assert r.status_code == 404
581
582
583 class TestE2ECheckAccess:
584 @pytest.mark.anyio
585 async def test_owner_access_is_owner_permission(
586 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
587 ) -> None:
588 repo_id = await _api_repo(client, auth_headers)
589 await db_session.commit()
590
591 # testuser is the owner; check their own permission
592 r = await client.get(
593 f"/api/repos/{repo_id}/collaborators/{_TEST_HANDLE}/permission",
594 headers=auth_headers,
595 )
596 assert r.status_code == 200
597 body = r.json()
598 assert body["permission"] == "owner"
599
600 @pytest.mark.anyio
601 async def test_collab_access_returns_permission(
602 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
603 ) -> None:
604 repo_id = await _api_repo(client, auth_headers)
605 await _db_collab(db_session, repo_id, "alice", permission="admin")
606 await db_session.commit()
607
608 r = await client.get(
609 f"/api/repos/{repo_id}/collaborators/alice/permission",
610 headers=auth_headers,
611 )
612 assert r.status_code == 200
613 assert r.json()["permission"] == "admin"
614
615 @pytest.mark.anyio
616 async def test_non_collab_404(
617 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
618 ) -> None:
619 repo_id = await _api_repo(client, auth_headers)
620 await db_session.commit()
621
622 r = await client.get(
623 f"/api/repos/{repo_id}/collaborators/stranger/permission",
624 headers=auth_headers,
625 )
626 assert r.status_code == 404
627
628 @pytest.mark.anyio
629 async def test_check_requires_auth(
630 self, client: AsyncClient, db_session: AsyncSession
631 ) -> None:
632 repo = await _db_repo(db_session)
633 await db_session.commit()
634
635 r = await client.get(
636 f"/api/repos/{repo.repo_id}/collaborators/{_TEST_HANDLE}/permission"
637 )
638 assert r.status_code == 401
639
640
641 # ===========================================================================
642 # Layer 4 — Stress
643 # ===========================================================================
644
645
646 class TestStress:
647 @pytest.mark.anyio
648 async def test_list_50_collaborators(
649 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
650 ) -> None:
651 repo_id = await _api_repo(client, auth_headers)
652 for i in range(50):
653 await _db_collab(db_session, repo_id, f"user{i}", permission="read")
654 await db_session.commit()
655
656 r = await client.get(f"/api/repos/{repo_id}/collaborators", headers=auth_headers)
657 assert r.status_code == 200
658 assert r.json()["total"] == 50
659
660 @pytest.mark.anyio
661 async def test_5_concurrent_list_calls(
662 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
663 ) -> None:
664 repo_id = await _api_repo(client, auth_headers)
665 for i in range(10):
666 await _db_collab(db_session, repo_id, f"stress{i}")
667 await db_session.commit()
668
669 responses = await asyncio.gather(
670 *[
671 client.get(
672 f"/api/repos/{repo_id}/collaborators", headers=auth_headers
673 )
674 for _ in range(5)
675 ]
676 )
677 assert all(r.status_code == 200 for r in responses)
678 assert all(r.json()["total"] == 10 for r in responses)
679
680
681 # ===========================================================================
682 # Layer 5 — Data Integrity
683 # ===========================================================================
684
685
686 class TestDataIntegrity:
687 @pytest.mark.anyio
688 async def test_invited_by_set_correctly(
689 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
690 ) -> None:
691 repo_id = await _api_repo(client, auth_headers)
692 await db_session.commit()
693
694 r = await client.post(
695 f"/api/repos/{repo_id}/collaborators",
696 json={"handle": "alice", "permission": "read"},
697 headers=auth_headers,
698 )
699 assert r.status_code == 201
700 assert r.json()["invitedBy"] == _TEST_HANDLE
701
702 @pytest.mark.anyio
703 async def test_permission_persisted_correctly(
704 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
705 ) -> None:
706 repo_id = await _api_repo(client, auth_headers)
707 await db_session.commit()
708
709 await client.post(
710 f"/api/repos/{repo_id}/collaborators",
711 json={"handle": "alice", "permission": "admin"},
712 headers=auth_headers,
713 )
714 db_session.expire_all()
715
716 row = (
717 await db_session.execute(
718 select(MusehubCollaborator).where(
719 MusehubCollaborator.repo_id == repo_id,
720 MusehubCollaborator.identity_handle == "alice",
721 )
722 )
723 ).scalar_one_or_none()
724 assert row is not None
725 assert row.permission == "admin"
726
727 @pytest.mark.anyio
728 async def test_update_persisted_in_db(
729 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
730 ) -> None:
731 repo_id = await _api_repo(client, auth_headers)
732 await _db_collab(db_session, repo_id, "alice", permission="read")
733 await db_session.commit()
734
735 await client.put(
736 f"/api/repos/{repo_id}/collaborators/alice/permission",
737 json={"permission": "admin"},
738 headers=auth_headers,
739 )
740 db_session.expire_all()
741
742 row = (
743 await db_session.execute(
744 select(MusehubCollaborator).where(
745 MusehubCollaborator.repo_id == repo_id,
746 MusehubCollaborator.identity_handle == "alice",
747 )
748 )
749 ).scalar_one_or_none()
750 assert row is not None
751 assert row.permission == "admin"
752
753 @pytest.mark.anyio
754 async def test_remove_deletes_db_row(
755 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
756 ) -> None:
757 repo_id = await _api_repo(client, auth_headers)
758 await _db_collab(db_session, repo_id, "alice")
759 await db_session.commit()
760
761 await client.delete(
762 f"/api/repos/{repo_id}/collaborators/alice", headers=auth_headers
763 )
764 db_session.expire_all()
765
766 row = (
767 await db_session.execute(
768 select(MusehubCollaborator).where(
769 MusehubCollaborator.repo_id == repo_id,
770 MusehubCollaborator.identity_handle == "alice",
771 )
772 )
773 ).scalar_one_or_none()
774 assert row is None
775
776 @pytest.mark.anyio
777 async def test_response_total_matches_actual_count(
778 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
779 ) -> None:
780 repo_id = await _api_repo(client, auth_headers)
781 for i in range(7):
782 await _db_collab(db_session, repo_id, f"u{i}")
783 await db_session.commit()
784
785 r = await client.get(f"/api/repos/{repo_id}/collaborators", headers=auth_headers)
786 body = r.json()
787 assert body["total"] == len(body["collaborators"])
788
789
790 # ===========================================================================
791 # Layer 6 — Security
792 # ===========================================================================
793
794
795 class TestSecurity:
796 @pytest.mark.anyio
797 async def test_all_endpoints_require_auth(
798 self, client: AsyncClient, db_session: AsyncSession
799 ) -> None:
800 repo = await _db_repo(db_session)
801 await _db_collab(db_session, repo.repo_id, "alice")
802 await db_session.commit()
803
804 endpoints = [
805 ("GET", f"/api/repos/{repo.repo_id}/collaborators"),
806 ("POST", f"/api/repos/{repo.repo_id}/collaborators"),
807 ("PUT", f"/api/repos/{repo.repo_id}/collaborators/alice/permission"),
808 ("DELETE", f"/api/repos/{repo.repo_id}/collaborators/alice"),
809 ]
810 for method, url in endpoints:
811 r = await client.request(method, url, json={"handle": "x", "permission": "read"})
812 assert r.status_code == 401, f"{method} {url} should require auth"
813
814 @pytest.mark.anyio
815 async def test_read_only_collab_cannot_invite(
816 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
817 ) -> None:
818 repo = await _db_repo(db_session, owner="alice")
819 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="read")
820 await db_session.commit()
821
822 r = await client.post(
823 f"/api/repos/{repo.repo_id}/collaborators",
824 json={"handle": "bob", "permission": "read"},
825 headers=auth_headers,
826 )
827 assert r.status_code == 403
828
829 @pytest.mark.anyio
830 async def test_write_collab_cannot_remove(
831 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
832 ) -> None:
833 repo = await _db_repo(db_session, owner="alice")
834 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="write")
835 await _db_collab(db_session, repo.repo_id, "bob", permission="read")
836 await db_session.commit()
837
838 r = await client.delete(
839 f"/api/repos/{repo.repo_id}/collaborators/bob", headers=auth_headers
840 )
841 assert r.status_code == 403
842
843 @pytest.mark.anyio
844 async def test_owner_permission_cannot_be_updated(
845 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
846 ) -> None:
847 repo = await _db_repo(db_session, owner="alice")
848 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="admin")
849 await _db_collab(db_session, repo.repo_id, "alice", permission="owner")
850 await db_session.commit()
851
852 r = await client.put(
853 f"/api/repos/{repo.repo_id}/collaborators/alice/permission",
854 json={"permission": "read"},
855 headers=auth_headers,
856 )
857 assert r.status_code == 403
858
859 @pytest.mark.anyio
860 async def test_owner_collab_cannot_be_removed(
861 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
862 ) -> None:
863 repo = await _db_repo(db_session, owner="alice")
864 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="admin")
865 await _db_collab(db_session, repo.repo_id, "alice", permission="owner")
866 await db_session.commit()
867
868 r = await client.delete(
869 f"/api/repos/{repo.repo_id}/collaborators/alice", headers=auth_headers
870 )
871 assert r.status_code == 403
872
873 @pytest.mark.anyio
874 async def test_check_access_requires_auth(
875 self, client: AsyncClient, db_session: AsyncSession
876 ) -> None:
877 repo = await _db_repo(db_session)
878 await db_session.commit()
879
880 r = await client.get(
881 f"/api/repos/{repo.repo_id}/collaborators/alice/permission"
882 )
883 assert r.status_code == 401
884
885 @pytest.mark.anyio
886 async def test_non_admin_cannot_update_permissions(
887 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
888 ) -> None:
889 repo = await _db_repo(db_session, owner="alice")
890 await _db_collab(db_session, repo.repo_id, _TEST_HANDLE, permission="write")
891 await _db_collab(db_session, repo.repo_id, "bob", permission="read")
892 await db_session.commit()
893
894 r = await client.put(
895 f"/api/repos/{repo.repo_id}/collaborators/bob/permission",
896 json={"permission": "admin"},
897 headers=auth_headers,
898 )
899 assert r.status_code == 403
900
901
902 # ===========================================================================
903 # Layer 7 — Performance
904 # ===========================================================================
905
906
907 class TestPerformance:
908 @pytest.mark.anyio
909 async def test_list_20_collaborators_under_100ms(
910 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
911 ) -> None:
912 repo_id = await _api_repo(client, auth_headers)
913 for i in range(20):
914 await _db_collab(db_session, repo_id, f"perf{i}")
915 await db_session.commit()
916
917 start = time.perf_counter()
918 r = await client.get(f"/api/repos/{repo_id}/collaborators", headers=auth_headers)
919 elapsed = time.perf_counter() - start
920
921 assert r.status_code == 200
922 assert elapsed < 0.1, f"list collaborators took {elapsed:.3f}s"
923
924 def test_has_permission_1m_calls_fast(self) -> None:
925 start = time.perf_counter()
926 for _ in range(1_000_000):
927 _has_permission("admin", Permission.write)
928 elapsed = time.perf_counter() - start
929 assert elapsed < 1.0, f"1M _has_permission calls took {elapsed:.3f}s"
930
931 @pytest.mark.anyio
932 async def test_invite_10_collabs_under_500ms(
933 self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession
934 ) -> None:
935 repo_id = await _api_repo(client, auth_headers)
936 await db_session.commit()
937
938 start = time.perf_counter()
939 for i in range(10):
940 r = await client.post(
941 f"/api/repos/{repo_id}/collaborators",
942 json={"handle": f"batch{i}", "permission": "read"},
943 headers=auth_headers,
944 )
945 assert r.status_code == 201
946 elapsed = time.perf_counter() - start
947 assert elapsed < 0.5, f"10 invite calls took {elapsed:.3f}s"
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago