feat: add PermissionMembership

Author: Charles Larivier

metabase/resources/permission_membership.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from typing import List
+
+from requests import HTTPError
+
+from resource import CreateResource, DeleteResource, GetResource, ListResource, UpdateResource
+
+
+class PermissionMembership(ListResource, CreateResource, DeleteResource):
+    ENDPOINT = "/api/permissions/membership"
+    PRIMARY_KEY = "membership_id"
+
+    membership_id: int
+    group_id: int
+    user_id: int
+
+    # TODO: allow for bulk updates through /api/permissions/membership/graph
+
+    @classmethod
+    def list(cls) -> List[PermissionMembership]:
+        response = cls.connection().get(cls.ENDPOINT)
+        all_memberships = [item for sublist in response.json().values() for item in sublist]
+        records = [cls(**record) for record in all_memberships]
+        return records
+
+    @classmethod
+    def create(cls, group_id: int, user_id: int, **kwargs) -> PermissionMembership:
+        response = cls.connection().post(cls.ENDPOINT, json={"group_id": group_id, "user_id": user_id})
+
+        if response.status_code != 200:
+            raise HTTPError(response.content.decode())
+
+        # metabase returns a list of all memberships for the given group_id
+        membership = next(filter(lambda x: x["user_id"] == user_id, response.json()))
+
+        return cls(**membership)

tests/resources/test_permission_membership.py

@@ -0,0 +1,51 @@
+from exceptions import NotFoundError
+from metabase import PermissionGroup, PermissionMembership
+
+from tests.helpers import IntegrationTestCase
+
+
+class PermissionMembershipTests(IntegrationTestCase):
+    def tearDown(self) -> None:
+        memberships = PermissionMembership.list()
+        for membership in memberships:
+            if membership.membership_id not in (1, 2):
+                # can't delete default memberships
+                membership.delete()
+
+        groups = PermissionGroup.list()
+        for group in groups:
+            if group.id not in (1, 2):
+                # can't delete default groups
+                group.delete()
+
+    def test_list(self):
+        """Ensure PermissionMembership.list returns a list of PermissionMembership instances."""
+        memberships = PermissionMembership.list()
+
+        self.assertIsInstance(memberships, list)
+        self.assertEqual(2, len(memberships))    # there are 2 default groups in Metabase
+        self.assertTrue(all([isinstance(m, PermissionMembership) for m in memberships]))
+
+    def test_create(self):
+        """Ensure PermissionMembership.create creates a Metric in Metabase and returns a PermissionMembership instance."""
+        group = PermissionGroup.create(name="My Group")
+        membership = PermissionMembership.create(
+            group_id=group.id, user_id=1
+        )
+
+        self.assertIsInstance(membership, PermissionMembership)
+        self.assertEqual(1, membership.user_id)
+
+    def test_delete(self):
+        """Ensure PermissionMembership.delete deletes a PermissionMembership in Metabase."""
+        # fixture
+        group = PermissionGroup.create(name="My Group")
+        membership = PermissionMembership.create(
+            group_id=group.id, user_id=1
+        )
+        self.assertIsInstance(membership, PermissionMembership)
+        self.assertTrue(membership.membership_id in [m.membership_id for m in PermissionMembership.list()])
+
+        membership.delete()
+
+        self.assertFalse(membership.membership_id in [m.membership_id for m in PermissionMembership.list()])