test: Add tests for SharedWorkspacesView

src/tests/unittests/test_views.py

@@ -13,7 +13,7 @@
 from tests.controllers.userauthenticationmethods.test_user_authentication import (
     TestUserAuthentication,
 )
-from user_workspaces_server.models import Job, Workspace
+from user_workspaces_server.models import Job, Workspace, SharedWorkspaceMapping
 
 
 class UserWorkspacesAPITestCase(APITestCase):
@@ -763,3 +763,267 @@ def test_job_types_get(self):
         self.client.force_authenticate(user=self.user)
         response = self.client.get(self.job_types_url)
         self.assertValidResponse(response, status.HTTP_200_OK, success=True, message="Successful.")
+
+
+class SharedWorkspaceAPITestCase(UserWorkspacesAPITestCase):
+    shared_workspaces_url = reverse("shared_workspaces")
+
+    @classmethod
+    def setUpTestData(cls):
+        super().setUpTestData()
+        cls.user_2 = User.objects.create_user("test_2", email="test_2@test.com")
+        # Create a Workspace first that we can share from
+        workspace_data = {
+            "user_id": cls.user,
+            "name": "Test Name",
+            "description": "Test Description",
+            "disk_space": 0,
+            "datetime_created": datetime.now(),
+            "workspace_details": {
+                "request_workspace_details": {"files": [], "symlinks": []},
+                "current_workspace_details": {"files": [], "symlinks": []},
+            },
+            "file_path": "test/1",
+        }
+        cls.original_workspace = Workspace(**workspace_data)
+        cls.original_workspace.save()
+
+        workspace_data["user_id"] = cls.user_2
+        workspace_data["file_path"] = "test/2"
+        cls.shared_workspace = Workspace(**workspace_data)
+        cls.shared_workspace.save()
+
+        shared_workspace_data = {
+            "original_workspace_id": cls.original_workspace,
+            "shared_workspace_id": cls.shared_workspace,
+            "datetime_share_created": datetime.now(),
+        }
+        cls.shared_workspace_mapping = SharedWorkspaceMapping(**shared_workspace_data)
+        cls.shared_workspace_mapping.save()
+
+
+class SharedWorkspaceGETAPITests(SharedWorkspaceAPITestCase):
+    def test_shared_workspace_id_get_original_workspace_user(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.get(
+            reverse("shared_workspaces_with_id", args=[self.shared_workspace.id])
+        )
+        self.assertValidResponse(response, status.HTTP_200_OK, success=True)
+
+    def test_shared_workspace_id_get_shared_workspace_user(self):
+        self.client.force_authenticate(user=self.user_2)
+        response = self.client.get(
+            reverse("shared_workspaces_with_id", args=[self.shared_workspace.id])
+        )
+        self.assertValidResponse(response, status.HTTP_200_OK, success=True)
+
+    def test_shared_workspace_query_param_name_get(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.get(f"{self.shared_workspaces_url}?is_accepted=False")
+        self.assertValidResponse(response, status.HTTP_200_OK, success=True)
+
+    def test_shared_workspaces_get(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.get(self.shared_workspaces_url)
+        self.assertValidResponse(response, status.HTTP_200_OK, success=True)
+
+    def test_shared_workspaces_get_none_found(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.get(f"{self.shared_workspaces_url}?is_accepted=True")
+        self.assertValidResponse(response, status.HTTP_200_OK, success=True)
+
+
+class SharedWorkspacePOSTAPITests(SharedWorkspaceAPITestCase):
+    def test_unauthenticated(self):
+        response = self.client.get(self.shared_workspaces_url, {})
+        self.assertValidResponse(response, status.HTTP_401_UNAUTHORIZED, success=False)
+
+    def test_empty_post(self):
+        self.client.force_authenticate(user=self.user)
+        body = {}
+        response = self.client.post(self.shared_workspaces_url, body)
+        self.assertValidResponse(response, status.HTTP_400_BAD_REQUEST, success=False)
+
+    def test_missing_body_post(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.post(self.shared_workspaces_url)
+        self.assertValidResponse(response, status.HTTP_400_BAD_REQUEST, success=False)
+
+    def test_missing_original_workspace_id_body_post(self):
+        self.client.force_authenticate(user=self.user)
+        body = {"shared_user_ids": []}
+        response = self.client.post(self.shared_workspaces_url, body)
+        self.assertValidResponse(
+            response,
+            status.HTTP_400_BAD_REQUEST,
+            success=False,
+            message="Missing required fields.",
+        )
+
+    def test_missing_shared_user_ids_body_post(self):
+        self.client.force_authenticate(user=self.user)
+        body = {"original_workspace_id": self.original_workspace.pk}
+        response = self.client.post(self.shared_workspaces_url, body)
+        self.assertValidResponse(
+            response,
+            status.HTTP_400_BAD_REQUEST,
+            success=False,
+            message="Missing required fields.",
+        )
+
+    def test_invalid_original_workspace_for_user(self):
+        self.client.force_authenticate(user=self.user)
+        body = {"shared_user_ids": [], "original_workspace_id": self.shared_workspace.pk}
+        response = self.client.post(self.shared_workspaces_url, body)
+        self.assertValidResponse(
+            response,
+            status.HTTP_404_NOT_FOUND,
+            success=False,
+            message=f"Workspace {body['original_workspace_id']} not found for user.",
+        )
+
+    def test_invalid_shared_user_ids_format(self):
+        self.client.force_authenticate(user=self.user)
+        body = {"shared_user_ids": "", "original_workspace_id": self.original_workspace.pk}
+        response = self.client.post(self.shared_workspaces_url, body)
+        self.assertValidResponse(
+            response,
+            status.HTTP_400_BAD_REQUEST,
+            success=False,
+            message="shared_user_ids is not a list.",
+        )
+
+    def test_invalid_shared_user_ids(self):
+        self.client.force_authenticate(user=self.user)
+        body = {
+            "shared_user_ids": [-9999],
+            "original_workspace_id": self.original_workspace.pk,
+        }
+        response = self.client.post(self.shared_workspaces_url, body)
+        self.assertValidResponse(
+            response,
+            status.HTTP_400_BAD_REQUEST,
+            success=False,
+            message="Invalid user id provided.",
+        )
+
+    def test_minimum_valid_post(self):
+        self.client.force_authenticate(user=self.user)
+        body = {
+            "shared_user_ids": [self.user_2.pk],
+            "original_workspace_id": self.original_workspace.pk,
+        }
+        response = self.client.post(self.shared_workspaces_url, body)
+        self.assertValidResponse(response, status.HTTP_200_OK, success=True, message="Successful.")
+        # TODO: Check body
+
+
+class SharedWorkspacePUTAPITests(SharedWorkspaceAPITestCase):
+    def test_shared_workspace_not_found_put(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.put(reverse("shared_workspaces_put_type", args=[9999, "test"]))
+        self.assertValidResponse(
+            response,
+            status.HTTP_404_NOT_FOUND,
+            success=False,
+            message="Shared workspace 9999 not found.",
+        )
+
+    def test_workspace_invalid_put_type_put(self):
+        self.client.force_authenticate(user=self.user_2)
+        response = self.client.put(
+            reverse("shared_workspaces_put_type", args=[self.shared_workspace.pk, "test"])
+        )
+        self.assertValidResponse(
+            response,
+            status.HTTP_400_BAD_REQUEST,
+            success=False,
+            message="Invalid PUT type passed.",
+        )
+
+    def test_shared_workspace_put_type_accept(self):
+        self.client.force_authenticate(user=self.user_2)
+        response = self.client.put(
+            reverse("shared_workspaces_put_type", args=[self.shared_workspace.pk, "accept"])
+        )
+        self.assertValidResponse(response, status.HTTP_200_OK, success=True, message="Successful.")
+
+
+class SharedWorkspaceDELETEAPITests(SharedWorkspaceAPITestCase):
+    def test_shared_workspace_not_found_delete(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.delete(reverse("shared_workspaces_with_id", args=[9999]))
+        self.assertValidResponse(
+            response,
+            status.HTTP_404_NOT_FOUND,
+            success=False,
+            message="Shared workspace 9999 not found.",
+        )
+
+    def test_shared_workspace_delete_accepted_workspace(self):
+        self.shared_workspace_mapping.is_accepted = True
+        self.shared_workspace_mapping.save()
+
+        self.client.force_authenticate(user=self.user)
+        response = self.client.delete(
+            reverse("shared_workspaces_with_id", args=[self.shared_workspace.pk])
+        )
+        self.assertValidResponse(
+            response,
+            status.HTTP_400_BAD_REQUEST,
+            success=False,
+            message=f"Shared workspace {self.shared_workspace.pk} has been accepted and cannot be deleted.",
+        )
+
+    def test_shared_workspace_delete_invalid_path(self):
+        self.shared_workspace.file_path = "."
+        self.shared_workspace.save()
+
+        self.client.force_authenticate(user=self.user)
+        response = self.client.delete(
+            reverse("shared_workspaces_with_id", args=[self.shared_workspace.pk])
+        )
+        self.assertValidResponse(
+            response,
+            status.HTTP_500_INTERNAL_SERVER_ERROR,
+            success=False,
+            message="Please contact a system administrator there is a failure with "
+            "the workspace directory that will not allow for this workspace to be deleted.",
+        )
+
+    def test_shared_workspace_delete_invalid_user(self):
+        invalid_user = User.objects.create_user("invalid_user", email="invalid@test.com")
+        self.client.force_authenticate(user=invalid_user)
+        response = self.client.delete(
+            reverse("shared_workspaces_with_id", args=[self.shared_workspace.pk])
+        )
+        self.assertValidResponse(
+            response,
+            status.HTTP_403_FORBIDDEN,
+            success=False,
+            message=f"User does not have permissions for shared workspace {self.shared_workspace.pk}.",
+        )
+
+    def test_shared_workspace_delete_original_user(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.delete(
+            reverse("shared_workspaces_with_id", args=[self.shared_workspace.pk])
+        )
+        self.assertValidResponse(
+            response,
+            status.HTTP_200_OK,
+            success=True,
+            message=f"Shared workspace {self.shared_workspace.pk} queued for deletion.",
+        )
+
+    def test_shared_workspace_delete_shared_user(self):
+        self.client.force_authenticate(user=self.user_2)
+        response = self.client.delete(
+            reverse("shared_workspaces_with_id", args=[self.shared_workspace.pk])
+        )
+        self.assertValidResponse(
+            response,
+            status.HTTP_200_OK,
+            success=True,
+            message=f"Shared workspace {self.shared_workspace.pk} queued for deletion.",
+        )