Merge pull request #702 from UW-GAC/feature/workspace-adapter-after-a…

…nvil-import Automatically share workspaces with admin group after import
UW-GAC · Aug 9, 2024 · 0e9f352 · 0e9f352
2 parents 0e166c1 + f340a71
commit 0e9f352
Show file tree

Hide file tree

Showing 2 changed files with 201 additions and 1 deletion.
diff --git a/primed/primed_anvil/adapters.py b/primed/primed_anvil/adapters.py
@@ -80,6 +80,35 @@ def after_anvil_create(self, workspace):
         )
         sharing.anvil_create_or_update()
 
+    def after_anvil_import(self, workspace):
+        super().after_anvil_import(workspace)
+        # # Check if the workspace is already shared with the ADMINs group.
+        try:
+            admins_group = ManagedGroup.objects.get(name=settings.ANVIL_CC_ADMINS_GROUP_NAME)
+        except ManagedGroup.DoesNotExist:
+            return
+        try:
+            sharing = WorkspaceGroupSharing.objects.get(
+                workspace=workspace,
+                group=admins_group,
+            )
+        except WorkspaceGroupSharing.DoesNotExist:
+            sharing = WorkspaceGroupSharing.objects.create(
+                workspace=workspace,
+                group=admins_group,
+                access=WorkspaceGroupSharing.OWNER,
+                can_compute=True,
+            )
+            sharing.save()
+            sharing.anvil_create_or_update()
+        else:
+            # If the existing sharing record exists, make sure it has the correct permissions.
+            if not sharing.can_compute or sharing.access != WorkspaceGroupSharing.OWNER:
+                sharing.can_compute = True
+                sharing.access = WorkspaceGroupSharing.OWNER
+                sharing.save()
+                sharing.anvil_create_or_update()
+
 
 class ManagedGroupAdapter(BaseManagedGroupAdapter):
     """Adapter for ManagedGroups."""

diff --git a/primed/primed_anvil/tests/test_adapters.py b/primed/primed_anvil/tests/test_adapters.py
@@ -1,7 +1,12 @@
 import responses
 from anvil_consortium_manager.adapters.default import DefaultWorkspaceAdapter
 from anvil_consortium_manager.models import Account, GroupGroupMembership, WorkspaceGroupSharing
-from anvil_consortium_manager.tests.factories import AccountFactory, ManagedGroupFactory, WorkspaceFactory
+from anvil_consortium_manager.tests.factories import (
+    AccountFactory,
+    ManagedGroupFactory,
+    WorkspaceFactory,
+    WorkspaceGroupSharingFactory,
+)
 from anvil_consortium_manager.tests.utils import AnVILAPIMockTestMixin
 from django.test import TestCase, override_settings
 
@@ -230,6 +235,172 @@ def test_after_anvil_create_no_admins_group(self):
         # No WorkspaceGroupSharing objects were created.
         self.assertEqual(WorkspaceGroupSharing.objects.count(), 0)
 
+    def test_after_anvil_import(self):
+        admins_group = ManagedGroupFactory.create(name="TEST_PRIMED_CC_ADMINS")
+        workspace = WorkspaceFactory.create(
+            billing_project__name="bar", name="foo", workspace_type=self.adapter.get_type()
+        )
+        # API response for admin group workspace owner.
+        acls = [
+            {
+                "email": "[email protected]",
+                "accessLevel": "OWNER",
+                "canShare": False,
+                "canCompute": True,
+            }
+        ]
+        self.anvil_response_mock.add(
+            responses.PATCH,
+            self.api_client.rawls_entry_point + "/api/workspaces/bar/foo/acl?inviteUsersNotFound=false",
+            status=200,
+            match=[responses.matchers.json_params_matcher(acls)],
+            json={"invitesSent": {}, "usersNotFound": {}, "usersUpdated": acls},
+        )
+        # Run the adapter method.
+        self.adapter.after_anvil_import(workspace)
+        # Check for WorkspaceGroupSharing.
+        self.assertEqual(WorkspaceGroupSharing.objects.count(), 1)
+        sharing = WorkspaceGroupSharing.objects.first()
+        self.assertEqual(sharing.workspace, workspace)
+        self.assertEqual(sharing.group, admins_group)
+        self.assertEqual(sharing.access, WorkspaceGroupSharing.OWNER)
+        self.assertTrue(sharing.can_compute)
+
+    @override_settings(ANVIL_CC_ADMINS_GROUP_NAME="foobar")
+    def test_after_anvil_import_different_admins_group(self):
+        admins_group = ManagedGroupFactory.create(name="foobar")
+        workspace = WorkspaceFactory.create(
+            billing_project__name="bar", name="foo", workspace_type=self.adapter.get_type()
+        )
+        # API response for admin group workspace owner.
+        acls = [
+            {
+                "email": "[email protected]",
+                "accessLevel": "OWNER",
+                "canShare": False,
+                "canCompute": True,
+            }
+        ]
+        self.anvil_response_mock.add(
+            responses.PATCH,
+            self.api_client.rawls_entry_point + "/api/workspaces/bar/foo/acl?inviteUsersNotFound=false",
+            status=200,
+            match=[responses.matchers.json_params_matcher(acls)],
+            json={"invitesSent": {}, "usersNotFound": {}, "usersUpdated": acls},
+        )
+        # Run the adapter method.
+        self.adapter.after_anvil_import(workspace)
+        # Check for WorkspaceGroupSharing.
+        self.assertEqual(WorkspaceGroupSharing.objects.count(), 1)
+        sharing = WorkspaceGroupSharing.objects.first()
+        self.assertEqual(sharing.workspace, workspace)
+        self.assertEqual(sharing.group, admins_group)
+        self.assertEqual(sharing.access, WorkspaceGroupSharing.OWNER)
+        self.assertTrue(sharing.can_compute)
+
+    def test_after_anvil_import_no_admins_group(self):
+        workspace = WorkspaceFactory.create(
+            billing_project__name="bar", name="foo", workspace_type=self.adapter.get_type()
+        )
+        # Run the adapter method.
+        self.adapter.after_anvil_import(workspace)
+        # No WorkspaceGroupSharing objects were created.
+        self.assertEqual(WorkspaceGroupSharing.objects.count(), 0)
+
+    def test_after_anvil_import_already_shared(self):
+        admins_group = ManagedGroupFactory.create(name="TEST_PRIMED_CC_ADMINS")
+        workspace = WorkspaceFactory.create(workspace_type=self.adapter.get_type())
+        WorkspaceGroupSharingFactory.create(
+            workspace=workspace,
+            group=admins_group,
+            access=WorkspaceGroupSharing.OWNER,
+            can_compute=True,
+        )
+        # No API call - record already exists.
+        # Run the adapter method.
+        self.adapter.after_anvil_import(workspace)
+        # Check for WorkspaceGroupSharing.
+        self.assertEqual(WorkspaceGroupSharing.objects.count(), 1)
+        sharing = WorkspaceGroupSharing.objects.first()
+        self.assertEqual(sharing.workspace, workspace)
+        self.assertEqual(sharing.group, admins_group)
+        self.assertEqual(sharing.access, WorkspaceGroupSharing.OWNER)
+        self.assertTrue(sharing.can_compute)
+
+    def test_after_anvil_import_already_shared_wrong_access(self):
+        admins_group = ManagedGroupFactory.create(name="TEST_PRIMED_CC_ADMINS")
+        workspace = WorkspaceFactory.create(
+            billing_project__name="bar", name="foo", workspace_type=self.adapter.get_type()
+        )
+        sharing = WorkspaceGroupSharingFactory.create(
+            workspace=workspace,
+            group=admins_group,
+            access=WorkspaceGroupSharing.READER,
+            can_compute=True,
+        )
+        # API response to update sharing.
+        acls = [
+            {
+                "email": "[email protected]",
+                "accessLevel": "OWNER",
+                "canShare": False,
+                "canCompute": True,
+            }
+        ]
+        self.anvil_response_mock.add(
+            responses.PATCH,
+            self.api_client.rawls_entry_point + "/api/workspaces/bar/foo/acl?inviteUsersNotFound=false",
+            status=200,
+            match=[responses.matchers.json_params_matcher(acls)],
+            json={"invitesSent": {}, "usersNotFound": {}, "usersUpdated": acls},
+        )
+        # Run the adapter method.
+        self.adapter.after_anvil_import(workspace)
+        # Check for WorkspaceGroupSharing.
+        self.assertEqual(WorkspaceGroupSharing.objects.count(), 1)
+        sharing.refresh_from_db()
+        self.assertEqual(sharing.workspace, workspace)
+        self.assertEqual(sharing.group, admins_group)
+        self.assertEqual(sharing.access, WorkspaceGroupSharing.OWNER)
+        self.assertTrue(sharing.can_compute)
+
+    def test_after_anvil_import_already_shared_wrong_can_compute(self):
+        admins_group = ManagedGroupFactory.create(name="TEST_PRIMED_CC_ADMINS")
+        workspace = WorkspaceFactory.create(
+            billing_project__name="bar", name="foo", workspace_type=self.adapter.get_type()
+        )
+        sharing = WorkspaceGroupSharingFactory.create(
+            workspace=workspace,
+            group=admins_group,
+            access=WorkspaceGroupSharing.OWNER,
+            can_compute=False,
+        )
+        # API response to update sharing.
+        acls = [
+            {
+                "email": "[email protected]",
+                "accessLevel": "OWNER",
+                "canShare": False,
+                "canCompute": True,
+            }
+        ]
+        self.anvil_response_mock.add(
+            responses.PATCH,
+            self.api_client.rawls_entry_point + "/api/workspaces/bar/foo/acl?inviteUsersNotFound=false",
+            status=200,
+            match=[responses.matchers.json_params_matcher(acls)],
+            json={"invitesSent": {}, "usersNotFound": {}, "usersUpdated": acls},
+        )
+        # Run the adapter method.
+        self.adapter.after_anvil_import(workspace)
+        # Check for WorkspaceGroupSharing.
+        self.assertEqual(WorkspaceGroupSharing.objects.count(), 1)
+        sharing.refresh_from_db()
+        self.assertEqual(sharing.workspace, workspace)
+        self.assertEqual(sharing.group, admins_group)
+        self.assertEqual(sharing.access, WorkspaceGroupSharing.OWNER)
+        self.assertTrue(sharing.can_compute)
+
 
 class ManagedGroupAdapterTest(AnVILAPIMockTestMixin, TestCase):
     """Tests for the custom PRIMED ManagedGroupAdapter."""