test_musehub_auth.py
python
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a
init: musehub initial commit
Human
171 days ago
| 1 | """Auth guard tests for MuseHub 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 | from musehub.muse_contracts.json_types import StrDict |
| 23 | |
| 24 | |
| 25 | # --------------------------------------------------------------------------- |
| 26 | # Write endpoints — always require auth (401 without token) |
| 27 | # Parametrized: eliminates five near-identical test functions. |
| 28 | # --------------------------------------------------------------------------- |
| 29 | |
| 30 | @pytest.mark.anyio |
| 31 | @pytest.mark.parametrize("method,url,body", [ |
| 32 | # POST endpoints that require auth regardless of whether the repo exists |
| 33 | ("POST", "/api/repos", {"name": "beats", "owner": "testuser"}), |
| 34 | ("POST", "/api/repos/any-repo-id/issues", {"title": "Bug report"}), |
| 35 | ("POST", "/api/repos/any-repo-id/issues/1/close", {}), |
| 36 | ]) |
| 37 | async def test_write_endpoints_require_auth( |
| 38 | client: AsyncClient, |
| 39 | method: str, |
| 40 | url: str, |
| 41 | body: JSONObject, |
| 42 | ) -> None: |
| 43 | """Write endpoints return 401 when no MSign Authorization header is supplied.""" |
| 44 | fn = getattr(client, method.lower()) |
| 45 | response = await fn(url, json=body) |
| 46 | assert response.status_code == 401, ( |
| 47 | f"{method} {url} expected 401, got {response.status_code}: {response.text[:200]}" |
| 48 | ) |
| 49 | |
| 50 | |
| 51 | @pytest.mark.anyio |
| 52 | async def test_delete_webhook_requires_auth(client: AsyncClient, db_session: AsyncSession) -> None: |
| 53 | """DELETE /webhooks/{id} returns 401 without a token.""" |
| 54 | response = await client.delete("/api/repos/any-repo-id/webhooks/fake-hook-id") |
| 55 | assert response.status_code == 401 |
| 56 | |
| 57 | |
| 58 | # --------------------------------------------------------------------------- |
| 59 | # GET endpoints — non-existent repos return 404 (not 401) without token |
| 60 | # |
| 61 | # Rationale: optional_token + visibility guard — unauthenticated requests |
| 62 | # reach the DB; a non-existent repo returns 404 before the auth check fires. |
| 63 | # --------------------------------------------------------------------------- |
| 64 | |
| 65 | @pytest.mark.anyio |
| 66 | @pytest.mark.parametrize("url", [ |
| 67 | "/api/repos/non-existent-repo-id", |
| 68 | "/api/repos/non-existent-repo-id/branches", |
| 69 | "/api/repos/non-existent-repo-id/commits", |
| 70 | "/api/repos/non-existent-repo-id/issues", |
| 71 | "/api/repos/non-existent-repo-id/issues/1", |
| 72 | "/api/repos/non-existent-repo-id/pulls", |
| 73 | "/api/repos/non-existent-repo-id/releases", |
| 74 | ]) |
| 75 | async def test_get_nonexistent_repo_returns_404_without_auth( |
| 76 | client: AsyncClient, |
| 77 | url: str, |
| 78 | ) -> None: |
| 79 | """GET on a non-existent resource returns 404 without auth (not 401). |
| 80 | |
| 81 | The DB lookup happens before the visibility guard fires, so a missing |
| 82 | repo surfaces as 404 regardless of authentication status. |
| 83 | """ |
| 84 | response = await client.get(url) |
| 85 | assert response.status_code == 404, ( |
| 86 | f"GET {url} expected 404, got {response.status_code}" |
| 87 | ) |
| 88 | |
| 89 | |
| 90 | # --------------------------------------------------------------------------- |
| 91 | # Private repo visibility — GET returns 401 for private repos without token |
| 92 | # --------------------------------------------------------------------------- |
| 93 | |
| 94 | @pytest.mark.anyio |
| 95 | async def test_private_repo_returns_401_without_auth( |
| 96 | client: AsyncClient, |
| 97 | auth_headers: StrDict, |
| 98 | ) -> None: |
| 99 | """GET /repos/{id} returns 401 for a private repo without a token.""" |
| 100 | from musehub.auth.request_signing import optional_signed_request, require_signed_request |
| 101 | from musehub.main import app as _app |
| 102 | |
| 103 | create_resp = await client.post( |
| 104 | "/api/repos", |
| 105 | json={"name": "private-auth-test", "owner": "authtest", "visibility": "private"}, |
| 106 | headers=auth_headers, |
| 107 | ) |
| 108 | assert create_resp.status_code == 201 |
| 109 | repo_id = create_resp.json()["repoId"] |
| 110 | |
| 111 | # Temporarily remove auth overrides to simulate unauthenticated request |
| 112 | _app.dependency_overrides.pop(require_signed_request, None) |
| 113 | _app.dependency_overrides.pop(optional_signed_request, None) |
| 114 | unauth_resp = await client.get(f"/api/repos/{repo_id}") |
| 115 | assert unauth_resp.status_code == 401, ( |
| 116 | f"Expected 401 for private repo, got {unauth_resp.status_code}" |
| 117 | ) |
| 118 | |
| 119 | |
| 120 | @pytest.mark.anyio |
| 121 | async def test_public_repo_accessible_without_auth( |
| 122 | client: AsyncClient, |
| 123 | auth_headers: StrDict, |
| 124 | ) -> None: |
| 125 | """GET /repos/{id} returns 200 for a public repo without a token.""" |
| 126 | from musehub.auth.request_signing import optional_signed_request, require_signed_request |
| 127 | from musehub.main import app as _app |
| 128 | |
| 129 | create_resp = await client.post( |
| 130 | "/api/repos", |
| 131 | json={"name": "public-auth-test", "owner": "authtest", "visibility": "public"}, |
| 132 | headers=auth_headers, |
| 133 | ) |
| 134 | assert create_resp.status_code == 201 |
| 135 | repo_id = create_resp.json()["repoId"] |
| 136 | |
| 137 | # Temporarily remove auth overrides to simulate unauthenticated request |
| 138 | _app.dependency_overrides.pop(require_signed_request, None) |
| 139 | _app.dependency_overrides.pop(optional_signed_request, None) |
| 140 | unauth_resp = await client.get(f"/api/repos/{repo_id}") |
| 141 | assert unauth_resp.status_code == 200, ( |
| 142 | f"Expected 200 for public repo, got {unauth_resp.status_code}: {unauth_resp.text}" |
| 143 | ) |
| 144 | # Body should contain the repo data |
| 145 | body = unauth_resp.json() |
| 146 | assert body["repoId"] == repo_id |
| 147 | assert body["visibility"] == "public" |
| 148 | |
| 149 | |
| 150 | # --------------------------------------------------------------------------- |
| 151 | # Authenticated requests are accepted |
| 152 | # --------------------------------------------------------------------------- |
| 153 | |
| 154 | @pytest.mark.anyio |
| 155 | async def test_hub_routes_accept_valid_token( |
| 156 | client: AsyncClient, |
| 157 | auth_headers: StrDict, |
| 158 | ) -> None: |
| 159 | """POST /musehub/repos succeeds (201) with a valid MSign auth header.""" |
| 160 | response = await client.post( |
| 161 | "/api/repos", |
| 162 | json={"name": "auth-sanity-repo", "owner": "testuser"}, |
| 163 | headers=auth_headers, |
| 164 | ) |
| 165 | assert response.status_code == 201 |
| 166 | body = response.json() |
| 167 | assert body["name"] == "auth-sanity-repo" |
| 168 | assert body["owner"] == "testuser" |
| 169 | assert "repoId" in body |
File History
1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a
init: musehub initial commit
Human
171 days ago