diff --git a/campus_python/auth/v1/__init__.py b/campus_python/auth/v1/__init__.py index 9afb6ac..114e57f 100644 --- a/campus_python/auth/v1/__init__.py +++ b/campus_python/auth/v1/__init__.py @@ -18,6 +18,7 @@ from ...json_client.interface import JsonClient from . import ( clients, + connections, credentials, logins, oauth, @@ -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 @@ -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.""" diff --git a/campus_python/auth/v1/connections.py b/campus_python/auth/v1/connections.py new file mode 100644 index 0000000..08fdc7e --- /dev/null +++ b/campus_python/auth/v1/connections.py @@ -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 diff --git a/tests/unit/test_connections.py b/tests/unit/test_connections.py new file mode 100644 index 0000000..da1c480 --- /dev/null +++ b/tests/unit/test_connections.py @@ -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// and /connections///. +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()