"""Tests for the MuseHub collaborators/team management UI page (SSR). Covers — GET /{owner}/{repo_slug}/settings/collaborators Test index: - test_collaborators_settings_page_returns_200 GET the settings/collaborators page as the owner returns 200 HTML. - test_collaborators_settings_page_requires_auth The page is owner-only — an unauthenticated GET returns 401. - test_collaborators_settings_page_unknown_repo_404 Unknown owner/slug combination returns 404. - test_collaborators_settings_page_has_invite_form_htmx The page embeds the invite form with hx-post attribute. - test_collaborators_settings_page_has_permission_badges The page renders colour-coded permission badge CSS classes. - test_collaborators_settings_page_has_owner_crown_badge The page marks owner permission with a crown emoji (👑). - test_collaborators_settings_page_has_remove_button_htmx Each non-owner row has an hx-delete remove form. - test_collaborators_settings_json_response_empty ?format=json returns CollaboratorListResponse with empty list for new repo. - test_collaborators_settings_json_response_with_collaborators ?format=json returns collaborators seeded in the DB. - test_collaborators_settings_page_has_settings_tabs The page includes the settings tab navigation bar. - test_collaborators_settings_page_has_invite_form_fields The invite form contains user_id and permission input fields. """ from __future__ import annotations import secrets import pytest import pytest_asyncio from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from datetime import datetime, timezone from muse.core.types import now_utc_iso from musehub.core.genesis import compute_collaborator_id, compute_identity_id, compute_repo_id from musehub.db.musehub_collaborator_models import MusehubCollaborator from musehub.db.musehub_repo_models import MusehubRepo # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _OWNER = "testuser" _SLUG = "collab-test-repo" async def _make_repo(db_session: AsyncSession) -> str: """Seed a minimal repo for collaborator tests and return its repo_id.""" owner_id = compute_identity_id(_OWNER.encode()) created_at = datetime.now(tz=timezone.utc) repo = MusehubRepo( repo_id=compute_repo_id(owner_id, _SLUG, "code", created_at.isoformat()), name=_SLUG, owner=_OWNER, slug=_SLUG, visibility="private", owner_user_id=owner_id, created_at=created_at, updated_at=created_at, ) db_session.add(repo) await db_session.commit() await db_session.refresh(repo) return str(repo.repo_id) async def _add_collaborator( db_session: AsyncSession, repo_id: str, user_id: str, permission: str = "write", invited_by: str | None = None, ) -> MusehubCollaborator: """Seed a collaborator record and return it.""" identity_id = compute_identity_id(user_id.encode()) collab = MusehubCollaborator( id=compute_collaborator_id(repo_id, identity_id, now_utc_iso()), repo_id=repo_id, identity_handle=user_id, permission=permission, invited_by_handle=invited_by, ) db_session.add(collab) await db_session.commit() await db_session.refresh(collab) return collab # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- async def test_collaborators_settings_page_returns_200( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """GET /{owner}/{slug}/settings/collaborators returns 200 HTML.""" await _make_repo(db_session) resp = await client.get(f"/{_OWNER}/{_SLUG}/settings/collaborators") assert resp.status_code == 200 assert "text/html" in resp.headers["content-type"] async def test_collaborators_settings_page_requires_auth( client: AsyncClient, db_session: AsyncSession, ) -> None: """The settings/collaborators page is owner-only: an unauthenticated GET returns 401. The page was migrated to a server-side owner gate (require_valid_token + claims.handle == owner); it no longer renders a shell for anonymous callers. """ await _make_repo(db_session) resp = await client.get( f"/{_OWNER}/{_SLUG}/settings/collaborators", headers={}, # explicit: no Authorization header ) assert resp.status_code == 401 async def test_collaborators_settings_page_unknown_repo_404( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """Unknown repo (owner matches the caller, slug not found) returns 404. The owner gate (claims.handle == owner) fires before the repo lookup, so the 404 path is only reachable when the caller owns the namespace. """ resp = await client.get(f"/{_OWNER}/nonexistent-repo/settings/collaborators") assert resp.status_code == 404 async def test_collaborators_settings_page_has_invite_form_htmx( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """The page embeds the invite form with hx-post for HTMX submission.""" await _make_repo(db_session) resp = await client.get(f"/{_OWNER}/{_SLUG}/settings/collaborators") assert resp.status_code == 200 assert "hx-post" in resp.text async def test_collaborators_settings_page_has_permission_badges( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """The page renders colour-coded permission badge CSS classes server-side.""" await _make_repo(db_session) resp = await client.get(f"/{_OWNER}/{_SLUG}/settings/collaborators") assert resp.status_code == 200 body = resp.text assert "badge-perm-read" in body assert "badge-perm-write" in body assert "badge-perm-admin" in body assert "badge-perm-owner" in body async def test_collaborators_settings_page_has_owner_crown_badge( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """The page marks owner permission with a crown emoji (👑).""" await _make_repo(db_session) resp = await client.get(f"/{_OWNER}/{_SLUG}/settings/collaborators") assert resp.status_code == 200 assert "👑" in resp.text async def test_collaborators_settings_page_has_remove_button_htmx( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """Non-owner collaborator rows carry hx-delete on the remove form.""" repo_id = await _make_repo(db_session) await _add_collaborator(db_session, repo_id, user_id=f"collab-{secrets.token_hex(4)}", permission="write") resp = await client.get(f"/{_OWNER}/{_SLUG}/settings/collaborators") assert resp.status_code == 200 assert "hx-delete" in resp.text async def test_collaborators_settings_json_response_empty( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """?format=json returns CollaboratorListResponse with empty list for a new repo.""" await _make_repo(db_session) resp = await client.get( f"/{_OWNER}/{_SLUG}/settings/collaborators?format=json" ) assert resp.status_code == 200 data = resp.json() assert "collaborators" in data assert "total" in data assert data["total"] == 0 assert data["collaborators"] == [] async def test_collaborators_settings_json_response_with_collaborators( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """?format=json returns collaborators seeded in the DB.""" repo_id = await _make_repo(db_session) collab_uid = secrets.token_hex(16) await _add_collaborator( db_session, repo_id, user_id=collab_uid, permission="write", invited_by="owner-user-id" ) resp = await client.get( f"/{_OWNER}/{_SLUG}/settings/collaborators?format=json" ) assert resp.status_code == 200 data = resp.json() assert data["total"] == 1 assert len(data["collaborators"]) == 1 collab = data["collaborators"][0] # camelCase keys (Pydantic by_alias=True via negotiate_response) assert collab["handle"] == collab_uid assert collab["permission"] == "write" async def test_collaborators_settings_page_has_settings_tabs( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """The page includes the settings tab navigation bar.""" await _make_repo(db_session) resp = await client.get(f"/{_OWNER}/{_SLUG}/settings/collaborators") assert resp.status_code == 200 body = resp.text assert "settings-tabs" in body assert "Collaborators" in body async def test_collaborators_settings_page_has_invite_form_fields( client: AsyncClient, db_session: AsyncSession, auth_headers: dict[str, str], ) -> None: """The invite form has user_id and permission input fields rendered server-side.""" await _make_repo(db_session) resp = await client.get(f"/{_OWNER}/{_SLUG}/settings/collaborators") assert resp.status_code == 200 body = resp.text assert 'name="user_id"' in body assert 'name="permission"' in body