gabriel / musehub public
test_musehub_repos.py python
2,160 lines 65.0 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Tests for MuseHub repo, branch, and commit endpoints.
2
3 Covers every acceptance criterion:
4 - POST /musehub/repos returns 201 with correct fields
5 - POST requires auth — unauthenticated requests return 401
6 - GET /repos/{repo_id} returns 200; 404 for unknown repo
7 - GET /repos/{repo_id}/branches returns empty list on new repo
8 - GET /repos/{repo_id}/commits returns newest first, respects ?limit
9
10 Covers (compare view API endpoint):
11 - test_compare_radar_data — compare endpoint returns 5 dimension scores
12 - test_compare_commit_list — commits unique to head are listed
13 - test_compare_unknown_ref_404 — unknown ref returns 422
14
15 All tests use the shared ``client`` and ``auth_headers`` fixtures from conftest.py.
16 """
17 from __future__ import annotations
18
19 import pytest
20 from httpx import AsyncClient
21 from sqlalchemy.ext.asyncio import AsyncSession
22
23 from musehub.db.musehub_models import MusehubCommit, MusehubRepo
24 from musehub.services import musehub_repository
25 from musehub.muse_contracts.json_types import StrDict
26
27
28 # ---------------------------------------------------------------------------
29 # POST /musehub/repos
30 # ---------------------------------------------------------------------------
31
32
33 @pytest.mark.anyio
34 async def test_create_repo_returns_201(
35 client: AsyncClient,
36 auth_headers: StrDict,
37 ) -> None:
38 """POST /musehub/repos creates a repo and returns all required fields."""
39 response = await client.post(
40 "/api/repos",
41 json={"name": "my-beats", "owner": "testuser", "visibility": "private"},
42 headers=auth_headers,
43 )
44 assert response.status_code == 201
45 body = response.json()
46 assert body["name"] == "my-beats"
47 assert body["visibility"] == "private"
48 assert "repoId" in body
49 assert "cloneUrl" in body
50 assert "ownerUserId" in body
51 assert "createdAt" in body
52
53
54 @pytest.mark.anyio
55 async def test_create_repo_requires_auth(client: AsyncClient) -> None:
56 """POST /musehub/repos returns 401 without a MSign Authorization header."""
57 response = await client.post(
58 "/api/repos",
59 json={"name": "my-beats", "owner": "testuser"},
60 )
61 assert response.status_code == 401
62
63
64 @pytest.mark.anyio
65 async def test_create_repo_default_visibility_is_private(
66 client: AsyncClient,
67 auth_headers: StrDict,
68 ) -> None:
69 """Omitting visibility defaults to 'private'."""
70 response = await client.post(
71 "/api/repos",
72 json={"name": "silent-sessions", "owner": "testuser"},
73 headers=auth_headers,
74 )
75 assert response.status_code == 201
76 assert response.json()["visibility"] == "private"
77
78
79 # ---------------------------------------------------------------------------
80 # GET /repos/{repo_id}
81 # ---------------------------------------------------------------------------
82
83
84 @pytest.mark.anyio
85 async def test_get_repo_returns_200(
86 client: AsyncClient,
87 auth_headers: StrDict,
88 ) -> None:
89 """GET /repos/{repo_id} returns the repo after creation."""
90 create = await client.post(
91 "/api/repos",
92 json={"name": "jazz-sessions", "owner": "testuser"},
93 headers=auth_headers,
94 )
95 assert create.status_code == 201
96 repo_id = create.json()["repoId"]
97
98 response = await client.get(f"/api/repos/{repo_id}", headers=auth_headers)
99 assert response.status_code == 200
100 assert response.json()["repoId"] == repo_id
101 assert response.json()["name"] == "jazz-sessions"
102
103
104 @pytest.mark.anyio
105 async def test_get_repo_not_found_returns_404(
106 client: AsyncClient,
107 auth_headers: StrDict,
108 ) -> None:
109 """GET /repos/{repo_id} returns 404 for unknown repo."""
110 response = await client.get(
111 "/api/repos/does-not-exist",
112 headers=auth_headers,
113 )
114 assert response.status_code == 404
115
116
117 @pytest.mark.anyio
118 async def test_get_nonexistent_repo_returns_404_without_auth(client: AsyncClient) -> None:
119 """GET /repos/{repo_id} returns 404 for a non-existent repo without auth.
120
121 Uses optional_token — auth is visibility-based; missing repo → 404 before auth check.
122 """
123 response = await client.get("/api/repos/non-existent-repo-id")
124 assert response.status_code == 404
125
126
127 # ---------------------------------------------------------------------------
128 # GET /repos/{repo_id}/branches
129 # ---------------------------------------------------------------------------
130
131
132 @pytest.mark.anyio
133 async def test_list_branches_empty_on_new_repo(
134 client: AsyncClient,
135 auth_headers: StrDict,
136 ) -> None:
137 """A newly created repo has an empty branches list when not initialized."""
138 create = await client.post(
139 "/api/repos",
140 json={"name": "drum-patterns", "owner": "testuser", "initialize": False},
141 headers=auth_headers,
142 )
143 repo_id = create.json()["repoId"]
144
145 response = await client.get(
146 f"/api/repos/{repo_id}/branches",
147 headers=auth_headers,
148 )
149 assert response.status_code == 200
150 assert response.json()["branches"] == []
151
152
153 @pytest.mark.anyio
154 async def test_list_branches_not_found_returns_404(
155 client: AsyncClient,
156 auth_headers: StrDict,
157 ) -> None:
158 """GET /branches returns 404 when the repo doesn't exist."""
159 response = await client.get(
160 "/api/repos/ghost-repo/branches",
161 headers=auth_headers,
162 )
163 assert response.status_code == 404
164
165
166 # ---------------------------------------------------------------------------
167 # GET /repos/{repo_id}/commits
168 # ---------------------------------------------------------------------------
169
170
171 @pytest.mark.anyio
172 async def test_list_commits_empty_on_new_repo(
173 client: AsyncClient,
174 auth_headers: StrDict,
175 ) -> None:
176 """A new repo has no commits when initialize=false."""
177 create = await client.post(
178 "/api/repos",
179 json={"name": "empty-repo", "owner": "testuser", "initialize": False},
180 headers=auth_headers,
181 )
182 repo_id = create.json()["repoId"]
183
184 response = await client.get(
185 f"/api/repos/{repo_id}/commits",
186 headers=auth_headers,
187 )
188 assert response.status_code == 200
189 body = response.json()
190 assert body["commits"] == []
191 assert body["total"] == 0
192
193
194 @pytest.mark.anyio
195 async def test_list_commits_returns_newest_first(
196 client: AsyncClient,
197 auth_headers: StrDict,
198 db_session: AsyncSession,
199 ) -> None:
200 """Commits are returned newest-first after being pushed."""
201 from datetime import datetime, timezone, timedelta
202
203 # Create repo via API (no init commit so we control the full history)
204 create = await client.post(
205 "/api/repos",
206 json={"name": "ordered-commits", "owner": "testuser", "initialize": False},
207 headers=auth_headers,
208 )
209 repo_id = create.json()["repoId"]
210
211 # Insert two commits directly with known timestamps
212 now = datetime.now(tz=timezone.utc)
213 older = MusehubCommit(
214 commit_id="aaa111",
215 repo_id=repo_id,
216 branch="main",
217 parent_ids=[],
218 message="first",
219 author="gabriel",
220 timestamp=now - timedelta(hours=1),
221 )
222 newer = MusehubCommit(
223 commit_id="bbb222",
224 repo_id=repo_id,
225 branch="main",
226 parent_ids=["aaa111"],
227 message="second",
228 author="gabriel",
229 timestamp=now,
230 )
231 db_session.add_all([older, newer])
232 await db_session.commit()
233
234 response = await client.get(
235 f"/api/repos/{repo_id}/commits",
236 headers=auth_headers,
237 )
238 assert response.status_code == 200
239 commits = response.json()["commits"]
240 assert len(commits) == 2
241 assert commits[0]["commitId"] == "bbb222"
242 assert commits[1]["commitId"] == "aaa111"
243
244
245 @pytest.mark.anyio
246 async def test_list_commits_limit_param(
247 client: AsyncClient,
248 auth_headers: StrDict,
249 db_session: AsyncSession,
250 ) -> None:
251 """?limit=1 returns exactly 1 commit."""
252 from datetime import datetime, timezone, timedelta
253
254 create = await client.post(
255 "/api/repos",
256 json={"name": "limited-repo", "owner": "testuser", "initialize": False},
257 headers=auth_headers,
258 )
259 repo_id = create.json()["repoId"]
260
261 now = datetime.now(tz=timezone.utc)
262 for i in range(3):
263 db_session.add(
264 MusehubCommit(
265 commit_id=f"commit-{i}",
266 repo_id=repo_id,
267 branch="main",
268 parent_ids=[],
269 message=f"commit {i}",
270 author="gabriel",
271 timestamp=now + timedelta(seconds=i),
272 )
273 )
274 await db_session.commit()
275
276 response = await client.get(
277 f"/api/repos/{repo_id}/commits?limit=1",
278 headers=auth_headers,
279 )
280 assert response.status_code == 200
281 body = response.json()
282 assert len(body["commits"]) == 1
283 assert body["total"] == 3
284
285
286 # ---------------------------------------------------------------------------
287 # Service layer — direct DB tests (no HTTP)
288 # ---------------------------------------------------------------------------
289
290
291 @pytest.mark.anyio
292 async def test_create_repo_service_persists_to_db(db_session: AsyncSession) -> None:
293 """musehub_repository.create_repo() persists the row."""
294 repo = await musehub_repository.create_repo(
295 db_session,
296 name="service-test-repo",
297 owner="testuser",
298 visibility="public",
299 owner_user_id="user-abc",
300 )
301 await db_session.commit()
302
303 fetched = await musehub_repository.get_repo(db_session, repo.repo_id)
304 assert fetched is not None
305 assert fetched.name == "service-test-repo"
306 assert fetched.visibility == "public"
307
308
309 @pytest.mark.anyio
310 async def test_get_repo_returns_none_when_missing(db_session: AsyncSession) -> None:
311 """get_repo() returns None for an unknown repo_id."""
312 result = await musehub_repository.get_repo(db_session, "nonexistent-id")
313 assert result is None
314
315
316 @pytest.mark.anyio
317 async def test_list_branches_returns_empty_for_new_repo(db_session: AsyncSession) -> None:
318 """list_branches() returns [] for a repo with no branches."""
319 repo = await musehub_repository.create_repo(
320 db_session,
321 name="branchless",
322 owner="testuser",
323 visibility="private",
324 owner_user_id="user-x",
325 )
326 await db_session.commit()
327 branches = await musehub_repository.list_branches(db_session, repo.repo_id)
328 assert branches == []
329
330
331 # ---------------------------------------------------------------------------
332 # GET /repos/{repo_id}/divergence
333 # ---------------------------------------------------------------------------
334
335
336 @pytest.mark.anyio
337 async def test_divergence_endpoint_returns_five_dimensions(
338 client: AsyncClient,
339 auth_headers: StrDict,
340 db_session: AsyncSession,
341 ) -> None:
342 """GET /divergence returns five dimension scores with level labels."""
343 from datetime import datetime, timezone, timedelta
344
345 create = await client.post(
346 "/api/repos",
347 json={"name": "divergence-test-repo", "owner": "testuser"},
348 headers=auth_headers,
349 )
350 assert create.status_code == 201
351 repo_id = create.json()["repoId"]
352
353 now = datetime.now(tz=timezone.utc)
354 db_session.add(
355 MusehubCommit(
356 commit_id="aaa-melody",
357 repo_id=repo_id,
358 branch="main",
359 parent_ids=[],
360 message="add lead melody line",
361 author="alice",
362 timestamp=now - timedelta(hours=2),
363 )
364 )
365 db_session.add(
366 MusehubCommit(
367 commit_id="bbb-chord",
368 repo_id=repo_id,
369 branch="feature",
370 parent_ids=[],
371 message="update chord progression",
372 author="bob",
373 timestamp=now - timedelta(hours=1),
374 )
375 )
376 await db_session.commit()
377
378 response = await client.get(
379 f"/api/repos/{repo_id}/divergence?branch_a=main&branch_b=feature",
380 headers=auth_headers,
381 )
382 assert response.status_code == 200
383 body = response.json()
384 assert "dimensions" in body
385 assert len(body["dimensions"]) == 5
386
387 dim_names = {d["dimension"] for d in body["dimensions"]}
388 assert dim_names == {"melodic", "harmonic", "rhythmic", "structural", "dynamic"}
389
390 for dim in body["dimensions"]:
391 assert "level" in dim
392 assert dim["level"] in {"NONE", "LOW", "MED", "HIGH"}
393 assert "score" in dim
394 assert 0.0 <= dim["score"] <= 1.0
395
396
397 @pytest.mark.anyio
398 async def test_divergence_overall_score_is_mean_of_dimensions(
399 client: AsyncClient,
400 auth_headers: StrDict,
401 db_session: AsyncSession,
402 ) -> None:
403 """Overall divergence score equals the mean of all five dimension scores."""
404 from datetime import datetime, timezone, timedelta
405
406 create = await client.post(
407 "/api/repos",
408 json={"name": "divergence-mean-repo", "owner": "testuser"},
409 headers=auth_headers,
410 )
411 repo_id = create.json()["repoId"]
412
413 now = datetime.now(tz=timezone.utc)
414 db_session.add(
415 MusehubCommit(
416 commit_id="c1-beat",
417 repo_id=repo_id,
418 branch="alpha",
419 parent_ids=[],
420 message="rework drum beat groove",
421 author="producer-a",
422 timestamp=now - timedelta(hours=3),
423 )
424 )
425 db_session.add(
426 MusehubCommit(
427 commit_id="c2-mix",
428 repo_id=repo_id,
429 branch="beta",
430 parent_ids=[],
431 message="fix master volume level",
432 author="producer-b",
433 timestamp=now - timedelta(hours=2),
434 )
435 )
436 await db_session.commit()
437
438 response = await client.get(
439 f"/api/repos/{repo_id}/divergence?branch_a=alpha&branch_b=beta",
440 headers=auth_headers,
441 )
442 assert response.status_code == 200
443 body = response.json()
444
445 dims = body["dimensions"]
446 computed_mean = round(sum(d["score"] for d in dims) / len(dims), 4)
447 assert abs(body["overallScore"] - computed_mean) < 1e-6
448
449
450 @pytest.mark.anyio
451 async def test_divergence_json_response_structure(
452 client: AsyncClient,
453 auth_headers: StrDict,
454 db_session: AsyncSession,
455 ) -> None:
456 """JSON response has all required top-level fields and camelCase keys."""
457 from datetime import datetime, timezone, timedelta
458
459 create = await client.post(
460 "/api/repos",
461 json={"name": "divergence-struct-repo", "owner": "testuser"},
462 headers=auth_headers,
463 )
464 repo_id = create.json()["repoId"]
465
466 now = datetime.now(tz=timezone.utc)
467 for i, (branch, msg) in enumerate(
468 [("main", "add melody riff"), ("dev", "update chorus section")]
469 ):
470 db_session.add(
471 MusehubCommit(
472 commit_id=f"struct-{i}",
473 repo_id=repo_id,
474 branch=branch,
475 parent_ids=[],
476 message=msg,
477 author="test",
478 timestamp=now + timedelta(seconds=i),
479 )
480 )
481 await db_session.commit()
482
483 response = await client.get(
484 f"/api/repos/{repo_id}/divergence?branch_a=main&branch_b=dev",
485 headers=auth_headers,
486 )
487 assert response.status_code == 200
488 body = response.json()
489
490 assert body["repoId"] == repo_id
491 assert body["branchA"] == "main"
492 assert body["branchB"] == "dev"
493 assert "commonAncestor" in body
494 assert "overallScore" in body
495 assert isinstance(body["overallScore"], float)
496 assert isinstance(body["dimensions"], list)
497 assert len(body["dimensions"]) == 5
498
499 for dim in body["dimensions"]:
500 assert "dimension" in dim
501 assert "level" in dim
502 assert "score" in dim
503 assert "description" in dim
504 assert "branchACommits" in dim
505 assert "branchBCommits" in dim
506
507
508 @pytest.mark.anyio
509 async def test_divergence_endpoint_returns_404_for_unknown_repo(
510 client: AsyncClient,
511 auth_headers: StrDict,
512 ) -> None:
513 """GET /divergence returns 404 for an unknown repo."""
514 response = await client.get(
515 "/api/repos/no-such-repo/divergence?branch_a=a&branch_b=b",
516 headers=auth_headers,
517 )
518 assert response.status_code == 404
519
520
521 @pytest.mark.anyio
522 async def test_divergence_endpoint_returns_422_for_empty_branch(
523 client: AsyncClient,
524 auth_headers: StrDict,
525 db_session: AsyncSession,
526 ) -> None:
527 """GET /divergence returns 422 when a branch has no commits."""
528 create = await client.post(
529 "/api/repos",
530 json={"name": "empty-branch-repo", "owner": "testuser"},
531 headers=auth_headers,
532 )
533 repo_id = create.json()["repoId"]
534
535 response = await client.get(
536 f"/api/repos/{repo_id}/divergence?branch_a=ghost&branch_b=also-ghost",
537 headers=auth_headers,
538 )
539 assert response.status_code == 422
540
541
542
543 # ---------------------------------------------------------------------------
544 # GET /repos/{repo_id}/dag
545 # ---------------------------------------------------------------------------
546
547
548 @pytest.mark.anyio
549 async def test_graph_dag_endpoint_returns_empty_for_new_repo(
550 client: AsyncClient,
551 auth_headers: StrDict,
552 ) -> None:
553 """GET /dag returns empty nodes/edges for a repo with no commits (initialize=false)."""
554 create = await client.post(
555 "/api/repos",
556 json={"name": "dag-empty", "owner": "testuser", "initialize": False},
557 headers=auth_headers,
558 )
559 repo_id = create.json()["repoId"]
560
561 response = await client.get(
562 f"/api/repos/{repo_id}/dag",
563 headers=auth_headers,
564 )
565 assert response.status_code == 200
566 body = response.json()
567 assert body["nodes"] == []
568 assert body["edges"] == []
569 assert body["headCommitId"] is None
570
571
572 @pytest.mark.anyio
573 async def test_graph_dag_has_edges(
574 client: AsyncClient,
575 auth_headers: StrDict,
576 db_session: AsyncSession,
577 ) -> None:
578 """DAG endpoint returns correct edges representing parent relationships."""
579 from datetime import datetime, timezone, timedelta
580
581 create = await client.post(
582 "/api/repos",
583 json={"name": "dag-edges", "owner": "testuser", "initialize": False},
584 headers=auth_headers,
585 )
586 repo_id = create.json()["repoId"]
587
588 now = datetime.now(tz=timezone.utc)
589 root = MusehubCommit(
590 commit_id="root111",
591 repo_id=repo_id,
592 branch="main",
593 parent_ids=[],
594 message="root commit",
595 author="gabriel",
596 timestamp=now - timedelta(hours=2),
597 )
598 child = MusehubCommit(
599 commit_id="child222",
600 repo_id=repo_id,
601 branch="main",
602 parent_ids=["root111"],
603 message="child commit",
604 author="gabriel",
605 timestamp=now - timedelta(hours=1),
606 )
607 db_session.add_all([root, child])
608 await db_session.commit()
609
610 response = await client.get(
611 f"/api/repos/{repo_id}/dag",
612 headers=auth_headers,
613 )
614 assert response.status_code == 200
615 body = response.json()
616 nodes = body["nodes"]
617 edges = body["edges"]
618
619 assert len(nodes) == 2
620 # Verify edge: child → root
621 assert any(e["source"] == "child222" and e["target"] == "root111" for e in edges)
622
623
624 @pytest.mark.anyio
625 async def test_graph_dag_endpoint_topological_order(
626 client: AsyncClient,
627 auth_headers: StrDict,
628 db_session: AsyncSession,
629 ) -> None:
630 """DAG endpoint returns nodes in topological order (oldest ancestor first)."""
631 from datetime import datetime, timedelta, timezone
632
633 create = await client.post(
634 "/api/repos",
635 json={"name": "dag-topo", "owner": "testuser"},
636 headers=auth_headers,
637 )
638 repo_id = create.json()["repoId"]
639
640 now = datetime.now(tz=timezone.utc)
641 commits = [
642 MusehubCommit(
643 commit_id="topo-a",
644 repo_id=repo_id,
645 branch="main",
646 parent_ids=[],
647 message="root",
648 author="gabriel",
649 timestamp=now - timedelta(hours=3),
650 ),
651 MusehubCommit(
652 commit_id="topo-b",
653 repo_id=repo_id,
654 branch="main",
655 parent_ids=["topo-a"],
656 message="second",
657 author="gabriel",
658 timestamp=now - timedelta(hours=2),
659 ),
660 MusehubCommit(
661 commit_id="topo-c",
662 repo_id=repo_id,
663 branch="main",
664 parent_ids=["topo-b"],
665 message="third",
666 author="gabriel",
667 timestamp=now - timedelta(hours=1),
668 ),
669 ]
670 db_session.add_all(commits)
671 await db_session.commit()
672
673 response = await client.get(
674 f"/api/repos/{repo_id}/dag",
675 headers=auth_headers,
676 )
677 assert response.status_code == 200
678 node_ids = [n["commitId"] for n in response.json()["nodes"]]
679 # Root must appear before children in topological order
680 assert node_ids.index("topo-a") < node_ids.index("topo-b")
681 assert node_ids.index("topo-b") < node_ids.index("topo-c")
682
683
684 @pytest.mark.anyio
685 async def test_graph_dag_nonexistent_repo_returns_404_without_auth(client: AsyncClient) -> None:
686 """GET /dag returns 404 for a non-existent repo without a token.
687
688 Uses optional_token — auth is visibility-based; missing repo → 404.
689 """
690 response = await client.get("/api/repos/non-existent-repo/dag")
691 assert response.status_code == 404
692
693
694 @pytest.mark.anyio
695 async def test_graph_dag_404_for_unknown_repo(
696 client: AsyncClient,
697 auth_headers: StrDict,
698 ) -> None:
699 """GET /dag returns 404 for a non-existent repo."""
700 response = await client.get(
701 "/api/repos/ghost-repo-dag/dag",
702 headers=auth_headers,
703 )
704 assert response.status_code == 404
705
706
707 @pytest.mark.anyio
708 async def test_graph_json_response_has_required_fields(
709 client: AsyncClient,
710 auth_headers: StrDict,
711 db_session: AsyncSession,
712 ) -> None:
713 """DAG JSON response includes nodes (with required fields) and edges arrays."""
714 from datetime import datetime, timezone
715
716 create = await client.post(
717 "/api/repos",
718 json={"name": "dag-fields", "owner": "testuser"},
719 headers=auth_headers,
720 )
721 repo_id = create.json()["repoId"]
722
723 db_session.add(
724 MusehubCommit(
725 commit_id="fields-aaa",
726 repo_id=repo_id,
727 branch="main",
728 parent_ids=[],
729 message="check fields",
730 author="tester",
731 timestamp=datetime.now(tz=timezone.utc),
732 )
733 )
734 await db_session.commit()
735
736 response = await client.get(
737 f"/api/repos/{repo_id}/dag",
738 headers=auth_headers,
739 )
740 assert response.status_code == 200
741 body = response.json()
742 assert "nodes" in body
743 assert "edges" in body
744 assert "headCommitId" in body
745
746 node = body["nodes"][0]
747 for field in ("commitId", "message", "author", "timestamp", "branch", "parentIds", "isHead"):
748 assert field in node, f"Missing field '{field}' in DAG node"
749
750 # ---------------------------------------------------------------------------
751 # GET /repos/{repo_id}/credits
752 # ---------------------------------------------------------------------------
753
754
755 async def _seed_credits_repo(db_session: AsyncSession) -> str:
756 """Create a repo with commits from two distinct authors and return repo_id."""
757 from datetime import datetime, timezone, timedelta
758
759 repo = MusehubRepo(name="liner-notes",
760 owner="testuser",
761 slug="liner-notes", visibility="public", owner_user_id="producer-1")
762 db_session.add(repo)
763 await db_session.flush()
764 repo_id = str(repo.repo_id)
765
766 now = datetime.now(tz=timezone.utc)
767 # Alice: 2 commits (most prolific), most recent 1 day ago
768 db_session.add(
769 MusehubCommit(
770 commit_id="alice-001",
771 repo_id=repo_id,
772 branch="main",
773 parent_ids=[],
774 message="compose the main melody",
775 author="Alice",
776 timestamp=now - timedelta(days=3),
777 )
778 )
779 db_session.add(
780 MusehubCommit(
781 commit_id="alice-002",
782 repo_id=repo_id,
783 branch="main",
784 parent_ids=["alice-001"],
785 message="mix the final arrangement",
786 author="Alice",
787 timestamp=now - timedelta(days=1),
788 )
789 )
790 # Bob: 1 commit, last active 5 days ago
791 db_session.add(
792 MusehubCommit(
793 commit_id="bob-001",
794 repo_id=repo_id,
795 branch="main",
796 parent_ids=[],
797 message="arrange the bridge section",
798 author="Bob",
799 timestamp=now - timedelta(days=5),
800 )
801 )
802 await db_session.commit()
803 return repo_id
804
805
806 @pytest.mark.anyio
807 async def test_credits_aggregation(
808 client: AsyncClient,
809 db_session: AsyncSession,
810 auth_headers: StrDict,
811 ) -> None:
812 """GET /api/repos/{repo_id}/credits aggregates contributors from commits."""
813 repo_id = await _seed_credits_repo(db_session)
814 response = await client.get(
815 f"/api/repos/{repo_id}/credits",
816 headers=auth_headers,
817 )
818 assert response.status_code == 200
819 body = response.json()
820 assert body["totalContributors"] == 2
821 authors = {c["author"] for c in body["contributors"]}
822 assert "Alice" in authors
823 assert "Bob" in authors
824
825
826 @pytest.mark.anyio
827 async def test_credits_sorted_by_count(
828 client: AsyncClient,
829 db_session: AsyncSession,
830 auth_headers: StrDict,
831 ) -> None:
832 """Default sort (count) puts the most prolific contributor first."""
833 repo_id = await _seed_credits_repo(db_session)
834 response = await client.get(
835 f"/api/repos/{repo_id}/credits?sort=count",
836 headers=auth_headers,
837 )
838 assert response.status_code == 200
839 contributors = response.json()["contributors"]
840 assert contributors[0]["author"] == "Alice"
841 assert contributors[0]["sessionCount"] == 2
842
843
844 @pytest.mark.anyio
845 async def test_credits_sorted_by_recency(
846 client: AsyncClient,
847 db_session: AsyncSession,
848 auth_headers: StrDict,
849 ) -> None:
850 """sort=recency puts the most recently active contributor first."""
851 repo_id = await _seed_credits_repo(db_session)
852 response = await client.get(
853 f"/api/repos/{repo_id}/credits?sort=recency",
854 headers=auth_headers,
855 )
856 assert response.status_code == 200
857 contributors = response.json()["contributors"]
858 # Alice has a commit 1 day ago; Bob's last was 5 days ago
859 assert contributors[0]["author"] == "Alice"
860
861
862 @pytest.mark.anyio
863 async def test_credits_sorted_by_alpha(
864 client: AsyncClient,
865 db_session: AsyncSession,
866 auth_headers: StrDict,
867 ) -> None:
868 """sort=alpha returns contributors in alphabetical order."""
869 repo_id = await _seed_credits_repo(db_session)
870 response = await client.get(
871 f"/api/repos/{repo_id}/credits?sort=alpha",
872 headers=auth_headers,
873 )
874 assert response.status_code == 200
875 contributors = response.json()["contributors"]
876 authors = [c["author"] for c in contributors]
877 assert authors == sorted(authors, key=str.lower)
878
879
880 @pytest.mark.anyio
881 async def test_credits_contribution_types_inferred(
882 client: AsyncClient,
883 db_session: AsyncSession,
884 auth_headers: StrDict,
885 ) -> None:
886 """Contribution types are inferred from commit messages."""
887 repo_id = await _seed_credits_repo(db_session)
888 response = await client.get(
889 f"/api/repos/{repo_id}/credits",
890 headers=auth_headers,
891 )
892 assert response.status_code == 200
893 contributors = response.json()["contributors"]
894 alice = next(c for c in contributors if c["author"] == "Alice")
895 # Alice's commits mention "compose" and "mix"
896 types = set(alice["contributionTypes"])
897 assert len(types) > 0
898
899
900 @pytest.mark.anyio
901 async def test_credits_404_for_unknown_repo(
902 client: AsyncClient,
903 auth_headers: StrDict,
904 ) -> None:
905 """GET /api/repos/{unknown}/credits returns 404."""
906 response = await client.get(
907 "/api/repos/does-not-exist/credits",
908 headers=auth_headers,
909 )
910 assert response.status_code == 404
911
912
913 @pytest.mark.anyio
914 async def test_credits_requires_auth(
915 client: AsyncClient,
916 db_session: AsyncSession,
917 ) -> None:
918 """GET /api/repos/{repo_id}/credits returns 401 without MSign auth."""
919 repo = MusehubRepo(name="auth-test-repo",
920 owner="testuser",
921 slug="auth-test-repo", visibility="private", owner_user_id="u1")
922 db_session.add(repo)
923 await db_session.commit()
924 await db_session.refresh(repo)
925 response = await client.get(f"/api/repos/{repo.repo_id}/credits")
926 assert response.status_code == 401
927
928
929 @pytest.mark.anyio
930 async def test_credits_invalid_sort_param(
931 client: AsyncClient,
932 db_session: AsyncSession,
933 auth_headers: StrDict,
934 ) -> None:
935 """GET /api/repos/{repo_id}/credits with invalid sort returns 422."""
936 repo = MusehubRepo(name="sort-test",
937 owner="testuser",
938 slug="sort-test", visibility="private", owner_user_id="u1")
939 db_session.add(repo)
940 await db_session.commit()
941 await db_session.refresh(repo)
942 response = await client.get(
943 f"/api/repos/{repo.repo_id}/credits?sort=invalid",
944 headers=auth_headers,
945 )
946 assert response.status_code == 422
947
948
949 @pytest.mark.anyio
950 async def test_credits_aggregation_service_direct(db_session: AsyncSession) -> None:
951 """musehub_credits.aggregate_credits() returns correct data without HTTP layer."""
952 from datetime import datetime, timezone
953
954 from musehub.services import musehub_credits
955
956 repo = MusehubRepo(name="direct-test",
957 owner="testuser",
958 slug="direct-test", visibility="private", owner_user_id="u1")
959 db_session.add(repo)
960 await db_session.flush()
961 repo_id = str(repo.repo_id)
962
963 now = datetime.now(tz=timezone.utc)
964 db_session.add(
965 MusehubCommit(
966 commit_id="svc-001",
967 repo_id=repo_id,
968 branch="main",
969 parent_ids=[],
970 message="produce and mix the drop",
971 author="Charlie",
972 timestamp=now,
973 )
974 )
975 await db_session.commit()
976
977 result = await musehub_credits.aggregate_credits(db_session, repo_id, sort="count")
978 assert result.total_contributors == 1
979 assert result.contributors[0].author == "Charlie"
980 assert result.contributors[0].session_count == 1
981
982
983 # ---------------------------------------------------------------------------
984 # Compare endpoint
985 # ---------------------------------------------------------------------------
986
987
988 async def _make_compare_repo(
989 db_session: AsyncSession,
990 client: AsyncClient,
991 auth_headers: StrDict,
992 ) -> str:
993 """Seed a repo with commits on two branches and return repo_id."""
994 from datetime import datetime, timezone
995
996 create = await client.post(
997 "/api/repos",
998 json={"name": "compare-test", "owner": "testuser", "visibility": "private"},
999 headers=auth_headers,
1000 )
1001 assert create.status_code == 201
1002 repo_id: str = str(create.json()["repoId"])
1003
1004 now = datetime.now(tz=timezone.utc)
1005 db_session.add(
1006 MusehubCommit(
1007 commit_id="base001",
1008 repo_id=repo_id,
1009 branch="main",
1010 parent_ids=[],
1011 message="add melody line",
1012 author="Alice",
1013 timestamp=now,
1014 )
1015 )
1016 db_session.add(
1017 MusehubCommit(
1018 commit_id="head001",
1019 repo_id=repo_id,
1020 branch="feature",
1021 parent_ids=["base001"],
1022 message="add chord progression",
1023 author="Bob",
1024 timestamp=now,
1025 )
1026 )
1027 await db_session.commit()
1028 return repo_id
1029
1030
1031 @pytest.mark.anyio
1032 async def test_compare_radar_data(
1033 client: AsyncClient,
1034 db_session: AsyncSession,
1035 auth_headers: StrDict,
1036 ) -> None:
1037 """GET /api/repos/{id}/compare returns 5 dimension scores."""
1038 repo_id = await _make_compare_repo(db_session, client, auth_headers)
1039 response = await client.get(
1040 f"/api/repos/{repo_id}/compare?base=main&head=feature",
1041 headers=auth_headers,
1042 )
1043 assert response.status_code == 200
1044 body = response.json()
1045 assert "dimensions" in body
1046 assert len(body["dimensions"]) == 5
1047 expected_dims = {"melodic", "harmonic", "rhythmic", "structural", "dynamic"}
1048 found_dims = {d["dimension"] for d in body["dimensions"]}
1049 assert found_dims == expected_dims
1050 for dim in body["dimensions"]:
1051 assert 0.0 <= dim["score"] <= 1.0
1052 assert dim["level"] in ("NONE", "LOW", "MED", "HIGH")
1053 assert "overallScore" in body
1054 assert 0.0 <= body["overallScore"] <= 1.0
1055
1056
1057 @pytest.mark.anyio
1058 async def test_compare_commit_list(
1059 client: AsyncClient,
1060 db_session: AsyncSession,
1061 auth_headers: StrDict,
1062 ) -> None:
1063 """Commits unique to head are listed in the compare response."""
1064 repo_id = await _make_compare_repo(db_session, client, auth_headers)
1065 response = await client.get(
1066 f"/api/repos/{repo_id}/compare?base=main&head=feature",
1067 headers=auth_headers,
1068 )
1069 assert response.status_code == 200
1070 body = response.json()
1071 assert "commits" in body
1072 # head001 is on feature but not on main
1073 commit_ids = [c["commitId"] for c in body["commits"]]
1074 assert "head001" in commit_ids
1075 # base001 is on main so should NOT appear as unique to head
1076 assert "base001" not in commit_ids
1077
1078
1079 @pytest.mark.anyio
1080 async def test_compare_unknown_ref_422(
1081 client: AsyncClient,
1082 db_session: AsyncSession,
1083 auth_headers: StrDict,
1084 ) -> None:
1085 """Unknown ref (branch with no commits) returns 422."""
1086 create = await client.post(
1087 "/api/repos",
1088 json={"name": "empty-compare", "owner": "testuser", "visibility": "private"},
1089 headers=auth_headers,
1090 )
1091 assert create.status_code == 201
1092 repo_id = create.json()["repoId"]
1093 response = await client.get(
1094 f"/api/repos/{repo_id}/compare?base=nonexistent&head=alsoabsent",
1095 headers=auth_headers,
1096 )
1097 assert response.status_code == 422
1098
1099
1100 @pytest.mark.anyio
1101 async def test_compare_emotion_diff_fields(
1102 client: AsyncClient,
1103 db_session: AsyncSession,
1104 auth_headers: StrDict,
1105 ) -> None:
1106 """Compare response includes emotion diff with required delta fields."""
1107 repo_id = await _make_compare_repo(db_session, client, auth_headers)
1108 response = await client.get(
1109 f"/api/repos/{repo_id}/compare?base=main&head=feature",
1110 headers=auth_headers,
1111 )
1112 assert response.status_code == 200
1113 body = response.json()
1114 assert "emotionDiff" in body
1115 ed = body["emotionDiff"]
1116 for field in ("energyDelta", "valenceDelta", "tensionDelta", "darknessDelta"):
1117 assert field in ed
1118 assert -1.0 <= ed[field] <= 1.0
1119 for field in ("baseEnergy", "headEnergy", "baseValence", "headValence"):
1120 assert field in ed
1121 assert 0.0 <= ed[field] <= 1.0
1122
1123
1124
1125 # ---------------------------------------------------------------------------
1126 # ---------------------------------------------------------------------------
1127 # GET /repos/{repo_id}/settings
1128 # ---------------------------------------------------------------------------
1129
1130 TEST_OWNER_USER_ID = "550e8400-e29b-41d4-a716-446655440000"
1131
1132
1133 @pytest.mark.anyio
1134 async def test_get_repo_settings_returns_defaults(
1135 client: AsyncClient,
1136 db_session: AsyncSession,
1137 auth_headers: StrDict,
1138 ) -> None:
1139 """GET /repos/{repo_id}/settings returns full settings with canonical defaults."""
1140 repo = MusehubRepo(
1141 name="settings-get-test",
1142 owner="testuser",
1143 slug="settings-get-test",
1144 visibility="private",
1145 owner_user_id=TEST_OWNER_USER_ID,
1146 )
1147 db_session.add(repo)
1148 await db_session.commit()
1149 await db_session.refresh(repo)
1150
1151 resp = await client.get(
1152 f"/api/repos/{repo.repo_id}/settings",
1153 headers=auth_headers,
1154 )
1155 assert resp.status_code == 200
1156 body = resp.json()
1157 assert body["name"] == "settings-get-test"
1158 assert body["visibility"] == "private"
1159 assert body["hasIssues"] is True
1160 assert body["allowMergeCommit"] is True
1161 assert body["allowRebaseMerge"] is False
1162 assert body["deleteBranchOnMerge"] is True
1163 assert body["defaultBranch"] == "main"
1164
1165
1166 @pytest.mark.anyio
1167 async def test_get_repo_settings_requires_auth(
1168 client: AsyncClient,
1169 db_session: AsyncSession,
1170 ) -> None:
1171 """GET /repos/{repo_id}/settings returns 401 without a MSign Authorization header."""
1172 repo = MusehubRepo(
1173 name="settings-noauth",
1174 owner="testuser",
1175 slug="settings-noauth",
1176 visibility="private",
1177 owner_user_id=TEST_OWNER_USER_ID,
1178 )
1179 db_session.add(repo)
1180 await db_session.commit()
1181 await db_session.refresh(repo)
1182
1183 resp = await client.get(f"/api/repos/{repo.repo_id}/settings")
1184 assert resp.status_code == 401
1185
1186
1187 @pytest.mark.anyio
1188 async def test_get_repo_settings_returns_403_for_non_admin(
1189 client: AsyncClient,
1190 db_session: AsyncSession,
1191 auth_headers: StrDict,
1192 ) -> None:
1193 """GET /repos/{repo_id}/settings returns 403 when caller is not owner or admin."""
1194 repo = MusehubRepo(
1195 name="settings-403-test",
1196 owner="other-owner",
1197 slug="settings-403-test",
1198 visibility="public",
1199 owner_user_id="other-user-id-not-test",
1200 )
1201 db_session.add(repo)
1202 await db_session.commit()
1203 await db_session.refresh(repo)
1204
1205 resp = await client.get(
1206 f"/api/repos/{repo.repo_id}/settings",
1207 headers=auth_headers,
1208 )
1209 assert resp.status_code == 403
1210
1211
1212 @pytest.mark.anyio
1213 async def test_get_repo_settings_returns_404_for_unknown_repo(
1214 client: AsyncClient,
1215 auth_headers: StrDict,
1216 ) -> None:
1217 """GET /repos/{repo_id}/settings returns 404 for a non-existent repo."""
1218 resp = await client.get(
1219 "/api/repos/nonexistent-repo-id/settings",
1220 headers=auth_headers,
1221 )
1222 assert resp.status_code == 404
1223
1224
1225 # ---------------------------------------------------------------------------
1226 # PATCH /repos/{repo_id}/settings
1227 # ---------------------------------------------------------------------------
1228
1229
1230 @pytest.mark.anyio
1231 async def test_patch_repo_settings_updates_fields(
1232 client: AsyncClient,
1233 db_session: AsyncSession,
1234 auth_headers: StrDict,
1235 ) -> None:
1236 """PATCH /repos/{repo_id}/settings owner can update dedicated and flag fields."""
1237 repo = MusehubRepo(
1238 name="settings-patch-test",
1239 owner="testuser",
1240 slug="settings-patch-test",
1241 visibility="private",
1242 owner_user_id=TEST_OWNER_USER_ID,
1243 )
1244 db_session.add(repo)
1245 await db_session.commit()
1246 await db_session.refresh(repo)
1247
1248 resp = await client.patch(
1249 f"/api/repos/{repo.repo_id}/settings",
1250 json={
1251 "description": "Updated description",
1252 "visibility": "public",
1253 "hasIssues": False,
1254 "allowRebaseMerge": True,
1255 "homepageUrl": "https://muse.app",
1256 "topics": ["classical", "baroque"],
1257 },
1258 headers=auth_headers,
1259 )
1260 assert resp.status_code == 200
1261 body = resp.json()
1262 assert body["description"] == "Updated description"
1263 assert body["visibility"] == "public"
1264 assert body["hasIssues"] is False
1265 assert body["allowRebaseMerge"] is True
1266 assert body["homepageUrl"] == "https://muse.app"
1267 assert body["topics"] == ["classical", "baroque"]
1268 # Untouched field should retain its default
1269 assert body["allowMergeCommit"] is True
1270
1271
1272 @pytest.mark.anyio
1273 async def test_patch_repo_settings_partial_update_preserves_other_fields(
1274 client: AsyncClient,
1275 db_session: AsyncSession,
1276 auth_headers: StrDict,
1277 ) -> None:
1278 """PATCH with a single field leaves all other settings unchanged."""
1279 repo = MusehubRepo(
1280 name="settings-partial-test",
1281 owner="testuser",
1282 slug="settings-partial-test",
1283 visibility="private",
1284 owner_user_id=TEST_OWNER_USER_ID,
1285 )
1286 db_session.add(repo)
1287 await db_session.commit()
1288 await db_session.refresh(repo)
1289
1290 resp = await client.patch(
1291 f"/api/repos/{repo.repo_id}/settings",
1292 json={"defaultBranch": "develop"},
1293 headers=auth_headers,
1294 )
1295 assert resp.status_code == 200
1296 body = resp.json()
1297 assert body["defaultBranch"] == "develop"
1298 # Other fields kept
1299 assert body["name"] == "settings-partial-test"
1300 assert body["visibility"] == "private"
1301 assert body["hasIssues"] is True
1302
1303
1304 @pytest.mark.anyio
1305 async def test_patch_repo_settings_requires_auth(
1306 client: AsyncClient,
1307 db_session: AsyncSession,
1308 ) -> None:
1309 """PATCH /repos/{repo_id}/settings returns 401 without a MSign Authorization header."""
1310 repo = MusehubRepo(
1311 name="settings-patch-noauth",
1312 owner="testuser",
1313 slug="settings-patch-noauth",
1314 visibility="private",
1315 owner_user_id=TEST_OWNER_USER_ID,
1316 )
1317 db_session.add(repo)
1318 await db_session.commit()
1319 await db_session.refresh(repo)
1320
1321 resp = await client.patch(
1322 f"/api/repos/{repo.repo_id}/settings",
1323 json={"visibility": "public"},
1324 )
1325 assert resp.status_code == 401
1326
1327
1328 @pytest.mark.anyio
1329 async def test_patch_repo_settings_returns_403_for_non_admin(
1330 client: AsyncClient,
1331 db_session: AsyncSession,
1332 auth_headers: StrDict,
1333 ) -> None:
1334 """PATCH /repos/{repo_id}/settings returns 403 when caller is not owner or admin."""
1335 repo = MusehubRepo(
1336 name="settings-patch-403",
1337 owner="other-owner",
1338 slug="settings-patch-403",
1339 visibility="public",
1340 owner_user_id="other-user-not-test",
1341 )
1342 db_session.add(repo)
1343 await db_session.commit()
1344 await db_session.refresh(repo)
1345
1346 resp = await client.patch(
1347 f"/api/repos/{repo.repo_id}/settings",
1348 json={"hasWiki": True},
1349 headers=auth_headers,
1350 )
1351 assert resp.status_code == 403
1352
1353
1354 # ---------------------------------------------------------------------------
1355 # DELETE /repos/{repo_id} — soft-delete
1356 # ---------------------------------------------------------------------------
1357
1358
1359 @pytest.mark.anyio
1360 async def test_delete_repo_returns_204(
1361 client: AsyncClient,
1362 auth_headers: StrDict,
1363 ) -> None:
1364 """DELETE /repos/{repo_id} soft-deletes a repo owned by the caller and returns 204."""
1365 create = await client.post(
1366 "/api/repos",
1367 json={"name": "to-delete", "owner": "testuser", "visibility": "private"},
1368 headers=auth_headers,
1369 )
1370 assert create.status_code == 201
1371 repo_id = create.json()["repoId"]
1372
1373 resp = await client.delete(f"/api/repos/{repo_id}", headers=auth_headers)
1374 assert resp.status_code == 204
1375
1376
1377 @pytest.mark.anyio
1378 async def test_delete_repo_hides_repo_from_get(
1379 client: AsyncClient,
1380 auth_headers: StrDict,
1381 ) -> None:
1382 """After DELETE, GET /repos/{repo_id} returns 404."""
1383 create = await client.post(
1384 "/api/repos",
1385 json={"name": "hidden-after-delete", "owner": "testuser", "visibility": "private"},
1386 headers=auth_headers,
1387 )
1388 repo_id = create.json()["repoId"]
1389
1390 await client.delete(f"/api/repos/{repo_id}", headers=auth_headers)
1391
1392 get_resp = await client.get(f"/api/repos/{repo_id}", headers=auth_headers)
1393 assert get_resp.status_code == 404
1394
1395
1396 @pytest.mark.anyio
1397 async def test_delete_repo_requires_auth(
1398 client: AsyncClient,
1399 db_session: AsyncSession,
1400 ) -> None:
1401 """DELETE /repos/{repo_id} returns 401 without a MSign Authorization header."""
1402 repo = MusehubRepo(
1403 name="delete-noauth",
1404 owner="testuser",
1405 slug="delete-noauth",
1406 visibility="public",
1407 owner_user_id=TEST_OWNER_USER_ID,
1408 )
1409 db_session.add(repo)
1410 await db_session.commit()
1411 await db_session.refresh(repo)
1412
1413 resp = await client.delete(f"/api/repos/{repo.repo_id}")
1414 assert resp.status_code == 401
1415
1416
1417 @pytest.mark.anyio
1418 async def test_delete_repo_returns_403_for_non_owner(
1419 client: AsyncClient,
1420 db_session: AsyncSession,
1421 auth_headers: StrDict,
1422 ) -> None:
1423 """DELETE /repos/{repo_id} returns 403 when caller is not the owner."""
1424 repo = MusehubRepo(
1425 name="delete-403",
1426 owner="other-owner",
1427 slug="delete-403",
1428 visibility="public",
1429 owner_user_id="some-other-user-id",
1430 )
1431 db_session.add(repo)
1432 await db_session.commit()
1433 await db_session.refresh(repo)
1434
1435 resp = await client.delete(
1436 f"/api/repos/{repo.repo_id}", headers=auth_headers
1437 )
1438 assert resp.status_code == 403
1439
1440
1441 @pytest.mark.anyio
1442 async def test_delete_repo_returns_404_for_unknown_repo(
1443 client: AsyncClient,
1444 auth_headers: StrDict,
1445 ) -> None:
1446 """DELETE /repos/{repo_id} returns 404 for a non-existent repo."""
1447 resp = await client.delete(
1448 "/api/repos/nonexistent-repo-id", headers=auth_headers
1449 )
1450 assert resp.status_code == 404
1451
1452
1453 @pytest.mark.anyio
1454 async def test_delete_repo_service_sets_deleted_at(
1455 db_session: AsyncSession,
1456 ) -> None:
1457 """delete_repo() service sets deleted_at on the row."""
1458 repo = await musehub_repository.create_repo(
1459 db_session,
1460 name="svc-delete-test",
1461 owner="testuser",
1462 visibility="private",
1463 owner_user_id="user-abc",
1464 )
1465 await db_session.commit()
1466
1467 deleted = await musehub_repository.delete_repo(db_session, repo.repo_id)
1468 await db_session.commit()
1469
1470 assert deleted is True
1471 # get_repo should return None for soft-deleted repos
1472 fetched = await musehub_repository.get_repo(db_session, repo.repo_id)
1473 assert fetched is None
1474
1475
1476 @pytest.mark.anyio
1477 async def test_delete_repo_service_returns_false_for_unknown(
1478 db_session: AsyncSession,
1479 ) -> None:
1480 """delete_repo() returns False for a non-existent repo."""
1481 result = await musehub_repository.delete_repo(db_session, "does-not-exist")
1482 assert result is False
1483
1484
1485 # ---------------------------------------------------------------------------
1486 # POST /repos/{repo_id}/transfer — transfer ownership
1487 # ---------------------------------------------------------------------------
1488
1489
1490 @pytest.mark.anyio
1491 async def test_transfer_repo_ownership_returns_200(
1492 client: AsyncClient,
1493 auth_headers: StrDict,
1494 ) -> None:
1495 """POST /repos/{repo_id}/transfer returns 200 with updated ownerUserId."""
1496 create = await client.post(
1497 "/api/repos",
1498 json={"name": "transfer-me", "owner": "testuser", "visibility": "private"},
1499 headers=auth_headers,
1500 )
1501 assert create.status_code == 201
1502 repo_id = create.json()["repoId"]
1503 new_owner = "another-user-uuid-1234"
1504
1505 resp = await client.post(
1506 f"/api/repos/{repo_id}/transfer",
1507 json={"newOwnerUserId": new_owner},
1508 headers=auth_headers,
1509 )
1510 assert resp.status_code == 200
1511 body = resp.json()
1512 assert body["ownerUserId"] == new_owner
1513 assert body["repoId"] == repo_id
1514
1515
1516 @pytest.mark.anyio
1517 async def test_transfer_repo_requires_auth(
1518 client: AsyncClient,
1519 db_session: AsyncSession,
1520 ) -> None:
1521 """POST /repos/{repo_id}/transfer returns 401 without an MSign token."""
1522 repo = MusehubRepo(
1523 name="transfer-noauth",
1524 owner="testuser",
1525 slug="transfer-noauth",
1526 visibility="public",
1527 owner_user_id=TEST_OWNER_USER_ID,
1528 )
1529 db_session.add(repo)
1530 await db_session.commit()
1531 await db_session.refresh(repo)
1532
1533 resp = await client.post(
1534 f"/api/repos/{repo.repo_id}/transfer",
1535 json={"newOwnerUserId": "new-user-id"},
1536 )
1537 assert resp.status_code == 401
1538
1539
1540 # ---------------------------------------------------------------------------
1541 # Wizard creation endpoint — # ---------------------------------------------------------------------------
1542
1543
1544 @pytest.mark.anyio
1545 async def test_create_repo_wizard_initialize_creates_branch_and_commit(
1546 client: AsyncClient,
1547 auth_headers: StrDict,
1548 db_session: AsyncSession,
1549 ) -> None:
1550 """POST /repos with initialize=true creates a default branch + initial commit."""
1551 resp = await client.post(
1552 "/api/repos",
1553 json={
1554 "name": "wizard-init-repo",
1555 "owner": "testuser",
1556 "visibility": "public",
1557 "initialize": True,
1558 "defaultBranch": "main",
1559 },
1560 headers=auth_headers,
1561 )
1562 assert resp.status_code == 201
1563 repo_id = resp.json()["repoId"]
1564
1565 branches_resp = await client.get(
1566 f"/api/repos/{repo_id}/branches",
1567 headers=auth_headers,
1568 )
1569 assert branches_resp.status_code == 200
1570 branches = branches_resp.json()["branches"]
1571 assert any(b["name"] == "main" for b in branches), "Expected 'main' branch to be created"
1572
1573 commits_resp = await client.get(
1574 f"/api/repos/{repo_id}/commits",
1575 headers=auth_headers,
1576 )
1577 assert commits_resp.status_code == 200
1578 commits = commits_resp.json()["commits"]
1579 assert len(commits) == 1
1580 assert commits[0]["message"] == "Initial commit"
1581
1582
1583 @pytest.mark.anyio
1584 async def test_create_repo_wizard_no_initialize_stays_empty(
1585 client: AsyncClient,
1586 auth_headers: StrDict,
1587 ) -> None:
1588 """POST /repos with initialize=false leaves branches and commits empty."""
1589 resp = await client.post(
1590 "/api/repos",
1591 json={
1592 "name": "wizard-noinit-repo",
1593 "owner": "testuser",
1594 "initialize": False,
1595 },
1596 headers=auth_headers,
1597 )
1598 assert resp.status_code == 201
1599 repo_id = resp.json()["repoId"]
1600
1601 branches_resp = await client.get(
1602 f"/api/repos/{repo_id}/branches",
1603 headers=auth_headers,
1604 )
1605 assert branches_resp.json()["branches"] == []
1606
1607 commits_resp = await client.get(
1608 f"/api/repos/{repo_id}/commits",
1609 headers=auth_headers,
1610 )
1611 assert commits_resp.json()["commits"] == []
1612
1613
1614 @pytest.mark.anyio
1615 async def test_create_repo_wizard_topics_merged_into_tags(
1616 client: AsyncClient,
1617 auth_headers: StrDict,
1618 ) -> None:
1619 """POST /repos with topics merges them into the tag list (deduplicated)."""
1620 resp = await client.post(
1621 "/api/repos",
1622 json={
1623 "name": "topics-test-repo",
1624 "owner": "testuser",
1625 "tags": ["jazz"],
1626 "topics": ["classical", "jazz"], # 'jazz' deduped
1627 "initialize": False,
1628 },
1629 headers=auth_headers,
1630 )
1631 assert resp.status_code == 201
1632 body = resp.json()
1633 tags: list[str] = body["tags"]
1634 assert "jazz" in tags
1635 assert "classical" in tags
1636 assert tags.count("jazz") == 1, "Duplicate 'jazz' must be removed"
1637
1638
1639 @pytest.mark.anyio
1640 async def test_create_repo_wizard_clone_url_uses_musehub_scheme(
1641 client: AsyncClient,
1642 auth_headers: StrDict,
1643 ) -> None:
1644 """Clone URL returned by POST /repos uses the musehub:// protocol scheme."""
1645 resp = await client.post(
1646 "/api/repos",
1647 json={"name": "clone-url-test", "owner": "testuser", "initialize": False},
1648 headers=auth_headers,
1649 )
1650 assert resp.status_code == 201
1651 clone_url: str = resp.json()["cloneUrl"]
1652 assert clone_url.startswith("musehub://"), f"Expected musehub:// prefix, got: {clone_url}"
1653 assert "testuser" in clone_url
1654
1655
1656 @pytest.mark.anyio
1657 async def test_create_repo_wizard_template_copies_description(
1658 client: AsyncClient,
1659 auth_headers: StrDict,
1660 db_session: AsyncSession,
1661 ) -> None:
1662 """POST /repos with template_repo_id copies description from a public template."""
1663 template = MusehubRepo(
1664 name="template-source",
1665 owner="template-owner",
1666 slug="template-source",
1667 visibility="public",
1668 owner_user_id="template-owner-id",
1669 description="A great neo-baroque composition template",
1670 tags=["baroque", "piano"],
1671 )
1672 db_session.add(template)
1673 await db_session.commit()
1674 await db_session.refresh(template)
1675 template_id = str(template.repo_id)
1676
1677 resp = await client.post(
1678 "/api/repos",
1679 json={
1680 "name": "from-template-repo",
1681 "owner": "testuser",
1682 "initialize": False,
1683 "templateRepoId": template_id,
1684 },
1685 headers=auth_headers,
1686 )
1687 assert resp.status_code == 201
1688 body = resp.json()
1689 assert body["description"] == "A great neo-baroque composition template"
1690 assert "baroque" in body["tags"]
1691 assert "piano" in body["tags"]
1692
1693
1694 @pytest.mark.anyio
1695 async def test_create_repo_wizard_private_template_not_copied(
1696 client: AsyncClient,
1697 auth_headers: StrDict,
1698 db_session: AsyncSession,
1699 ) -> None:
1700 """Private template repo metadata is NOT copied (must be public)."""
1701 private_template = MusehubRepo(
1702 name="private-template",
1703 owner="secret-owner",
1704 slug="private-template",
1705 visibility="private",
1706 owner_user_id="secret-id",
1707 description="Secret description",
1708 tags=["secret"],
1709 )
1710 db_session.add(private_template)
1711 await db_session.commit()
1712 await db_session.refresh(private_template)
1713 template_id = str(private_template.repo_id)
1714
1715 resp = await client.post(
1716 "/api/repos",
1717 json={
1718 "name": "refused-template-repo",
1719 "owner": "testuser",
1720 "description": "My own description",
1721 "initialize": False,
1722 "templateRepoId": template_id,
1723 },
1724 headers=auth_headers,
1725 )
1726 assert resp.status_code == 201
1727 body = resp.json()
1728 # Private template must not override user's own description
1729 assert body["description"] == "My own description"
1730 assert "secret" not in body["tags"]
1731
1732
1733 @pytest.mark.anyio
1734 async def test_create_repo_wizard_custom_default_branch(
1735 client: AsyncClient,
1736 auth_headers: StrDict,
1737 ) -> None:
1738 """POST /repos with initialize=true and custom defaultBranch creates the right branch."""
1739 resp = await client.post(
1740 "/api/repos",
1741 json={
1742 "name": "custom-branch-repo",
1743 "owner": "testuser",
1744 "initialize": True,
1745 "defaultBranch": "develop",
1746 },
1747 headers=auth_headers,
1748 )
1749 assert resp.status_code == 201
1750 repo_id = resp.json()["repoId"]
1751
1752 branches_resp = await client.get(
1753 f"/api/repos/{repo_id}/branches",
1754 headers=auth_headers,
1755 )
1756 branch_names = [b["name"] for b in branches_resp.json()["branches"]]
1757 assert "develop" in branch_names
1758 assert "main" not in branch_names
1759
1760
1761 # ---------------------------------------------------------------------------
1762 # GET /repos — list repos for authenticated user
1763 # ---------------------------------------------------------------------------
1764
1765
1766 @pytest.mark.anyio
1767 async def test_list_my_repos_returns_owned_repos(
1768 client: AsyncClient,
1769 auth_headers: StrDict,
1770 ) -> None:
1771 """GET /repos returns repos created by the authenticated user."""
1772 # Create two repos
1773 for name in ("owned-repo-a", "owned-repo-b"):
1774 await client.post(
1775 "/api/repos",
1776 json={"name": name, "owner": "testuser", "initialize": False},
1777 headers=auth_headers,
1778 )
1779
1780 resp = await client.get("/api/repos", headers=auth_headers)
1781 assert resp.status_code == 200
1782 body = resp.json()
1783 assert "repos" in body
1784 assert "total" in body
1785 assert "nextCursor" in body
1786 names = [r["name"] for r in body["repos"]]
1787 assert "owned-repo-a" in names
1788 assert "owned-repo-b" in names
1789
1790
1791 @pytest.mark.anyio
1792 async def test_list_my_repos_requires_auth(client: AsyncClient) -> None:
1793 """GET /repos returns 401 without an MSign token."""
1794 resp = await client.get("/api/repos")
1795 assert resp.status_code == 401
1796
1797
1798 @pytest.mark.anyio
1799 async def test_transfer_repo_returns_403_for_non_owner(
1800 client: AsyncClient,
1801 db_session: AsyncSession,
1802 auth_headers: StrDict,
1803 ) -> None:
1804 """POST /repos/{repo_id}/transfer returns 403 when caller is not the owner."""
1805 repo = MusehubRepo(
1806 name="transfer-403",
1807 owner="other-owner",
1808 slug="transfer-403",
1809 visibility="public",
1810 owner_user_id="some-other-user-id",
1811 )
1812 db_session.add(repo)
1813 await db_session.commit()
1814 await db_session.refresh(repo)
1815
1816 resp = await client.post(
1817 f"/api/repos/{repo.repo_id}/transfer",
1818 json={"newOwnerUserId": "attacker-user-id"},
1819 headers=auth_headers,
1820 )
1821 assert resp.status_code == 403
1822
1823
1824 @pytest.mark.anyio
1825 async def test_transfer_repo_returns_404_for_unknown_repo(
1826 client: AsyncClient,
1827 auth_headers: StrDict,
1828 ) -> None:
1829 """POST /repos/{repo_id}/transfer returns 404 for a non-existent repo."""
1830 resp = await client.post(
1831 "/api/repos/nonexistent-repo-id/transfer",
1832 json={"newOwnerUserId": "some-user"},
1833 headers=auth_headers,
1834 )
1835 assert resp.status_code == 404
1836
1837
1838 @pytest.mark.anyio
1839 async def test_transfer_repo_service_updates_owner_user_id(
1840 db_session: AsyncSession,
1841 ) -> None:
1842 """transfer_repo_ownership() service updates owner_user_id on the row."""
1843 repo = await musehub_repository.create_repo(
1844 db_session,
1845 name="svc-transfer-test",
1846 owner="testuser",
1847 visibility="private",
1848 owner_user_id="original-owner-id",
1849 )
1850 await db_session.commit()
1851
1852 updated = await musehub_repository.transfer_repo_ownership(
1853 db_session, repo.repo_id, "new-owner-id"
1854 )
1855 await db_session.commit()
1856
1857 assert updated is not None
1858 assert updated.owner_user_id == "new-owner-id"
1859 # Verify persisted
1860 fetched = await musehub_repository.get_repo(db_session, repo.repo_id)
1861 assert fetched is not None
1862 assert fetched.owner_user_id == "new-owner-id"
1863
1864
1865 @pytest.mark.anyio
1866 async def test_transfer_repo_service_returns_none_for_unknown(
1867 db_session: AsyncSession,
1868 ) -> None:
1869 """transfer_repo_ownership() returns None for a non-existent repo."""
1870 result = await musehub_repository.transfer_repo_ownership(
1871 db_session, "does-not-exist", "new-owner"
1872 )
1873 assert result is None
1874
1875
1876 # ---------------------------------------------------------------------------
1877 # GET /repos — list repos for authenticated user
1878 # ---------------------------------------------------------------------------
1879
1880
1881 @pytest.mark.anyio
1882 async def test_list_my_repos_total_matches_count(
1883 client: AsyncClient,
1884 auth_headers: StrDict,
1885 ) -> None:
1886 """total field in GET /repos matches the number of repos created."""
1887 initial = await client.get("/api/repos", headers=auth_headers)
1888 initial_total: int = initial.json()["total"]
1889
1890 await client.post(
1891 "/api/repos",
1892 json={"name": "total-count-test", "owner": "testuser", "initialize": False},
1893 headers=auth_headers,
1894 )
1895
1896 resp = await client.get("/api/repos", headers=auth_headers)
1897 assert resp.status_code == 200
1898 assert resp.json()["total"] == initial_total + 1
1899
1900
1901 @pytest.mark.anyio
1902 async def test_list_my_repos_pagination_cursor(
1903 client: AsyncClient,
1904 auth_headers: StrDict,
1905 db_session: AsyncSession,
1906 ) -> None:
1907 """GET /repos with limit=1 returns a nextCursor that fetches the next page."""
1908 from datetime import datetime, timedelta, timezone
1909
1910 owner_user_id = "550e8400-e29b-41d4-a716-446655440000"
1911 now = datetime.now(tz=timezone.utc)
1912 for i in range(3):
1913 repo = MusehubRepo(
1914 name=f"paged-repo-{i}",
1915 owner="testuser",
1916 slug=f"paged-repo-{i}",
1917 visibility="public",
1918 owner_user_id=owner_user_id,
1919 )
1920 repo.created_at = now - timedelta(seconds=i)
1921 db_session.add(repo)
1922 await db_session.commit()
1923
1924 first_page = await client.get(
1925 "/api/repos?limit=1",
1926 headers=auth_headers,
1927 )
1928 assert first_page.status_code == 200
1929 body = first_page.json()
1930 assert len(body["repos"]) == 1
1931 next_cursor = body["nextCursor"]
1932 assert next_cursor is not None
1933
1934 second_page = await client.get(
1935 f"/api/repos?limit=1&cursor={next_cursor}",
1936 headers=auth_headers,
1937 )
1938 assert second_page.status_code == 200
1939 second_body = second_page.json()
1940 assert len(second_body["repos"]) == 1
1941 # Pages must not overlap
1942 first_id = body["repos"][0]["repoId"]
1943 second_id = second_body["repos"][0]["repoId"]
1944 assert first_id != second_id
1945
1946
1947 @pytest.mark.anyio
1948 async def test_list_my_repos_service_direct(db_session: AsyncSession) -> None:
1949 """list_repos_for_user() returns only repos owned by the given user."""
1950 from musehub.services.musehub_repository import list_repos_for_user
1951
1952 owner_handle = "user-list-direct"
1953 other_handle = "user-other-direct"
1954
1955 repo_mine = MusehubRepo(
1956 name="mine-direct",
1957 owner=owner_handle,
1958 slug="mine-direct",
1959 visibility="private",
1960 owner_user_id="uid-list-direct",
1961 )
1962 repo_other = MusehubRepo(
1963 name="not-mine-direct",
1964 owner=other_handle,
1965 slug="not-mine-direct",
1966 visibility="private",
1967 owner_user_id="uid-other-direct",
1968 )
1969 db_session.add_all([repo_mine, repo_other])
1970 await db_session.commit()
1971
1972 result = await list_repos_for_user(db_session, owner_handle)
1973 repo_ids = {r.repo_id for r in result.repos}
1974 assert str(repo_mine.repo_id) in repo_ids
1975 assert str(repo_other.repo_id) not in repo_ids
1976
1977
1978 # ---------------------------------------------------------------------------
1979 # GET /repos/{repo_id}/collaborators/{username}/permission
1980 # ---------------------------------------------------------------------------
1981
1982
1983 @pytest.mark.anyio
1984 async def test_collab_access_owner_returns_owner_permission(
1985 client: AsyncClient,
1986 db_session: AsyncSession,
1987 auth_headers: StrDict,
1988 ) -> None:
1989 """Owner's username returns permission='owner' with accepted_at=null."""
1990 from musehub.db.musehub_collaborator_models import MusehubCollaborator
1991
1992 owner_id = TEST_OWNER_USER_ID
1993 repo = MusehubRepo(
1994 name="access-owner-test",
1995 owner="testuser",
1996 slug="access-owner-test",
1997 visibility="private",
1998 owner_user_id=owner_id,
1999 )
2000 db_session.add(repo)
2001 await db_session.commit()
2002 await db_session.refresh(repo)
2003
2004 resp = await client.get(
2005 f"/api/repos/{repo.repo_id}/collaborators/{owner_id}/permission",
2006 headers=auth_headers,
2007 )
2008 assert resp.status_code == 200
2009 body = resp.json()
2010 assert body["username"] == owner_id
2011 assert body["permission"] == "owner"
2012 assert body["acceptedAt"] is None
2013
2014
2015 @pytest.mark.anyio
2016 async def test_collab_access_collaborator_returns_permission(
2017 client: AsyncClient,
2018 db_session: AsyncSession,
2019 auth_headers: StrDict,
2020 ) -> None:
2021 """A known collaborator returns their permission level and accepted_at."""
2022 from datetime import datetime, timezone
2023
2024 from musehub.db.musehub_collaborator_models import MusehubCollaborator
2025
2026 owner_id = TEST_OWNER_USER_ID
2027 collab_user_id = "collab-user-write"
2028
2029 repo = MusehubRepo(
2030 name="access-collab-test",
2031 owner="testuser",
2032 slug="access-collab-test",
2033 visibility="private",
2034 owner_user_id=owner_id,
2035 )
2036 db_session.add(repo)
2037 await db_session.commit()
2038 await db_session.refresh(repo)
2039
2040 accepted = datetime(2026, 1, 10, 10, 0, 0, tzinfo=timezone.utc)
2041 collab = MusehubCollaborator(
2042 repo_id=str(repo.repo_id),
2043 identity_handle=collab_user_id,
2044 permission="write",
2045 accepted_at=accepted,
2046 )
2047 db_session.add(collab)
2048 await db_session.commit()
2049
2050 resp = await client.get(
2051 f"/api/repos/{repo.repo_id}/collaborators/{collab_user_id}/permission",
2052 headers=auth_headers,
2053 )
2054 assert resp.status_code == 200
2055 body = resp.json()
2056 assert body["username"] == collab_user_id
2057 assert body["permission"] == "write"
2058 assert body["acceptedAt"] is not None
2059
2060
2061 @pytest.mark.anyio
2062 async def test_collab_access_non_collaborator_returns_404(
2063 client: AsyncClient,
2064 db_session: AsyncSession,
2065 auth_headers: StrDict,
2066 ) -> None:
2067 """A user who is not a collaborator returns 404 with an informative message."""
2068 repo = MusehubRepo(
2069 name="access-404-test",
2070 owner="testuser",
2071 slug="access-404-test",
2072 visibility="private",
2073 owner_user_id=TEST_OWNER_USER_ID,
2074 )
2075 db_session.add(repo)
2076 await db_session.commit()
2077 await db_session.refresh(repo)
2078
2079 stranger = "total-stranger-user"
2080 resp = await client.get(
2081 f"/api/repos/{repo.repo_id}/collaborators/{stranger}/permission",
2082 headers=auth_headers,
2083 )
2084 assert resp.status_code == 404
2085 assert stranger in resp.json()["detail"]
2086
2087
2088 @pytest.mark.anyio
2089 async def test_collab_access_unknown_repo_returns_404(
2090 client: AsyncClient,
2091 auth_headers: StrDict,
2092 ) -> None:
2093 """Querying an unknown repo_id returns 404."""
2094 resp = await client.get(
2095 "/api/repos/nonexistent-repo/collaborators/anyone/permission",
2096 headers=auth_headers,
2097 )
2098 assert resp.status_code == 404
2099
2100
2101 @pytest.mark.anyio
2102 async def test_collab_access_requires_auth(
2103 client: AsyncClient,
2104 db_session: AsyncSession,
2105 ) -> None:
2106 """GET /collaborators/{username}/permission returns 401 without an MSign token."""
2107 repo = MusehubRepo(
2108 name="access-auth-test",
2109 owner="testuser",
2110 slug="access-auth-test",
2111 visibility="public",
2112 owner_user_id=TEST_OWNER_USER_ID,
2113 )
2114 db_session.add(repo)
2115 await db_session.commit()
2116 await db_session.refresh(repo)
2117
2118 resp = await client.get(
2119 f"/api/repos/{repo.repo_id}/collaborators/anyone/permission"
2120 )
2121 assert resp.status_code == 401
2122
2123
2124 @pytest.mark.anyio
2125 async def test_collab_access_admin_permission(
2126 client: AsyncClient,
2127 db_session: AsyncSession,
2128 auth_headers: StrDict,
2129 ) -> None:
2130 """A collaborator with admin permission returns permission='admin'."""
2131 from musehub.db.musehub_collaborator_models import MusehubCollaborator
2132
2133 repo = MusehubRepo(
2134 name="access-admin-test",
2135 owner="testuser",
2136 slug="access-admin-test",
2137 visibility="private",
2138 owner_user_id=TEST_OWNER_USER_ID,
2139 )
2140 db_session.add(repo)
2141 await db_session.commit()
2142 await db_session.refresh(repo)
2143
2144 admin_user = "admin-collab-user"
2145 collab = MusehubCollaborator(
2146 repo_id=str(repo.repo_id),
2147 identity_handle=admin_user,
2148 permission="admin",
2149 accepted_at=None,
2150 )
2151 db_session.add(collab)
2152 await db_session.commit()
2153
2154 resp = await client.get(
2155 f"/api/repos/{repo.repo_id}/collaborators/{admin_user}/permission",
2156 headers=auth_headers,
2157 )
2158 assert resp.status_code == 200
2159 body = resp.json()
2160 assert body["permission"] == "admin"
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago