diff --git a/.idea/inspectionProfiles/Project_Default.xml b/.idea/inspectionProfiles/Project_Default.xml index 4d74dbe..91884f2 100644 --- a/.idea/inspectionProfiles/Project_Default.xml +++ b/.idea/inspectionProfiles/Project_Default.xml @@ -37,6 +37,13 @@ + + + diff --git a/backend/sophon/core/errors.py b/backend/sophon/core/errors.py new file mode 100644 index 0000000..7039720 --- /dev/null +++ b/backend/sophon/core/errors.py @@ -0,0 +1,13 @@ +from rest_framework.response import Response + + +class HTTPException(Exception): + """ + An exception that can be raised in :class:`.SophonViewSet` hooks to respond to a request with an HTTP error. + """ + + def __init__(self, status: int): + self.status = status + + def as_response(self) -> Response: + return Response(status=self.status) diff --git a/backend/sophon/core/tests.py b/backend/sophon/core/tests.py index ad2f7fa..999ce84 100644 --- a/backend/sophon/core/tests.py +++ b/backend/sophon/core/tests.py @@ -30,9 +30,9 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta): self.assertTrue(isinstance(response.data, dict)) return response.data - def list_fail(self) -> None: + def list_fail(self, code) -> None: response = self.list() - self.assertTrue(response.status_code >= 400) + self.assertEqual(response.status_code, code) def retrieve(self, pk) -> Response: url = self.get_url("detail", pk=pk) @@ -44,9 +44,9 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta): self.assertTrue(isinstance(response.data, dict)) return response.data - def retrieve_fail(self, pk) -> None: + def retrieve_fail(self, pk, code) -> None: response = self.retrieve(pk=pk) - self.assertTrue(response.status_code >= 400) + self.assertEqual(response.status_code, code) def create(self, data) -> Response: url = self.get_url("list") @@ -58,9 +58,9 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta): self.assertTrue(isinstance(response.data, dict)) return response.data - def create_fail(self, data) -> None: + def create_fail(self, data, code) -> None: response = self.create(data) - self.assertTrue(response.status_code >= 400) + self.assertEqual(response.status_code, code) def update(self, pk, data) -> Response: url = self.get_url("detail", pk=pk) @@ -72,6 +72,10 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta): self.assertTrue(isinstance(response.data, dict)) return response.data + def update_fail(self, pk, data, code) -> None: + response = self.update(pk, data) + self.assertEqual(response.status_code, code) + def destroy(self, pk) -> Response: url = self.get_url("detail", pk=pk) return self.client.delete(url, format="json") @@ -80,9 +84,9 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta): response = self.destroy(pk=pk) self.assertEqual(response.status_code, 204) - def destroy_fail(self, pk) -> None: + def destroy_fail(self, pk, code) -> None: response = self.destroy(pk) - self.assertTrue(response.status_code >= 400) + self.assertEqual(response.status_code, code) class ResearchGroupTests(SophonModelTestCase): @@ -91,10 +95,12 @@ class ResearchGroupTests(SophonModelTestCase): return "research-group" test_user: User = None + other_user: User = None @classmethod def setUpTestData(cls): cls.test_user = User.objects.create_user(username="TEST", password="TheGreatDjangoTest") + cls.other_user = User.objects.create_user(username="TAST", password="TheGreatDjangoTast") models.ResearchGroup.objects.create( slug="alpha", @@ -118,66 +124,67 @@ class ResearchGroupTests(SophonModelTestCase): count = r["count"] self.assertEqual(count, 2) - results = r["results"] + list_page = r["results"] - self.assertIn("slug", results[0]) - self.assertIn("name", results[0]) - self.assertIn("description", results[0]) - self.assertIn("owner", results[0]) - self.assertIn("members", results[0]) - self.assertIn("access", results[0]) + self.assertIn("slug", list_page[0]) + self.assertIn("name", list_page[0]) + self.assertIn("description", list_page[0]) + self.assertIn("owner", list_page[0]) + self.assertIn("members", list_page[0]) + self.assertIn("access", list_page[0]) def test_retrieve_valid(self): - result = self.retrieve_unwrap("alpha") + retrieved = self.retrieve_unwrap("alpha") - self.assertIn("slug", result) - self.assertIn("name", result) - self.assertIn("description", result) - self.assertIn("owner", result) - self.assertIn("members", result) - self.assertIn("access", result) + self.assertIn("slug", retrieved) + self.assertIn("name", retrieved) + self.assertIn("description", retrieved) + self.assertIn("owner", retrieved) + self.assertIn("members", retrieved) + self.assertIn("access", retrieved) - self.assertEqual(result["slug"], "alpha") - self.assertEqual(result["name"], "Alpha") - self.assertEqual(result["description"], "First test group.") - self.assertEqual(result["owner"], self.test_user.id) - self.assertEqual(result["members"], []) - self.assertEqual(result["access"], "MANUAL") + self.assertEqual(retrieved["slug"], "alpha") + self.assertEqual(retrieved["name"], "Alpha") + self.assertEqual(retrieved["description"], "First test group.") + self.assertEqual(retrieved["owner"], self.test_user.id) + self.assertEqual(retrieved["members"], []) + self.assertEqual(retrieved["access"], "MANUAL") def test_retrieve_not_existing(self): - self.retrieve_fail("banana") + self.retrieve_fail("banana", 404) def test_create_valid(self): self.client.login(username="TEST", password="TheGreatDjangoTest") - result = self.create_unwrap({ + created = self.create_unwrap({ "slug": "omega", "name": "Omega", "description": "Last test group.", "members": [], "access": "OPEN", }) - self.assertIn("slug", result) - self.assertIn("name", result) - self.assertIn("description", result) - self.assertIn("members", result) - self.assertIn("access", result) + self.assertIn("slug", created) + self.assertIn("name", created) + self.assertIn("description", created) + self.assertIn("owner", created) + self.assertIn("members", created) + self.assertIn("access", created) - check = self.retrieve_unwrap("omega") + retrieved = self.retrieve_unwrap("omega") - self.assertIn("slug", check) - self.assertIn("name", check) - self.assertIn("description", check) - self.assertIn("owner", check) - self.assertIn("members", check) - self.assertIn("access", check) + self.assertIn("slug", retrieved) + self.assertIn("name", retrieved) + self.assertIn("description", retrieved) + self.assertIn("owner", retrieved) + self.assertIn("members", retrieved) + self.assertIn("access", retrieved) - self.assertEqual(check["slug"], "omega") - self.assertEqual(check["name"], "Omega") - self.assertEqual(check["description"], "Last test group.") - self.assertEqual(result["owner"], self.test_user.id) - self.assertEqual(result["members"], []) - self.assertEqual(result["access"], "OPEN") + self.assertEqual(retrieved["slug"], "omega") + self.assertEqual(retrieved["name"], "Omega") + self.assertEqual(retrieved["description"], "Last test group.") + self.assertEqual(retrieved["owner"], self.test_user.id) + self.assertEqual(retrieved["members"], []) + self.assertEqual(retrieved["access"], "OPEN") def test_create_not_logged_in(self): self.create_fail({ @@ -186,7 +193,7 @@ class ResearchGroupTests(SophonModelTestCase): "description": "This creation should fail.", "members": [], "access": "OPEN", - }) + }, 401) def test_create_invalid_schema(self): self.client.login(username="TEST", password="TheGreatDjangoTest") @@ -194,34 +201,29 @@ class ResearchGroupTests(SophonModelTestCase): self.create_fail({ "potato": "sweet", "access": "OPEN", - }) + }, 400) def test_update_valid(self): self.client.login(username="TEST", password="TheGreatDjangoTest") - creation = self.create_unwrap({ + self.create_unwrap({ "slug": "gamma", "name": "Gamma", "description": "A test group to update.", "members": [], "access": "OPEN", }) - self.assertIn("slug", creation) - self.assertIn("name", creation) - self.assertIn("description", creation) - self.assertIn("members", creation) - self.assertIn("access", creation) - check = self.retrieve_unwrap("gamma") + retrieved = self.retrieve_unwrap("gamma") - self.assertEqual(check["slug"], "gamma") - self.assertEqual(check["name"], "Gamma") - self.assertEqual(check["description"], "A test group to update.") - self.assertEqual(check["owner"], self.test_user.id) - self.assertEqual(check["members"], []) - self.assertEqual(check["access"], "OPEN") + self.assertEqual(retrieved["slug"], "gamma") + self.assertEqual(retrieved["name"], "Gamma") + self.assertEqual(retrieved["description"], "A test group to update.") + self.assertEqual(retrieved["owner"], self.test_user.id) + self.assertEqual(retrieved["members"], []) + self.assertEqual(retrieved["access"], "OPEN") - update = self.update_unwrap("gamma", { + updated = self.update_unwrap("gamma", { "slug": "gamma", "name": "Gamma", "description": "An updated test group.", @@ -229,69 +231,99 @@ class ResearchGroupTests(SophonModelTestCase): "access": "MANUAL", }) - self.assertIn("slug", update) - self.assertIn("name", update) - self.assertIn("description", update) - self.assertIn("owner", update) - self.assertIn("members", update) - self.assertIn("access", update) + self.assertIn("slug", updated) + self.assertIn("name", updated) + self.assertIn("description", updated) + self.assertIn("owner", updated) + self.assertIn("members", updated) + self.assertIn("access", updated) - self.assertEqual(update["slug"], "gamma") - self.assertEqual(update["name"], "Gamma") - self.assertEqual(update["description"], "An updated test group.") - self.assertEqual(update["owner"], self.test_user.id) - self.assertEqual(update["members"], []) - self.assertEqual(update["access"], "MANUAL") + self.assertEqual(updated["slug"], "gamma") + self.assertEqual(updated["name"], "Gamma") + self.assertEqual(updated["description"], "An updated test group.") + self.assertEqual(updated["owner"], self.test_user.id) + self.assertEqual(updated["members"], []) + self.assertEqual(updated["access"], "MANUAL") - check2 = self.retrieve_unwrap("gamma") + retrieved2 = self.retrieve_unwrap("gamma") - self.assertEqual(check2["slug"], "gamma") - self.assertEqual(check2["name"], "Gamma") - self.assertEqual(check2["description"], "An updated test group.") - self.assertEqual(check2["owner"], self.test_user.id) - self.assertEqual(check2["members"], []) - self.assertEqual(check2["access"], "MANUAL") + self.assertEqual(retrieved2["slug"], "gamma") + self.assertEqual(retrieved2["name"], "Gamma") + self.assertEqual(retrieved2["description"], "An updated test group.") + self.assertEqual(retrieved2["owner"], self.test_user.id) + self.assertEqual(retrieved2["members"], []) + self.assertEqual(retrieved2["access"], "MANUAL") def test_update_not_logged_in(self): - result = self.update_unwrap("alpha", { + self.update_fail("alpha", { "slug": "alpha", "name": "AAAAA", "description": "An hacker has updated the Alpha group without permissions!", "members": [], "access": "MANUAL", - }) + }, 401) - self.assertIn("slug", result) - self.assertIn("name", result) - self.assertIn("description", result) - self.assertIn("owner", result) - self.assertIn("members", result) - self.assertIn("access", result) + def test_update_unauthorized(self): + self.client.login(username="TAST", password="TheGreatDjangoTast") - self.assertEqual(result["slug"], "alpha") - self.assertEqual(result["name"], "Alpha") - self.assertEqual(result["description"], "First test group.") - self.assertEqual(result["owner"], self.test_user.id) - self.assertEqual(result["members"], []) - self.assertEqual(result["access"], "MANUAL") + self.update_fail("alpha", { + "slug": "alpha", + "name": "AAAAA", + "description": "An hacker has updated the Alpha group without permissions!", + "members": [], + "access": "MANUAL", + }, 403) def test_update_invalid_schema(self): - result = self.update_unwrap("alpha", { + self.client.login(username="TEST", password="TheGreatDjangoTest") + + self.update_fail("alpha", { "hahaha": "soccer", + }, 400) + + def test_destroy_valid(self): + self.client.login(username="TEST", password="TheGreatDjangoTest") + + self.create_unwrap({ + "slug": "boom", + "name": "Boom!!!", + "description": "A group that should explode.", + "members": [], + "access": "OPEN", }) - self.assertIn("slug", result) - self.assertIn("name", result) - self.assertIn("description", result) - self.assertIn("owner", result) - self.assertIn("members", result) - self.assertIn("access", result) + self.destroy_unwrap("boom") + self.retrieve_fail("boom", 404) - self.assertEqual(result["slug"], "alpha") - self.assertEqual(result["name"], "Alpha") - self.assertEqual(result["description"], "First test group.") - self.assertEqual(result["owner"], self.test_user.id) - self.assertEqual(result["members"], []) - self.assertEqual(result["access"], "MANUAL") + def test_destroy_not_logged_in(self): + self.client.login(username="TEST", password="TheGreatDjangoTest") - # TODO: Create destroy test + self.create_unwrap({ + "slug": "boom", + "name": "Boom!!!", + "description": "A group that should explode.", + "members": [], + "access": "OPEN", + }) + + self.client.logout() + + self.destroy_fail("boom", 401) + self.retrieve_unwrap("boom") + + def test_destroy_unauthorized(self): + self.client.login(username="TEST", password="TheGreatDjangoTest") + + self.create_unwrap({ + "slug": "doom", + "name": "Doom!!!", + "description": "A group about a game.", + "members": [], + "access": "OPEN", + }) + + self.client.logout() + self.client.login(username="TAST", password="TheGreatDjangoTast") + + self.destroy_fail("doom", 403) + self.retrieve_unwrap("doom") diff --git a/backend/sophon/core/urls.py b/backend/sophon/core/urls.py index 06f6d30..60865cb 100644 --- a/backend/sophon/core/urls.py +++ b/backend/sophon/core/urls.py @@ -5,8 +5,8 @@ from . import views router = rest_framework.routers.DefaultRouter() router.register("groups", views.ResearchGroupViewSet, basename="research-group") -router.register("users", views.UserViewSet, basename="user") - +router.register("users/by-id", views.UsersByIdViewSet, basename="user-by-id") +router.register("users/by-username", views.UsersByUsernameViewSet, basename="user-by-username") urlpatterns = [ path("", include(router.urls)), diff --git a/backend/sophon/core/views.py b/backend/sophon/core/views.py index 5c05cdd..ab317cb 100644 --- a/backend/sophon/core/views.py +++ b/backend/sophon/core/views.py @@ -3,6 +3,7 @@ import typing as t import deprecation from django.contrib.auth.models import User +from django.db.models import QuerySet from rest_framework import status as s from rest_framework.decorators import action from rest_framework.response import Response @@ -10,30 +11,24 @@ from rest_framework.serializers import Serializer from rest_framework.views import APIView from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet +from . import errors from . import models from . import permissions from . import serializers -class HTTPException(Exception): - """ - An exception that can be raised in :class:`.SophonViewSet` hooks to respond to a request with an HTTP error. - """ - def __init__(self, status: int): - self.status = status - - def as_response(self) -> Response: - return Response(status=self.status) - - class ReadSophonViewSet(ReadOnlyModelViewSet, metaclass=abc.ABCMeta): """ - An extension to :class:`~rest_framework.viewsets.ReadOnlyModelViewSet` including some essential (but missing) methods. + An extension to :class:`~rest_framework.viewsets.ReadOnlyModelViewSet` that includes some essential (but missing) methods. """ - # A QuerySet should be specified, probably @abc.abstractmethod - def get_queryset(self): + def get_queryset(self) -> QuerySet: + """ + :return: The :class:`~django.db.models.QuerySet` to use in the API request. + + .. note:: Ensure the requesting user can view the objects in the queryset! + """ raise NotImplementedError() # Override the permission_classes property with this hack, as ModelViewSet doesn't have the get_permission_classes method yet @@ -43,14 +38,16 @@ class ReadSophonViewSet(ReadOnlyModelViewSet, metaclass=abc.ABCMeta): # noinspection PyMethodMayBeStatic def get_permission_classes(self) -> t.Collection[t.Type[permissions.BasePermission]]: - """ - The "method" version of the :attr:`~rest_framework.viewsets.ModelViewSet.permission_classes` property. + return [permissions.AllowAny] - :return: A collection of permission classes. + def get_serializer_class(self) -> t.Type[Serializer]: """ - return permissions.AllowAny, + :return: The :class:`~rest_framework.serializers.Serializer` to use in the API request. - def get_serializer_class(self): + .. note:: You probably won't need to edit this; the serializer is usually automatically selected based on the access that the user performing the request has on the requested resource. + + .. seealso:: :meth:`.models.SophonModel.get_access_serializer` and :meth:`.get_custom_serializer_classes` + """ if self.action in ["list"]: return self.get_queryset().model.get_view_serializer() if self.action in ["create", "metadata"]: @@ -61,9 +58,9 @@ class ReadSophonViewSet(ReadOnlyModelViewSet, metaclass=abc.ABCMeta): return self.get_custom_serializer_classes() # noinspection PyMethodMayBeStatic - def get_custom_serializer_classes(self): + def get_custom_serializer_classes(self) -> t.Type[Serializer]: """ - .. todo:: Define this. + :return: The :class:`~rest_framework.serializers.Serializer` to use in the API request if a custom action is being run. """ return serializers.NoneSerializer @@ -79,8 +76,8 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta) Hook called on ``create`` actions after the serializer is validated but before it is saved. :param serializer: The validated serializer containing the data of the object about to be created. - :raises HTTPException: If the request should be answered with an error. - :return: A :class:`dict` of fields to be added / overriden to the object saved by the serializer. + :raises errors.HTTPException: If the request should be answered with an error. + :return: A :class:`dict` of fields to be merged to the object saved by the serializer. """ return {} @@ -88,16 +85,21 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta) def perform_create(self, serializer): """ .. warning:: This function does nothing and may not be called on :class:`SophonViewSet`\\ s. + + :raises RuntimeError: Always. """ raise RuntimeError(f"`perform_create` may not be called on `SophonViewSet`s.") - def create(self, request, *args, **kwargs): + def create(self, request, *args, **kwargs) -> Response: + """ + Custom version of :meth:`rest_framework.viewsets.ModelViewSet.create` that allows adding fields to the serializer before it is saved and interrupting the request with an exception. + """ serializer: Serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) try: hook = self.hook_create(serializer) - except HTTPException as e: + except errors.HTTPException as e: return e.as_response() serializer.save(**hook) @@ -119,10 +121,16 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta) def perform_update(self, serializer): """ .. warning:: This function does nothing and may not be called on :class:`SophonViewSet`\\ s. + + :raises RuntimeError: Always. """ raise RuntimeError(f"`perform_update` may not be called on `SophonViewSet`s.") def update(self, request, *args, **kwargs): + """ + Custom version of :meth:`rest_framework.viewsets.ModelViewSet.update` that allows adding fields to the serializer before it is saved and interrupting the request with an exception. + """ + partial = kwargs.pop('partial', False) instance = self.get_object() serializer = self.get_serializer(instance, data=request.data, partial=partial) @@ -130,7 +138,7 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta) try: hook = self.hook_update(serializer) - except HTTPException as e: + except errors.HTTPException as e: return e.as_response() serializer.save(**hook) @@ -155,61 +163,91 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta) def perform_destroy(self, serializer): """ .. warning:: This function does nothing and may not be called on :class:`SophonViewSet`\\ s. + + :raises RuntimeError: Always. """ raise RuntimeError(f"`perform_destroy` may not be called on `SophonViewSet`s.") def destroy(self, request, *args, **kwargs): + """ + Custom version of :meth:`rest_framework.viewsets.ModelViewSet.update` that allows interrupting the request with an exception. + """ instance = self.get_object() try: self.hook_destroy() - except HTTPException as e: + except errors.HTTPException as e: return e.as_response() instance.delete() return Response(status=s.HTTP_204_NO_CONTENT) -class UserViewSet(ReadSophonViewSet): +class UsersByIdViewSet(ReadSophonViewSet): """ - A viewset to list registered users. + A read-only ViewSet that displays the users registered to the Sophon instance, accessible by id. """ + + def get_queryset(self): + return User.objects.order_by("id").all() + + def get_serializer_class(self): + return serializers.UserSerializer + + def get_object(self): + pk = self.kwargs["pk"] + return User.objects.filter(id=pk).get() + + +class UsersByUsernameViewSet(ReadSophonViewSet): + """ + A read-only ViewSet that displays the users registered to the Sophon instance, accessible by username. + """ + def get_queryset(self): return User.objects.order_by("username").all() def get_serializer_class(self): return serializers.UserSerializer - # Small hack to make users listable by username - # Might break if someone's username is all-numbers, but I'm not sure Django allows that def get_object(self): pk = self.kwargs["pk"] - try: - pk = int(pk) - except ValueError: - return User.objects.filter(username=pk).get() - else: - return User.objects.filter(id=pk).get() + return User.objects.filter(username=pk).get() class ResearchGroupViewSet(WriteSophonViewSet): """ - The viewset for :class:`~.models.ResearchGroup`\\ s. + A ViewSet that allows interactions with the Sophon Research Group. """ def get_queryset(self): - # All research groups are public, so it's fine to do this return models.ResearchGroup.objects.order_by("slug").all() def hook_create(self, serializer) -> dict[str, t.Any]: + # Disallow group creation from anonymous users if self.request.user.is_anonymous: - raise HTTPException(status=403) + raise errors.HTTPException(status=401) # Add the owner field to the serializer return { "owner": self.request.user, } + def hook_destroy(self) -> None: + group: models.ResearchGroup = models.ResearchGroup.objects.get(pk=self.kwargs["pk"]) + + # Disallow group destruction if the user doesn't have admin access to the group + if not group.can_admin(self.request.user): + raise errors.HTTPException(s.HTTP_403_FORBIDDEN) + + def get_permission_classes(self) -> t.Collection[t.Type[permissions.BasePermission]]: + if self.action in ["destroy"]: + return permissions.Admin, + elif self.action in ["update", "partial_update"]: + return permissions.Edit, + else: + return permissions.AllowAny, + def get_custom_serializer_classes(self): if self.action in ["join", "leave"]: return self.get_object().get_access_serializer(self.request.user) @@ -231,9 +269,8 @@ class ResearchGroupViewSet(WriteSophonViewSet): # Add the user to the group group.members.add(self.request.user) - # noinspection PyPep8Naming - Serializer = group.get_access_serializer(self.request.user) - serializer = Serializer(instance=group) + serializer_class = group.get_access_serializer(self.request.user) + serializer = serializer_class(instance=group) return Response(data=serializer.data, status=s.HTTP_200_OK) @@ -252,9 +289,8 @@ class ResearchGroupViewSet(WriteSophonViewSet): # Add the user to the group group.members.remove(self.request.user) - # noinspection PyPep8Naming - Serializer = group.get_access_serializer(self.request.user) - serializer = Serializer(instance=group) + serializer_class = group.get_access_serializer(self.request.user) + serializer = serializer_class(instance=group) return Response(data=serializer.data, status=s.HTTP_200_OK) @@ -276,7 +312,7 @@ class SophonGroupViewSet(WriteSophonViewSet, metaclass=abc.ABCMeta): # Allow creation of objects only on groups the user has Edit access on group = self.get_group_from_serializer(serializer) if not group.can_edit(self.request.user): - raise HTTPException(s.HTTP_403_FORBIDDEN) + raise errors.HTTPException(s.HTTP_403_FORBIDDEN) return {} @@ -284,7 +320,7 @@ class SophonGroupViewSet(WriteSophonViewSet, metaclass=abc.ABCMeta): # Allow group transfers only to groups the user has Edit access on group: models.ResearchGroup = self.get_group_from_serializer(serializer) if not group.can_edit(self.request.user): - raise HTTPException(s.HTTP_403_FORBIDDEN) + raise errors.HTTPException(s.HTTP_403_FORBIDDEN) return {}