test_musehub_auth.py
python
| 1 | """Auth guard tests for Muse Hub routes. |
| 2 | |
| 3 | Auth model (updated in Phase 0–4 UX overhaul): |
| 4 | - GET endpoints use ``optional_token`` — public repos are accessible |
| 5 | unauthenticated; private repos return 401. |
| 6 | - POST / DELETE / write endpoints always use ``require_valid_token``. |
| 7 | - Non-existent repos return 404 regardless of auth status (no auth |
| 8 | pre-filter that exposes 401 before a DB lookup for GET routes). |
| 9 | |
| 10 | Covers: |
| 11 | - Write endpoints (POST/DELETE) always return 401 without a token. |
| 12 | - GET endpoints return 404 (not 401) for non-existent repos without a token, |
| 13 | because the auth check is deferred to the visibility guard. |
| 14 | - GET endpoints return 401 for real private repos without a token. |
| 15 | - Valid tokens are accepted on write endpoints. |
| 16 | """ |
| 17 | from __future__ import annotations |
| 18 | |
| 19 | import pytest |
| 20 | from httpx import AsyncClient |
| 21 | from sqlalchemy.ext.asyncio import AsyncSession |
| 22 | |
| 23 | |
| 24 | # --------------------------------------------------------------------------- |
| 25 | # Write endpoints — always require auth (401 without token) |
| 26 | # Parametrized: eliminates five near-identical test functions. |
| 27 | # --------------------------------------------------------------------------- |
| 28 | |
| 29 | @pytest.mark.anyio |
| 30 | @pytest.mark.parametrize("method,url,body", [ |
| 31 | # POST endpoints that require auth regardless of whether the repo exists |
| 32 | ("POST", "/api/v1/musehub/repos", {"name": "beats", "owner": "testuser"}), |
| 33 | ("POST", "/api/v1/musehub/repos/any-repo-id/issues", {"title": "Bug report"}), |
| 34 | ("POST", "/api/v1/musehub/repos/any-repo-id/issues/1/close", {}), |
| 35 | ]) |
| 36 | async def test_write_endpoints_require_auth( |
| 37 | client: AsyncClient, |
| 38 | method: str, |
| 39 | url: str, |
| 40 | body: dict, |
| 41 | ) -> None: |
| 42 | """Write endpoints return 401 when no Bearer token is supplied.""" |
| 43 | fn = getattr(client, method.lower()) |
| 44 | response = await fn(url, json=body) |
| 45 | assert response.status_code == 401, ( |
| 46 | f"{method} {url} expected 401, got {response.status_code}: {response.text[:200]}" |
| 47 | ) |
| 48 | |
| 49 | |
| 50 | @pytest.mark.anyio |
| 51 | async def test_delete_webhook_requires_auth(client: AsyncClient, db_session: AsyncSession) -> None: |
| 52 | """DELETE /webhooks/{id} returns 401 without a token.""" |
| 53 | response = await client.delete("/api/v1/musehub/repos/any-repo-id/webhooks/fake-hook-id") |
| 54 | assert response.status_code == 401 |
| 55 | |
| 56 | |
| 57 | # --------------------------------------------------------------------------- |
| 58 | # GET endpoints — non-existent repos return 404 (not 401) without token |
| 59 | # |
| 60 | # Rationale: optional_token + visibility guard — unauthenticated requests |
| 61 | # reach the DB; a non-existent repo returns 404 before the auth check fires. |
| 62 | # --------------------------------------------------------------------------- |
| 63 | |
| 64 | @pytest.mark.anyio |
| 65 | @pytest.mark.parametrize("url", [ |
| 66 | "/api/v1/musehub/repos/non-existent-repo-id", |
| 67 | "/api/v1/musehub/repos/non-existent-repo-id/branches", |
| 68 | "/api/v1/musehub/repos/non-existent-repo-id/commits", |
| 69 | "/api/v1/musehub/repos/non-existent-repo-id/issues", |
| 70 | "/api/v1/musehub/repos/non-existent-repo-id/issues/1", |
| 71 | "/api/v1/musehub/repos/non-existent-repo-id/pulls", |
| 72 | "/api/v1/musehub/repos/non-existent-repo-id/releases", |
| 73 | ]) |
| 74 | async def test_get_nonexistent_repo_returns_404_without_auth( |
| 75 | client: AsyncClient, |
| 76 | url: str, |
| 77 | ) -> None: |
| 78 | """GET on a non-existent resource returns 404 without auth (not 401). |
| 79 | |
| 80 | The DB lookup happens before the visibility guard fires, so a missing |
| 81 | repo surfaces as 404 regardless of authentication status. |
| 82 | """ |
| 83 | response = await client.get(url) |
| 84 | assert response.status_code == 404, ( |
| 85 | f"GET {url} expected 404, got {response.status_code}" |
| 86 | ) |
| 87 | |
| 88 | |
| 89 | # --------------------------------------------------------------------------- |
| 90 | # Private repo visibility — GET returns 401 for private repos without token |
| 91 | # --------------------------------------------------------------------------- |
| 92 | |
| 93 | @pytest.mark.anyio |
| 94 | async def test_private_repo_returns_401_without_auth( |
| 95 | client: AsyncClient, |
| 96 | auth_headers: dict[str, str], |
| 97 | ) -> None: |
| 98 | """GET /musehub/repos/{id} returns 401 for a private repo without a token.""" |
| 99 | create_resp = await client.post( |
| 100 | "/api/v1/musehub/repos", |
| 101 | json={"name": "private-auth-test", "owner": "authtest", "visibility": "private"}, |
| 102 | headers=auth_headers, |
| 103 | ) |
| 104 | assert create_resp.status_code == 201 |
| 105 | repo_id = create_resp.json()["repoId"] |
| 106 | |
| 107 | unauth_resp = await client.get(f"/api/v1/musehub/repos/{repo_id}") |
| 108 | assert unauth_resp.status_code == 401, ( |
| 109 | f"Expected 401 for private repo, got {unauth_resp.status_code}" |
| 110 | ) |
| 111 | |
| 112 | |
| 113 | @pytest.mark.anyio |
| 114 | async def test_public_repo_accessible_without_auth( |
| 115 | client: AsyncClient, |
| 116 | auth_headers: dict[str, str], |
| 117 | ) -> None: |
| 118 | """GET /musehub/repos/{id} returns 200 for a public repo without a token.""" |
| 119 | create_resp = await client.post( |
| 120 | "/api/v1/musehub/repos", |
| 121 | json={"name": "public-auth-test", "owner": "authtest", "visibility": "public"}, |
| 122 | headers=auth_headers, |
| 123 | ) |
| 124 | assert create_resp.status_code == 201 |
| 125 | repo_id = create_resp.json()["repoId"] |
| 126 | |
| 127 | unauth_resp = await client.get(f"/api/v1/musehub/repos/{repo_id}") |
| 128 | assert unauth_resp.status_code == 200, ( |
| 129 | f"Expected 200 for public repo, got {unauth_resp.status_code}: {unauth_resp.text}" |
| 130 | ) |
| 131 | # Body should contain the repo data |
| 132 | body = unauth_resp.json() |
| 133 | assert body["repoId"] == repo_id |
| 134 | assert body["visibility"] == "public" |
| 135 | |
| 136 | |
| 137 | # --------------------------------------------------------------------------- |
| 138 | # Authenticated requests are accepted |
| 139 | # --------------------------------------------------------------------------- |
| 140 | |
| 141 | @pytest.mark.anyio |
| 142 | async def test_hub_routes_accept_valid_token( |
| 143 | client: AsyncClient, |
| 144 | auth_headers: dict[str, str], |
| 145 | ) -> None: |
| 146 | """POST /musehub/repos succeeds (201) with a valid Bearer token.""" |
| 147 | response = await client.post( |
| 148 | "/api/v1/musehub/repos", |
| 149 | json={"name": "auth-sanity-repo", "owner": "testuser"}, |
| 150 | headers=auth_headers, |
| 151 | ) |
| 152 | assert response.status_code == 201 |
| 153 | body = response.json() |
| 154 | assert body["name"] == "auth-sanity-repo" |
| 155 | assert body["owner"] == "testuser" |
| 156 | assert "repoId" in body |