"""Tests for MuseHub label management endpoints. Covers all acceptance criteria: - GET /repos/{repo_id}/labels — list labels (public) - POST /repos/{repo_id}/labels — create label (auth required) - PATCH /repos/{repo_id}/labels/{label_id} — update label (auth required) - DELETE /repos/{repo_id}/labels/{label_id} — delete label (auth required) - POST .../issues/{number}/labels — assign labels to issue (auth required) - DELETE .../issues/{number}/labels/{label_id} — remove label from issue (auth required) - POST .../proposals/{proposal_id}/labels — assign labels to proposal (auth required) - DELETE .../proposals/{proposal_id}/labels/{label_id} — remove label from proposal (auth required) All tests use the shared ``client``, ``auth_headers``, and ``db_session`` fixtures from conftest.py. """ from __future__ import annotations from datetime import datetime, timezone import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import fake_id from musehub.core.genesis import compute_branch_id from musehub.db.musehub_repo_models import MusehubBranch, MusehubCommit, MusehubCommitRef from musehub.types.json_types import JSONObject, StrDict # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- async def _create_repo(client: AsyncClient, auth_headers: StrDict, name: str = "label-test-repo") -> str: """Create a repo and return its repo_id.""" response = await client.post( "/api/repos", json={"name": name, "owner": "testuser", "initialize": False}, headers=auth_headers, ) assert response.status_code == 201 repo_id: str = response.json()["repoId"] return repo_id async def _create_label( client: AsyncClient, auth_headers: StrDict, repo_id: str, name: str = "test-label", color: str = "#112233", description: str | None = "A test label", ) -> JSONObject: """Create a label and return the response body.""" payload = {"name": name, "color": color} if description is not None: payload["description"] = description response = await client.post( f"/api/repos/{repo_id}/labels", json=payload, headers=auth_headers, ) assert response.status_code == 201 label: JSONObject = response.json() return label async def _create_issue( client: AsyncClient, auth_headers: StrDict, repo_id: str, title: str = "Test issue", ) -> JSONObject: """Create an issue and return the response body.""" response = await client.post( f"/api/repos/{repo_id}/issues", json={"title": title, "body": "", "labels": []}, headers=auth_headers, ) assert response.status_code == 201 issue: JSONObject = response.json() return issue async def _push_branch(db: AsyncSession, repo_id: str, branch_name: str) -> str: """Insert a branch with one commit so the branch exists (required before creating a proposal).""" commit_id = fake_id(f"{repo_id}{branch_name}") commit = MusehubCommit( commit_id=commit_id, branch=branch_name, parent_ids=[], message=f"Initial commit on {branch_name}", author="testuser", timestamp=datetime.now(tz=timezone.utc), ) branch = MusehubBranch( branch_id=compute_branch_id(repo_id, branch_name), repo_id=repo_id, name=branch_name, head_commit_id=commit_id, ) db.add(commit) db.add(MusehubCommitRef(repo_id=repo_id, commit_id=commit_id)) db.add(branch) await db.commit() return commit_id async def _create_proposal( client: AsyncClient, auth_headers: StrDict, repo_id: str, title: str = "Test Proposal", ) -> JSONObject: """Create a proposal and return the response body.""" response = await client.post( f"/api/repos/{repo_id}/proposals", json={"title": title, "body": "", "fromBranch": "feature", "toBranch": "main"}, headers=auth_headers, ) assert response.status_code == 201, response.text proposal: JSONObject = response.json() return proposal # --------------------------------------------------------------------------- # POST /repos/{repo_id}/labels # --------------------------------------------------------------------------- async def test_create_label_returns_201( client: AsyncClient, auth_headers: StrDict, ) -> None: """POST /labels creates a label and returns 201 with the label data.""" repo_id = await _create_repo(client, auth_headers, "create-label-repo") label = await _create_label(client, auth_headers, repo_id) assert label["name"] == "test-label" assert label["color"] == "#112233" assert label["description"] == "A test label" assert "labelId" in label or "label_id" in label assert label.get("repoId") == repo_id or label.get("repo_id") == repo_id async def test_create_label_requires_auth( client: AsyncClient, ) -> None: """POST /labels without auth returns 401.""" response = await client.post( "/api/repos/nonexistent/labels", json={"name": "bug", "color": "#d73a4a"}, ) assert response.status_code == 401 async def test_create_label_unknown_repo_returns_404( client: AsyncClient, auth_headers: StrDict, ) -> None: """POST /labels for a non-existent repo returns 404.""" response = await client.post( "/api/repos/does-not-exist/labels", json={"name": "bug", "color": "#d73a4a"}, headers=auth_headers, ) assert response.status_code == 404 async def test_create_label_duplicate_name_returns_409( client: AsyncClient, auth_headers: StrDict, ) -> None: """POST /labels with a duplicate name returns 409 Conflict.""" repo_id = await _create_repo(client, auth_headers, "dupe-label-repo") # "bug" is seeded by default on repo creation — creating it again yields 409. response = await client.post( f"/api/repos/{repo_id}/labels", json={"name": "bug", "color": "#aabbcc"}, headers=auth_headers, ) assert response.status_code == 409 async def test_create_label_invalid_color_returns_422( client: AsyncClient, auth_headers: StrDict, ) -> None: """POST /labels with an invalid colour format returns 422.""" repo_id = await _create_repo(client, auth_headers, "color-invalid-repo") response = await client.post( f"/api/repos/{repo_id}/labels", json={"name": "bug", "color": "red"}, headers=auth_headers, ) assert response.status_code == 422 # --------------------------------------------------------------------------- # GET /repos/{repo_id}/labels # --------------------------------------------------------------------------- async def test_list_labels_public_access( client: AsyncClient, auth_headers: StrDict, ) -> None: """GET /labels is publicly accessible and returns all repo labels.""" repo_id = await _create_repo(client, auth_headers, "list-labels-repo") # Seeded defaults already include "bug" and "enhancement"; just add one extra. await _create_label(client, auth_headers, repo_id, name="custom-label", color="#123456") # No auth headers — public endpoint. response = await client.get(f"/api/repos/{repo_id}/labels") assert response.status_code == 200 body = response.json() assert "items" in body assert body["total"] > 0 names = [item["name"] for item in body["items"]] # Default-seeded labels must be present. assert "bug" in names assert "enhancement" in names # The extra label we created must also be there. assert "custom-label" in names async def test_list_labels_unknown_repo_returns_404( client: AsyncClient, ) -> None: """GET /labels for a non-existent repo returns 404.""" response = await client.get("/api/repos/no-such-repo/labels") assert response.status_code == 404 async def test_list_labels_empty_repo( client: AsyncClient, auth_headers: StrDict, ) -> None: """GET /labels for a new repo returns the seeded default labels.""" repo_id = await _create_repo(client, auth_headers, "empty-labels-repo") response = await client.get(f"/api/repos/{repo_id}/labels") assert response.status_code == 200 body = response.json() # Repos are seeded with default labels on creation — the list is never truly empty. assert isinstance(body["items"], list) assert body["total"] > 0 names = {lbl["name"] for lbl in body["items"]} assert "bug" in names # --------------------------------------------------------------------------- # PATCH /repos/{repo_id}/labels/{label_id} # --------------------------------------------------------------------------- async def test_update_label_name( client: AsyncClient, auth_headers: StrDict, ) -> None: """PATCH /labels/{id} updates the label name.""" repo_id = await _create_repo(client, auth_headers, "update-label-repo") label = await _create_label(client, auth_headers, repo_id, name="old-name", color="#aabbcc") label_id = label.get("label_id") or label.get("labelId") response = await client.patch( f"/api/repos/{repo_id}/labels/{label_id}", json={"name": "new-name"}, headers=auth_headers, ) assert response.status_code == 200 assert response.json()["name"] == "new-name" assert response.json()["color"] == "#aabbcc" async def test_update_label_requires_auth( client: AsyncClient, auth_headers: StrDict, ) -> None: """PATCH /labels/{id} without auth returns 401.""" from musehub.auth.request_signing import optional_signed_request, require_signed_request from musehub.main import app as _app repo_id = await _create_repo(client, auth_headers, "update-auth-label-repo") label = await _create_label(client, auth_headers, repo_id) label_id = label.get("label_id") or label.get("labelId") _app.dependency_overrides.pop(require_signed_request, None) _app.dependency_overrides.pop(optional_signed_request, None) response = await client.patch( f"/api/repos/{repo_id}/labels/{label_id}", json={"name": "hacked"}, ) assert response.status_code == 401 async def test_update_label_not_found_returns_404( client: AsyncClient, auth_headers: StrDict, ) -> None: """PATCH /labels/{id} with an unknown label_id returns 404.""" repo_id = await _create_repo(client, auth_headers, "update-404-repo") response = await client.patch( f"/api/repos/{repo_id}/labels/00000000-0000-0000-0000-000000000000", json={"name": "ghost"}, headers=auth_headers, ) assert response.status_code == 404 # --------------------------------------------------------------------------- # DELETE /repos/{repo_id}/labels/{label_id} # --------------------------------------------------------------------------- async def test_delete_label_returns_204( client: AsyncClient, auth_headers: StrDict, ) -> None: """DELETE /labels/{id} removes the label and returns 204.""" repo_id = await _create_repo(client, auth_headers, "delete-label-repo") label = await _create_label(client, auth_headers, repo_id) label_id = label.get("label_id") or label.get("labelId") response = await client.delete( f"/api/repos/{repo_id}/labels/{label_id}", headers=auth_headers, ) assert response.status_code == 204 # Confirm the specific label is gone (seeded defaults remain). list_resp = await client.get(f"/api/repos/{repo_id}/labels") remaining_ids = {lbl.get("label_id") or lbl.get("labelId") for lbl in list_resp.json()["items"]} assert label_id not in remaining_ids async def test_delete_label_requires_auth( client: AsyncClient, auth_headers: StrDict, ) -> None: """DELETE /labels/{id} without auth returns 401.""" from musehub.auth.request_signing import optional_signed_request, require_signed_request from musehub.main import app as _app repo_id = await _create_repo(client, auth_headers, "delete-auth-repo") label = await _create_label(client, auth_headers, repo_id) label_id = label.get("label_id") or label.get("labelId") _app.dependency_overrides.pop(require_signed_request, None) _app.dependency_overrides.pop(optional_signed_request, None) response = await client.delete( f"/api/repos/{repo_id}/labels/{label_id}", ) assert response.status_code == 401 # --------------------------------------------------------------------------- # Issue label assignments # --------------------------------------------------------------------------- async def test_assign_labels_to_issue( client: AsyncClient, auth_headers: StrDict, ) -> None: """POST .../issues/{number}/labels assigns labels and returns the updated issue.""" repo_id = await _create_repo(client, auth_headers, "issue-label-assign-repo") # "bug" is seeded by default — no need to create it separately. issue = await _create_issue(client, auth_headers, repo_id) issue_number = issue["number"] response = await client.post( f"/api/repos/{repo_id}/issues/{issue_number}/labels", json={"labels": ["bug"]}, headers=auth_headers, ) assert response.status_code == 200 updated_issue = response.json() assert "bug" in updated_issue.get("labels", []) async def test_assign_labels_to_issue_idempotent( client: AsyncClient, auth_headers: StrDict, ) -> None: """Assigning the same label twice does not raise an error.""" repo_id = await _create_repo(client, auth_headers, "issue-label-idem-repo") # "bug" is seeded by default — no need to create it separately. issue = await _create_issue(client, auth_headers, repo_id) issue_number = issue["number"] for _ in range(2): response = await client.post( f"/api/repos/{repo_id}/issues/{issue_number}/labels", json={"labels": ["bug"]}, headers=auth_headers, ) assert response.status_code == 200 async def test_remove_label_from_issue( client: AsyncClient, auth_headers: StrDict, ) -> None: """DELETE .../issues/{number}/labels/{label_name} removes the association.""" repo_id = await _create_repo(client, auth_headers, "issue-label-remove-repo") # "bug" is seeded by default — no need to create it separately. issue = await _create_issue(client, auth_headers, repo_id) issue_number = issue["number"] # Assign first. await client.post( f"/api/repos/{repo_id}/issues/{issue_number}/labels", json={"labels": ["bug"]}, headers=auth_headers, ) # Then remove (by label name, returns updated issue with 200). response = await client.delete( f"/api/repos/{repo_id}/issues/{issue_number}/labels/bug", headers=auth_headers, ) assert response.status_code == 200 assert "bug" not in response.json().get("labels", []) async def test_remove_label_from_issue_unknown_issue_returns_404( client: AsyncClient, auth_headers: StrDict, ) -> None: """DELETE .../issues/{number}/labels/{label_id} for an unknown issue returns 404.""" repo_id = await _create_repo(client, auth_headers, "issue-label-404-repo") label = await _create_label(client, auth_headers, repo_id) label_id = label.get("label_id") or label.get("labelId") response = await client.delete( f"/api/repos/{repo_id}/issues/9999/labels/{label_id}", headers=auth_headers, ) assert response.status_code == 404 # --------------------------------------------------------------------------- # Proposal label assignments # --------------------------------------------------------------------------- async def test_assign_labels_to_proposal( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """POST .../proposals/{proposal_id}/labels assigns labels and returns them.""" repo_id = await _create_repo(client, auth_headers, "proposal-label-assign-repo") await _push_branch(db_session, repo_id, "main") await _push_branch(db_session, repo_id, "feature") # "enhancement" is seeded by default — look it up from the repo's label list. labels_resp = await client.get(f"/api/repos/{repo_id}/labels") enhancement = next(lbl for lbl in labels_resp.json()["items"] if lbl["name"] == "enhancement") label_id = enhancement.get("label_id") or enhancement.get("labelId") proposal = await _create_proposal(client, auth_headers, repo_id) proposal_id = proposal.get("proposalId") or proposal.get("proposal_id") response = await client.post( f"/api/repos/{repo_id}/proposals/{proposal_id}/labels", json={"label_ids": [label_id]}, headers=auth_headers, ) assert response.status_code == 200 assigned = response.json() assert len(assigned) == 1 assert assigned[0]["name"] == "enhancement" async def test_remove_label_from_proposal( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """DELETE .../proposals/{proposal_id}/labels/{label_id} removes the association.""" repo_id = await _create_repo(client, auth_headers, "proposal-label-remove-repo") await _push_branch(db_session, repo_id, "main") await _push_branch(db_session, repo_id, "feature") label = await _create_label(client, auth_headers, repo_id) label_id = label.get("label_id") or label.get("labelId") proposal = await _create_proposal(client, auth_headers, repo_id) proposal_id = proposal.get("proposalId") or proposal.get("proposal_id") # Assign first. await client.post( f"/api/repos/{repo_id}/proposals/{proposal_id}/labels", json={"label_ids": [label_id]}, headers=auth_headers, ) # Then remove — should be idempotent too. response = await client.delete( f"/api/repos/{repo_id}/proposals/{proposal_id}/labels/{label_id}", headers=auth_headers, ) assert response.status_code == 204 async def test_remove_label_from_proposal_unknown_returns_404( client: AsyncClient, auth_headers: StrDict, ) -> None: """DELETE .../proposals/{proposal_id}/labels/{label_id} for an unknown proposal returns 404.""" repo_id = await _create_repo(client, auth_headers, "proposal-label-404-repo") label = await _create_label(client, auth_headers, repo_id) label_id = label.get("label_id") or label.get("labelId") response = await client.delete( f"/api/repos/{repo_id}/proposals/00000000-0000-0000-0000-000000000000/labels/{label_id}", headers=auth_headers, ) assert response.status_code == 404 async def test_delete_label_cascades_to_issue_associations( client: AsyncClient, auth_headers: StrDict, ) -> None: """Deleting a label removes it from all issue associations (cascade).""" repo_id = await _create_repo(client, auth_headers, "cascade-delete-repo") label = await _create_label(client, auth_headers, repo_id) label_id = label.get("label_id") or label.get("labelId") issue = await _create_issue(client, auth_headers, repo_id) issue_number = issue["number"] await client.post( f"/api/repos/{repo_id}/issues/{issue_number}/labels", json={"label_ids": [label_id]}, headers=auth_headers, ) delete_resp = await client.delete( f"/api/repos/{repo_id}/labels/{label_id}", headers=auth_headers, ) assert delete_resp.status_code == 204 # The deleted label must not appear in the repo's label list (seeded defaults remain). list_resp = await client.get(f"/api/repos/{repo_id}/labels") remaining_ids = {lbl.get("label_id") or lbl.get("labelId") for lbl in list_resp.json()["items"]} assert label_id not in remaining_ids # ── Seed default labels ────────────────────────────────────────────────────── async def test_create_repo_seeds_default_labels( client: AsyncClient, auth_headers: StrDict, ) -> None: """Creating a repo must automatically seed the default label set.""" repo_resp = await client.post( "/api/repos", json={"name": "seed-test-repo", "owner": "testuser", "initialize": False}, headers=auth_headers, ) assert repo_resp.status_code == 201 repo_id: str = repo_resp.json()["repoId"] label_resp = await client.get(f"/api/repos/{repo_id}/labels") assert label_resp.status_code == 200 data = label_resp.json() assert data["total"] > 0 names = {lbl["name"] for lbl in data["items"]} # Standard VCS labels expected. assert "bug" in names assert "enhancement" in names assert "documentation" in names # Music-domain labels must NOT be present. assert "needs-arrangement" not in names assert "musical-theory" not in names async def test_create_label_forbidden_for_non_owner( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """POST /labels as a non-owner returns 403.""" from datetime import datetime, timezone from musehub.core.genesis import compute_identity_id, compute_repo_id from musehub.db.musehub_repo_models import MusehubRepo # Create a repo owned by someone other than "testuser". _created_at = datetime.now(tz=timezone.utc) _owner_id = compute_identity_id(b"other-owner") other_repo = MusehubRepo( repo_id=compute_repo_id(_owner_id, "other-owner-repo", "code", _created_at.isoformat()), name="other-owner-repo", owner="other-owner", slug="other-owner-repo", visibility="public", owner_user_id=_owner_id, created_at=_created_at, updated_at=_created_at, ) db_session.add(other_repo) await db_session.commit() response = await client.post( f"/api/repos/{other_repo.repo_id}/labels", json={"name": "bug", "color": "#d73a4a"}, headers=auth_headers, ) assert response.status_code == 403 async def test_delete_label_forbidden_for_non_owner( client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """DELETE /labels/{id} as a non-owner returns 403.""" from datetime import datetime, timezone from musehub.core.genesis import compute_identity_id, compute_repo_id from musehub.db.musehub_repo_models import MusehubRepo _created_at = datetime.now(tz=timezone.utc) _owner_id = compute_identity_id(b"other-owner") other_repo = MusehubRepo( repo_id=compute_repo_id(_owner_id, "other-owner-repo-del", "code", _created_at.isoformat()), name="other-owner-repo-del", owner="other-owner", slug="other-owner-repo-del", visibility="public", owner_user_id=_owner_id, created_at=_created_at, updated_at=_created_at, ) db_session.add(other_repo) await db_session.commit() response = await client.delete( f"/api/repos/{other_repo.repo_id}/labels/00000000-0000-0000-0000-000000000000", headers=auth_headers, ) assert response.status_code == 403 async def test_seed_default_labels_is_idempotent( client: AsyncClient, auth_headers: StrDict, ) -> None: """seed_default_labels must not create duplicates when called twice.""" from musehub.db.database import AsyncSessionLocal from musehub.api.routes.musehub.labels import seed_default_labels repo_resp = await client.post( "/api/repos", json={"name": "idempotent-seed-repo", "owner": "testuser", "initialize": False}, headers=auth_headers, ) assert repo_resp.status_code == 201 repo_id: str = repo_resp.json()["repoId"] # Call seed a second time — should not raise and should not add duplicates. async with AsyncSessionLocal() as session: await seed_default_labels(session, repo_id) await session.commit() label_resp = await client.get(f"/api/repos/{repo_id}/labels") data = label_resp.json() names = [lbl["name"] for lbl in data["items"]] # No duplicate names. assert len(names) == len(set(names))