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
9 changes: 9 additions & 0 deletions campus_python/auth/v1/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from ...json_client.interface import JsonClient
from . import (
clients,
connections,
credentials,
logins,
oauth,
Expand All @@ -37,6 +38,7 @@ class AuthRoot(ResourceRoot):
def __init__(self, json_client: JsonClient):
super().__init__(json_client=json_client)
self._clients = None
self._connections = None
self._credentials = None
self._logins = None
self._oauth = None
Expand All @@ -52,6 +54,13 @@ def clients(self) -> clients.Clients:
self._clients = clients.Clients(root=self)
return self._clients

@property
def connections(self) -> connections.Connections:
"""Get the connections resource."""
if not self._connections:
self._connections = connections.Connections(root=self)
return self._connections

@property
def credentials(self) -> credentials.Credentials:
"""Get the credentials resource."""
Expand Down
90 changes: 90 additions & 0 deletions campus_python/auth/v1/connections.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
"""campus.python.auth.v1.connections

Campus Auth connections resource (v1).

The user-facing view of the upstream credentials Campus custodies:
which providers/integrations a user has granted, and an explicit way
to revoke that grant. Metadata only — token values never appear here
(release stays broker-only); disconnecting the campus provider is
logout and is rejected by the server (revoke via oauth.revoke instead).
"""

from ...interface import JsonDict, Resource, ResourceCollection


class Connections(ResourceCollection):
"""Campus Auth Connections resource."""
path = "connections/"

def __getitem__(self, provider: str) -> "Connections.Connection":
"""Get a specific provider's connection resource."""
return Connections.Connection(provider, parent=self)

def list(
self,
*,
user_id: "str | None" = None,
) -> "list[JsonDict]":
"""List the target user's upstream connections.

Each entry carries {provider, integration, scopes, connected_at,
expires_at} — never token values. Auth shape mirrors the server:
a bearer token acts for its own user (user_id ignored), while
basic (client-credentials) auth must name a user via user_id.

Args:
user_id: Delegated target user (required for basic auth)

Returns:
List of connection metadata dicts
"""
query = {"user_id": user_id} if user_id else None
resp = self.client.get(self.make_path(), query=query)
resp.raise_for_status()
return resp.json()["connections"]

class Connection(Resource):
"""Single provider connection resource."""

def __getitem__(
self,
integration: str,
) -> "Connections.Connection.Integration":
"""Get an integration-namespaced connection resource."""
return Connections.Connection.Integration(
integration, parent=self
)

def delete(self, *, user_id: "str | None" = None) -> None:
"""Disconnect every credential for this base provider.

404 (NotFoundError) means there was nothing to disconnect;
callers treating disconnect as idempotent can catch it.

Args:
user_id: Delegated target user (required for basic auth)
"""
query = {"user_id": user_id} if user_id else None
resp = self.client.delete(
self.make_path(end_slash=True),
query=query,
)
resp.raise_for_status()
return None

class Integration(Resource):
"""Integration-namespaced connection resource."""

def delete(self, *, user_id: "str | None" = None) -> None:
"""Disconnect this provider integration (e.g. google/classroom).

Args:
user_id: Delegated target user (required for basic auth)
"""
query = {"user_id": user_id} if user_id else None
resp = self.client.delete(
self.make_path(end_slash=True),
query=query,
)
resp.raise_for_status()
return None
87 changes: 87 additions & 0 deletions tests/unit/test_connections.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
"""Contract tests for the auth connections resource (issue #71).

Routes mirror campus/auth/routes/connections.py (campus weekly): the
auth app sets strict_slashes, so DELETE needs the trailing slash on
/connections/<provider>/ and /connections/<provider>/<integration>/.
user_id travels as a query parameter for delegated (basic auth) calls
and is ignored by the server under a bearer token.
"""

import unittest
from unittest.mock import Mock

from campus_python.auth.v1 import AuthRoot


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


CONNECTION = {
"provider": "google.classroom",
"integration": "classroom",
"scopes": ["https://www.googleapis.com/auth/classroom.rosters"],
"connected_at": "2026-10-04T00:00:00+00:00",
"expires_at": None,
}


class TestConnectionsList(unittest.TestCase):
"""connections.list() must GET the collection, optionally delegated."""

def setUp(self):
self.auth, self.client = make_auth()
self.client.get.return_value.json.return_value = {
"connections": [CONNECTION]
}

def test_list_gets_connections_collection(self):
connections = self.auth.connections.list()
self.client.get.assert_called_once_with(
"/auth/v1/connections/", query=None
)
self.assertEqual(connections, [CONNECTION])

def test_list_sends_delegated_user_id_query(self):
self.auth.connections.list(user_id="user-1")
self.client.get.assert_called_once_with(
"/auth/v1/connections/", query={"user_id": "user-1"}
)


class TestConnectionsDelete(unittest.TestCase):
"""Disconnect routes must carry trailing slashes and optional user_id."""

def setUp(self):
self.auth, self.client = make_auth()

def test_provider_delete_uses_trailing_slash(self):
self.auth.connections["google"].delete()
self.client.delete.assert_called_once_with(
"/auth/v1/connections/google/", query=None
)

def test_provider_delete_sends_delegated_user_id(self):
self.auth.connections["google"].delete(user_id="user-1")
self.client.delete.assert_called_once_with(
"/auth/v1/connections/google/", query={"user_id": "user-1"}
)

def test_integration_delete_targets_namespaced_route(self):
self.auth.connections["google"]["classroom"].delete()
self.client.delete.assert_called_once_with(
"/auth/v1/connections/google/classroom/", query=None
)

def test_integration_delete_sends_delegated_user_id(self):
self.auth.connections["google"]["classroom"].delete(user_id="user-1")
self.client.delete.assert_called_once_with(
"/auth/v1/connections/google/classroom/",
query={"user_id": "user-1"},
)


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