feat: adds permissions find, find_one, and update (#145)

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,95 @@
+from __future__ import annotations
+
+from typing import List, overload
+
+from requests.sessions import Session as Session
+
+from . import urls
+
+from .config import Config
+from .resources import Resource, Resources
+
+
+class Permission(Resource):
+    @property
+    def id(self) -> str:
+        return self.get("id")  # 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.id}"
+        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
@@ -160,10 +160,12 @@ def __init__(self, config: Config, session: requests.Session) -> None:
     @overload
     def find(
         self, prefix: str = ..., user_role: str = ..., account_status: str = ...
-    ) -> List[User]: ...
+    ) -> List[User]:
+        ...
 
     @overload
-    def find(self, *args, **kwargs) -> List[User]: ...
+    def find(self, *args, **kwargs) -> List[User]:
+        ...
 
     def find(self, *args, **kwargs):
         params = dict(*args, **kwargs)
@@ -181,10 +183,12 @@ def find(self, *args, **kwargs):
     @overload
     def find_one(
         self, prefix: str = ..., user_role: str = ..., account_status: str = ...
-    ) -> User | None: ...
+    ) -> User | None:
+        ...
 
     @overload
-    def find_one(self, *args, **kwargs) -> User | None: ...
+    def find_one(self, *args, **kwargs) -> User | None:
+        ...
 
     def find_one(self, *args, **kwargs) -> User | None:
         params = dict(*args, **kwargs)

tests/posit/connect/__api__/v1/content/f2f37341-e21d-3d80-c698-a935ad614066/permissions.json

@@ -0,0 +1,16 @@
+[
+    {
+        "id": 94,
+        "content_guid": "f2f37341-e21d-3d80-c698-a935ad614066",
+        "principal_guid": "78974391-d89f-4f11-916a-ba50cfe993db",
+        "principal_type": "user",
+        "role": "owner"
+    },
+    {
+        "id": 59,
+        "content_guid": "f2f37341-e21d-3d80-c698-a935ad614066",
+        "principal_guid": "75b95fc0-ae02-4d85-8732-79a845143eed",
+        "principal_type": "group",
+        "role": "viewer"
+    }
+]

tests/posit/connect/__api__/v1/content/f2f37341-e21d-3d80-c698-a935ad614066/permissions/94.json

@@ -0,0 +1,7 @@
+{
+    "id": 94,
+    "content_guid": "f2f37341-e21d-3d80-c698-a935ad614066",
+    "principal_guid": "78974391-d89f-4f11-916a-ba50cfe993db",
+    "principal_type": "user",
+    "role": "owner"
+}

tests/posit/connect/test_permissions.py

@@ -0,0 +1,156 @@
+import random
+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
+
+from .api import load_mock  # type: ignore
+
+
+class TestPermissionUpdate:
+    @responses.activate
+    def test_request_shape(self):
+        # test data
+        id = random.randint(0, 100)
+        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/{id}",
+            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,
+            id=id,
+            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 = "owner"
+        new_role = "viewer"
+
+        id = "94"
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        fake_permission = {
+            **load_mock(f"v1/content/{content_guid}/permissions/{id}.json"),
+            "role": old_role,
+        }
+
+        # define api behavior
+        id = random.randint(0, 100)
+        content_guid = str(uuid.uuid4())
+        responses.put(
+            f"https://connect.example/__api__/v1/content/{content_guid}/permissions/{id}",
+            json={**fake_permission, "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, id=id, 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 = "f2f37341-e21d-3d80-c698-a935ad614066"
+        fake_permissions = load_mock(f"v1/content/{content_guid}/permissions.json")
+
+        # 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 = "f2f37341-e21d-3d80-c698-a935ad614066"
+        fake_permissions = load_mock(f"v1/content/{content_guid}/permissions.json")
+
+        # 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]