feat: adds permissions find, find_one, and update

src/posit/connect/content.py

@@ -10,22 +10,12 @@
 from . import urls
 
 from .config import Config
+from .permissions import Permissions
 from .resources import Resources, Resource
 
 
 class ContentItem(Resource):
-    """A piece of content.
-
-    Parameters
-    ----------
-    Resource : _type_
-        _description_
-
-    Returns
-    -------
-    _type_
-        _description_
-    """
+    """A piece of content."""
 
     @property
     def guid(self) -> str:
@@ -199,6 +189,10 @@ def app_role(self) -> str:
     def id(self) -> str:
         return self.get("id")  # type: ignore
 
+    @property
+    def permissions(self) -> Permissions:
+        return Permissions(self.config, self.session, content_guid=self.guid)
+
     @overload
     def update(
         self,
@@ -297,7 +291,7 @@ def update(self, *args, **kwargs) -> None:
         super().update(**response.json())
 
 
-class Content(Resources[ContentItem]):
+class Content(Resources):
     def __init__(self, config: Config, session: Session) -> None:
         self.url = urls.append_path(config.url, "v1/content")
         self.config = config

src/posit/connect/permissions.py

@@ -0,0 +1,106 @@
+from typing import List, overload
+
+from requests.sessions import Session as Session
+
+from posit.connect.config import Config
+
+from . import urls
+
+from .resources import Resource, Resources
+
+
+class Permission(Resource):
+    @property
+    def guid(self) -> str:
+        """The globally unique identifier.
+
+        An alias for the field 'id'.
+
+        The conventional name for this field is 'guid'. As of v2024.03.0, the API uses the field name 'id'.
+
+        Returns
+        -------
+        str
+        """
+        # Alias 'id' to 'guid'.
+        # The conventional name for this field across applications is 'guid'.
+        return self.get("id", self.get("guid"))  # type: ignore
+
+    @property
+    def content_guid(self) -> str:
+        return self.get("content_guid")  # type: ignore
+
+    @property
+    def principal_guid(self) -> str:
+        return self.get("principal_guid")  # type: ignore
+
+    @property
+    def principal_type(self) -> str:
+        return self.get("principal_type")  # type: ignore
+
+    @property
+    def role(self) -> str:
+        return self.get("role")  # type: ignore
+
+    @overload
+    def update(self, role: str) -> None:
+        """Update a permission.
+
+        Parameters
+        ----------
+        role : str
+            The principal role.
+        """
+        ...
+
+    @overload
+    def update(self, *args, **kwargs) -> None:
+        """Update a permission."""
+        ...
+
+    def update(self, *args, **kwargs) -> None:
+        """Update a permission."""
+        body = dict(*args, **kwargs)
+        path = f"v1/content/{self.content_guid}/permissions/{self.guid}"
+        url = urls.append_path(self.config.url, path)
+        response = self.session.put(
+            url,
+            json={
+                "principal_guid": self.principal_guid,
+                "principal_type": self.principal_type,
+                "role": self.role,
+                # shorthand to overwrite the above fields with method arguments
+                **body,
+            },
+        )
+        super().update(**response.json())
+
+
+class Permissions(Resources):
+    def __init__(self, config: Config, session: Session, *, content_guid: str) -> None:
+        super().__init__(config, session)
+        self.content_guid = content_guid
+
+    def find(self, *args, **kwargs) -> List[Permission]:
+        """Find permissions.
+
+        Returns
+        -------
+        List[Permission]
+        """
+        body = dict(*args, **kwargs)
+        path = f"v1/content/{self.content_guid}/permissions"
+        url = urls.append_path(self.config.url, path)
+        response = self.session.get(url, json=body)
+        results = response.json()
+        return [Permission(self.config, self.session, **result) for result in results]
+
+    def find_one(self, *args, **kwargs) -> Permission | None:
+        """Find a permission.
+
+        Returns
+        -------
+        Permission | None
+        """
+        permissions = self.find(*args, **kwargs)
+        return next(iter(permissions), None)

src/posit/connect/resources.py

@@ -44,31 +44,7 @@ def __setattr__(self, name: str, value: Any) -> None:
         raise AttributeError("cannot set attributes: use update() instead")
 
 
-class Resources(ABC, Generic[T]):
-    @abstractmethod
-    def create(self, *args, **kwargs) -> T:
-        raise NotImplementedError()
-
-    @abstractmethod
-    def delete(self, *args, **kwargs) -> None:
-        raise NotImplementedError()
-
-    @abstractmethod
-    def find(self, *args, **kwargs) -> List[T]:
-        raise NotImplementedError()
-
-    @abstractmethod
-    def find_one(self, *args, **kwargs) -> Optional[T]:
-        raise NotImplementedError()
-
-    @abstractmethod
-    def get(self, *args, **kwargs) -> T:
-        raise NotImplementedError()
-
-    @abstractmethod
-    def update(self, *args, **kwargs) -> T:
-        raise NotImplementedError()
-
-    @abstractmethod
-    def count(self, *args, **kwargs) -> int:
-        raise NotImplementedError()
+class Resources(ABC):
+    def __init__(self, config: Config, session: requests.Session) -> None:
+        self.config = config
+        self.session = session

src/posit/connect/users.py

@@ -151,7 +151,7 @@ def update(self, *args, **kwargs) -> None:
         super().update(**response.json())
 
 
-class Users(Resources[User]):
+class Users(Resources):
     def __init__(self, config: Config, session: requests.Session) -> None:
         self.url = urls.append_path(config.url, "v1/users")
         self.config = config

tests/posit/connect/test_permissions.py

@@ -0,0 +1,192 @@
+import uuid
+
+from unittest.mock import Mock
+
+import requests
+import responses
+
+from responses import matchers
+
+from posit.connect.config import Config
+from posit.connect.permissions import Permission, Permissions
+
+
+class TestPermissionGuid:
+    def test_from_id(self):
+        config = Mock()
+        session = Mock()
+        guid = str(uuid.uuid4())
+        permission = Permission(config, session, guid=guid)
+        assert permission.guid == guid
+
+    def test_from_guid(self):
+        config = Mock()
+        session = Mock()
+        guid = str(uuid.uuid4())
+        permission = Permission(config, session, id=guid)
+        assert permission.guid == guid
+
+
+class TestPermissionUpdate:
+    @responses.activate
+    def test_request_shape(self):
+        # test data
+        guid = str(uuid.uuid4())
+        content_guid = str(uuid.uuid4())
+        principal_guid = str(uuid.uuid4())
+        principal_type = "principal_type"
+        role = "role"
+        extraneous = "extraneous"
+
+        # define api behavior
+        responses.put(
+            f"https://connect.example/__api__/v1/content/{content_guid}/permissions/{guid}",
+            json={
+                # doesn't matter for this test
+            },
+            match=[
+                # assertion
+                matchers.json_params_matcher(
+                    {
+                        # validate that initial permission fields are set
+                        "principal_guid": principal_guid,
+                        "principal_type": principal_type,
+                        "role": role,
+                        # validate that arguments passed to update are set
+                        "extraneous": extraneous,
+                    }
+                )
+            ],
+        )
+
+        # setup
+        config = Config(api_key="12345", url="https://connect.example/")
+        session = requests.Session()
+        permission = Permission(
+            config,
+            session,
+            guid=guid,
+            content_guid=content_guid,
+            principal_guid=principal_guid,
+            principal_type=principal_type,
+            role=role,
+        )
+
+        # invoke
+        # assertion occurs in match above
+        permission.update(extraneous=extraneous)
+
+    @responses.activate
+    def test_role_update(self):
+        # test data
+        old_role = "old_role"
+        new_role = "new_role"
+
+        # define api behavior
+        guid = str(uuid.uuid4())
+        content_guid = str(uuid.uuid4())
+        responses.put(
+            f"https://connect.example/__api__/v1/content/{content_guid}/permissions/{guid}",
+            json={"role": new_role},
+            match=[
+                matchers.json_params_matcher(
+                    {
+                        "principal_guid": None,
+                        "principal_type": None,
+                        "role": new_role,
+                    }
+                )
+            ],
+        )
+
+        # setup
+        config = Config(api_key="12345", url="https://connect.example/")
+        session = requests.Session()
+        permission = Permission(
+            config, session, guid=guid, content_guid=content_guid, role=old_role
+        )
+
+        # assert role change with respect to api response
+        assert permission.role == old_role
+        permission.update(role=new_role)
+        assert permission.role == new_role
+
+
+class TestPermissionsFind:
+    @responses.activate
+    def test(self):
+        # test data
+        content_guid = str(uuid.uuid4())
+        fake_permissions = [
+            {
+                "guid": str(uuid.uuid4()),
+                "content_guid": content_guid,
+                "principal_guid": str(uuid.uuid4()),
+                "principal_type": "user",
+                "role": "read",
+            },
+            {
+                "guid": str(uuid.uuid4()),
+                "content_guid": content_guid,
+                "principal_guid": str(uuid.uuid4()),
+                "principal_type": "group",
+                "role": "write",
+            },
+        ]
+
+        # define api behavior
+        responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/permissions",
+            json=fake_permissions,
+        )
+
+        # setup
+        config = Config(api_key="12345", url="https://connect.example/")
+        session = requests.Session()
+        permissions = Permissions(config, session, content_guid=content_guid)
+
+        # invoke
+        permissions = permissions.find()
+
+        # assert response
+        assert permissions == fake_permissions
+
+
+class TestPermissionsFindOne:
+    @responses.activate
+    def test(self):
+        # test data
+        content_guid = str(uuid.uuid4())
+        fake_permissions = [
+            {
+                "guid": str(uuid.uuid4()),
+                "content_guid": content_guid,
+                "principal_guid": str(uuid.uuid4()),
+                "principal_type": "user",
+                "role": "read",
+            },
+            {
+                "guid": str(uuid.uuid4()),
+                "content_guid": content_guid,
+                "principal_guid": str(uuid.uuid4()),
+                "principal_type": "group",
+                "role": "write",
+            },
+        ]
+
+        # define api behavior
+        responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/permissions",
+            json=fake_permissions,
+        )
+
+        # setup
+        config = Config(api_key="12345", url="https://connect.example/")
+        session = requests.Session()
+        permissions = Permissions(config, session, content_guid=content_guid)
+
+        # invoke
+        permission = permissions.find_one()
+
+        # assert response
+        assert permission == fake_permissions[0]