Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 35 additions & 2 deletions campus_python/auth/v1/clients.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,10 @@ def new(
name: str,
description: str,
is_public: bool = False,
redirect_uris: list[str] | None = None
redirect_uris: list[str] | None = None,
allowed_scopes: list[str] | None = None,
upstream_scopes: dict[str, list[str]] | None = None,
token_bridge: bool | None = None
) -> campus.model.Client:
"""Create a new client.

Expand All @@ -40,16 +43,28 @@ def new(
description: Client description
is_public: True for public clients (CLI/mobile apps) without secrets
redirect_uris: OAuth redirect URIs for public clients
allowed_scopes: Token-scope allowlist (fail-closed: an
empty list grants nothing)
upstream_scopes: Per-provider upstream scope map
(e.g. {"google": [...]})
token_bridge: Token bridge access flag (upstream token
release; rejected by the server for public clients)

Returns:
The created Client
"""
json_data = {
json_data: JsonDict = {
"name": name,
"description": description,
"is_public": is_public,
"redirect_uris": redirect_uris or []
}
if allowed_scopes is not None:
json_data["allowed_scopes"] = allowed_scopes
if upstream_scopes is not None:
json_data["upstream_scopes"] = upstream_scopes
if token_bridge is not None:
json_data["token_bridge"] = token_bridge
resp = self.client.post(self.make_path(), json=json_data)
# Raise error if status code is not 2XX or 3XX
resp.raise_for_status()
Expand Down Expand Up @@ -158,6 +173,24 @@ def get(
resp.raise_for_status()
return resp.json()

def check(
self,
vault: str,
permission: int,
) -> JsonDict:
"""Check whether the client holds a permission on a vault.

GET /<client_id>/access/check?vault=&permission=
Returns: {"vault": ..., "permission": <bool>}
"""
resp = self.client.get(
self.make_path("check", end_slash=False),
query={"vault": vault, "permission": permission}
)
# Raise error if status code is not 2XX or 3XX
resp.raise_for_status()
return resp.json()

def grant(
self,
vault: str,
Expand Down
17 changes: 15 additions & 2 deletions campus_python/auth/v1/credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,9 +82,22 @@ def new(
scopes: "list[str]",
expires_in: int,
) -> campus.model.UserCredentials:
raise NotImplementedError(
"Method not expected to be called on API"
"""Issue new credentials for this provider and user.

POST /credentials/<provider>/<user_id> with body
{scopes, expires_in}; the client_id comes from the
request's auth context, not the body.

Returns:
The created UserCredentials
"""
resp = self.client.post(
self.make_path(),
json={"scopes": scopes, "expires_in": expires_in}
)
# Raise error if status code is not 2XX or 3XX
resp.raise_for_status()
return campus.model.UserCredentials.from_resource(resp.json())

def update(self, token: campus.model.OAuthToken) -> None:
client_id = env.CLIENT_ID
Expand Down
139 changes: 139 additions & 0 deletions tests/unit/test_auth_coverage.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
"""Contract tests for the auth coverage remainder (issue #76).

- Clients.Client.access.check() — GET /auth/v1/clients/<id>/access/check
(leaf path, no trailing slash; vault/permission as query params)
- Clients.new() — POST /auth/v1/clients/ also accepts allowed_scopes,
upstream_scopes, token_bridge (as the server route does)
- Credentials.Provider.User.new() — POST /auth/v1/credentials/
<provider>/<user_id> with {scopes, expires_in}; the client_id comes
from the request's auth context, not the body
"""

import unittest
from unittest.mock import Mock

import campus.model

from campus_python.auth.v1 import AuthRoot

# Client resource shape emitted by campus weekly
# (campus/auth/routes/clients.py responses).
CLIENT_RESOURCE = {
"id": "cid123",
"created_at": "2026-09-30T06:31:24.582763+00:00",
"name": "campus-app",
"description": "Server-mode app",
"is_public": False,
"redirect_uris": [],
}

# UserCredentials resource (scope-only token emission, campus #657).
CREDENTIALS_RESOURCE = {
"id": "cred1",
"created_at": "2026-09-30T06:31:24.582763+00:00",
"provider": "campus",
"client_id": "cid123",
"user_id": "user1",
"token": {
"id": "tok123",
"created_at": "2026-09-30T06:31:24.582763+00:00",
"expires_at": "2026-09-30T07:31:24.582763+00:00",
"expires_in": 3600,
"token_type": "Bearer",
"refresh_token": "rt123",
"refresh_token_expires_at": None,
"scope": "campus.profile campus.identities",
},
}


def make_auth() -> tuple[AuthRoot, Mock]:
"""Create an AuthRoot backed by a mock JSON client."""
client = Mock()
return AuthRoot(json_client=client), client


class TestClientAccessCheck(unittest.TestCase):
"""access.check() must GET the /access/check leaf route."""

def setUp(self):
self.auth, self.client = make_auth()
self.client.get.return_value.json.return_value = {
"vault": "campus.vault", "permission": True
}

def test_check_sends_vault_and_permission_as_query(self):
result = self.auth.clients["cid123"].access.check(
vault="campus.vault", permission=3
)
self.client.get.assert_called_once_with(
"/auth/v1/clients/cid123/access/check",
query={"vault": "campus.vault", "permission": 3},
)
self.assertEqual(result, {"vault": "campus.vault", "permission": True})


class TestClientsNewScopeFields(unittest.TestCase):
"""Clients.new() must forward the scope/bridge registration fields."""

def setUp(self):
self.auth, self.client = make_auth()
self.client.post.return_value.json.return_value = CLIENT_RESOURCE

def test_new_sends_scope_fields_when_given(self):
self.auth.clients.new(
name="campus-app",
description="Server-mode app",
allowed_scopes=["campus.api"],
upstream_scopes={"google": ["https://www.googleapis.com/auth/calendar"]},
token_bridge=True,
)
self.assertEqual(
self.client.post.call_args.kwargs["json"],
{
"name": "campus-app",
"description": "Server-mode app",
"is_public": False,
"redirect_uris": [],
"allowed_scopes": ["campus.api"],
"upstream_scopes": {
"google": ["https://www.googleapis.com/auth/calendar"]
},
"token_bridge": True,
},
)

def test_new_omits_unset_scope_fields(self):
self.auth.clients.new(name="campus-app", description="d")
self.assertEqual(
self.client.post.call_args.kwargs["json"],
{
"name": "campus-app",
"description": "d",
"is_public": False,
"redirect_uris": [],
},
)


class TestCredentialsUserNew(unittest.TestCase):
"""User.new() must POST the live endpoint instead of raising."""

def setUp(self):
self.auth, self.client = make_auth()
self.client.post.return_value.json.return_value = CREDENTIALS_RESOURCE

def test_new_posts_scopes_and_expires_in(self):
creds = self.auth.credentials["campus"]["user1"].new(
scopes=["campus.profile"], expires_in=3600
)
self.client.post.assert_called_once_with(
"/auth/v1/credentials/campus/user1",
json={"scopes": ["campus.profile"], "expires_in": 3600},
)
self.assertIsInstance(creds, campus.model.UserCredentials)
self.assertEqual(creds.token.id, "tok123")


if __name__ == "__main__":
unittest.main()
Loading