steffo/sophon
steffo/sophon
1
Fork 0
mirror of https://github.com/Steffo99/sophon.git synced 2024-12-22 14:54:22 +00:00
Code Activity

💥 Fix tests and bugs

Browse source
This commit is contained in:
Steffo 2021-10-27 20:26:47 +02:00
parent 7b52f3381c
commit f5efbbcb9d
5 changed files with 249 additions and 161 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
.idea/inspectionProfiles/Project_Default.xml
View file

@ -37,6 +37,13 @@
</value> </value>
</option> </option>
</inspection_tool> </inspection_tool>
<inspection_tool class="PyPep8Inspection" enabled="true" level="WEAK WARNING" enabled_by_default="true">
<option name="ignoredErrors">
<list>
<option value="E501" />
</list>
</option>
</inspection_tool>
<inspection_tool class="PyRelativeImportInspection" enabled="true" level="WARNING" enabled_by_default="true" /> <inspection_tool class="PyRelativeImportInspection" enabled="true" level="WARNING" enabled_by_default="true" />
<inspection_tool class="PyReturnFromInitInspection" enabled="true" level="ERROR" enabled_by_default="true" /> <inspection_tool class="PyReturnFromInitInspection" enabled="true" level="ERROR" enabled_by_default="true" />
<inspection_tool class="PySetFunctionToLiteralInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true" /> <inspection_tool class="PySetFunctionToLiteralInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true" />

13
backend/sophon/core/errors.py Normal file
View file

@ -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)

254
backend/sophon/core/tests.py
View file

@ -30,9 +30,9 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta):
self.assertTrue(isinstance(response.data, dict)) self.assertTrue(isinstance(response.data, dict))
return response.data return response.data
def list_fail(self) -> None: def list_fail(self, code) -> None:
response = self.list() response = self.list()
self.assertTrue(response.status_code >= 400) self.assertEqual(response.status_code, code)
def retrieve(self, pk) -> Response: def retrieve(self, pk) -> Response:
url = self.get_url("detail", pk=pk) url = self.get_url("detail", pk=pk)
@ -44,9 +44,9 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta):
self.assertTrue(isinstance(response.data, dict)) self.assertTrue(isinstance(response.data, dict))
return response.data return response.data
def retrieve_fail(self, pk) -> None: def retrieve_fail(self, pk, code) -> None:
response = self.retrieve(pk=pk) response = self.retrieve(pk=pk)
self.assertTrue(response.status_code >= 400) self.assertEqual(response.status_code, code)
def create(self, data) -> Response: def create(self, data) -> Response:
url = self.get_url("list") url = self.get_url("list")
@ -58,9 +58,9 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta):
self.assertTrue(isinstance(response.data, dict)) self.assertTrue(isinstance(response.data, dict))
return response.data return response.data
def create_fail(self, data) -> None: def create_fail(self, data, code) -> None:
response = self.create(data) response = self.create(data)
self.assertTrue(response.status_code >= 400) self.assertEqual(response.status_code, code)
def update(self, pk, data) -> Response: def update(self, pk, data) -> Response:
url = self.get_url("detail", pk=pk) url = self.get_url("detail", pk=pk)
@ -72,6 +72,10 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta):
self.assertTrue(isinstance(response.data, dict)) self.assertTrue(isinstance(response.data, dict))
return response.data 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: def destroy(self, pk) -> Response:
url = self.get_url("detail", pk=pk) url = self.get_url("detail", pk=pk)
return self.client.delete(url, format="json") return self.client.delete(url, format="json")
@ -80,9 +84,9 @@ class SophonModelTestCase(APITestCase, metaclass=abc.ABCMeta):
response = self.destroy(pk=pk) response = self.destroy(pk=pk)
self.assertEqual(response.status_code, 204) self.assertEqual(response.status_code, 204)
def destroy_fail(self, pk) -> None: def destroy_fail(self, pk, code) -> None:
response = self.destroy(pk) response = self.destroy(pk)
self.assertTrue(response.status_code >= 400) self.assertEqual(response.status_code, code)
class ResearchGroupTests(SophonModelTestCase): class ResearchGroupTests(SophonModelTestCase):
@ -91,10 +95,12 @@ class ResearchGroupTests(SophonModelTestCase):
return "research-group" return "research-group"
test_user: User = None test_user: User = None
other_user: User = None
@classmethod @classmethod
def setUpTestData(cls): def setUpTestData(cls):
cls.test_user = User.objects.create_user(username="TEST", password="TheGreatDjangoTest") 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( models.ResearchGroup.objects.create(
slug="alpha", slug="alpha",
@ -118,66 +124,67 @@ class ResearchGroupTests(SophonModelTestCase):
count = r["count"] count = r["count"]
self.assertEqual(count, 2) self.assertEqual(count, 2)
results = r["results"] list_page = r["results"]
self.assertIn("slug", results[0]) self.assertIn("slug", list_page[0])
self.assertIn("name", results[0]) self.assertIn("name", list_page[0])
self.assertIn("description", results[0]) self.assertIn("description", list_page[0])
self.assertIn("owner", results[0]) self.assertIn("owner", list_page[0])
self.assertIn("members", results[0]) self.assertIn("members", list_page[0])
self.assertIn("access", results[0]) self.assertIn("access", list_page[0])
def test_retrieve_valid(self): def test_retrieve_valid(self):
result = self.retrieve_unwrap("alpha") retrieved = self.retrieve_unwrap("alpha")
self.assertIn("slug", result) self.assertIn("slug", retrieved)
self.assertIn("name", result) self.assertIn("name", retrieved)
self.assertIn("description", result) self.assertIn("description", retrieved)
self.assertIn("owner", result) self.assertIn("owner", retrieved)
self.assertIn("members", result) self.assertIn("members", retrieved)
self.assertIn("access", result) self.assertIn("access", retrieved)
self.assertEqual(result["slug"], "alpha") self.assertEqual(retrieved["slug"], "alpha")
self.assertEqual(result["name"], "Alpha") self.assertEqual(retrieved["name"], "Alpha")
self.assertEqual(result["description"], "First test group.") self.assertEqual(retrieved["description"], "First test group.")
self.assertEqual(result["owner"], self.test_user.id) self.assertEqual(retrieved["owner"], self.test_user.id)
self.assertEqual(result["members"], []) self.assertEqual(retrieved["members"], [])
self.assertEqual(result["access"], "MANUAL") self.assertEqual(retrieved["access"], "MANUAL")
def test_retrieve_not_existing(self): def test_retrieve_not_existing(self):
self.retrieve_fail("banana") self.retrieve_fail("banana", 404)
def test_create_valid(self): def test_create_valid(self):
self.client.login(username="TEST", password="TheGreatDjangoTest") self.client.login(username="TEST", password="TheGreatDjangoTest")
result = self.create_unwrap({ created = self.create_unwrap({
"slug": "omega", "slug": "omega",
"name": "Omega", "name": "Omega",
"description": "Last test group.", "description": "Last test group.",
"members": [], "members": [],
"access": "OPEN", "access": "OPEN",
}) })
self.assertIn("slug", result) self.assertIn("slug", created)
self.assertIn("name", result) self.assertIn("name", created)
self.assertIn("description", result) self.assertIn("description", created)
self.assertIn("members", result) self.assertIn("owner", created)
self.assertIn("access", result) self.assertIn("members", created)
self.assertIn("access", created)
check = self.retrieve_unwrap("omega") retrieved = self.retrieve_unwrap("omega")
self.assertIn("slug", check) self.assertIn("slug", retrieved)
self.assertIn("name", check) self.assertIn("name", retrieved)
self.assertIn("description", check) self.assertIn("description", retrieved)
self.assertIn("owner", check) self.assertIn("owner", retrieved)
self.assertIn("members", check) self.assertIn("members", retrieved)
self.assertIn("access", check) self.assertIn("access", retrieved)
self.assertEqual(check["slug"], "omega") self.assertEqual(retrieved["slug"], "omega")
self.assertEqual(check["name"], "Omega") self.assertEqual(retrieved["name"], "Omega")
self.assertEqual(check["description"], "Last test group.") self.assertEqual(retrieved["description"], "Last test group.")
self.assertEqual(result["owner"], self.test_user.id) self.assertEqual(retrieved["owner"], self.test_user.id)
self.assertEqual(result["members"], []) self.assertEqual(retrieved["members"], [])
self.assertEqual(result["access"], "OPEN") self.assertEqual(retrieved["access"], "OPEN")
def test_create_not_logged_in(self): def test_create_not_logged_in(self):
self.create_fail({ self.create_fail({
@ -186,7 +193,7 @@ class ResearchGroupTests(SophonModelTestCase):
"description": "This creation should fail.", "description": "This creation should fail.",
"members": [], "members": [],
"access": "OPEN", "access": "OPEN",
}) }, 401)
def test_create_invalid_schema(self): def test_create_invalid_schema(self):
self.client.login(username="TEST", password="TheGreatDjangoTest") self.client.login(username="TEST", password="TheGreatDjangoTest")
@ -194,34 +201,29 @@ class ResearchGroupTests(SophonModelTestCase):
self.create_fail({ self.create_fail({
"potato": "sweet", "potato": "sweet",
"access": "OPEN", "access": "OPEN",
}) }, 400)
def test_update_valid(self): def test_update_valid(self):
self.client.login(username="TEST", password="TheGreatDjangoTest") self.client.login(username="TEST", password="TheGreatDjangoTest")
creation = self.create_unwrap({ self.create_unwrap({
"slug": "gamma", "slug": "gamma",
"name": "Gamma", "name": "Gamma",
"description": "A test group to update.", "description": "A test group to update.",
"members": [], "members": [],
"access": "OPEN", "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(retrieved["slug"], "gamma")
self.assertEqual(check["name"], "Gamma") self.assertEqual(retrieved["name"], "Gamma")
self.assertEqual(check["description"], "A test group to update.") self.assertEqual(retrieved["description"], "A test group to update.")
self.assertEqual(check["owner"], self.test_user.id) self.assertEqual(retrieved["owner"], self.test_user.id)
self.assertEqual(check["members"], []) self.assertEqual(retrieved["members"], [])
self.assertEqual(check["access"], "OPEN") self.assertEqual(retrieved["access"], "OPEN")
update = self.update_unwrap("gamma", { updated = self.update_unwrap("gamma", {
"slug": "gamma", "slug": "gamma",
"name": "Gamma", "name": "Gamma",
"description": "An updated test group.", "description": "An updated test group.",
@ -229,69 +231,99 @@ class ResearchGroupTests(SophonModelTestCase):
"access": "MANUAL", "access": "MANUAL",
}) })
self.assertIn("slug", update) self.assertIn("slug", updated)
self.assertIn("name", update) self.assertIn("name", updated)
self.assertIn("description", update) self.assertIn("description", updated)
self.assertIn("owner", update) self.assertIn("owner", updated)
self.assertIn("members", update) self.assertIn("members", updated)
self.assertIn("access", update) self.assertIn("access", updated)
self.assertEqual(update["slug"], "gamma") self.assertEqual(updated["slug"], "gamma")
self.assertEqual(update["name"], "Gamma") self.assertEqual(updated["name"], "Gamma")
self.assertEqual(update["description"], "An updated test group.") self.assertEqual(updated["description"], "An updated test group.")
self.assertEqual(update["owner"], self.test_user.id) self.assertEqual(updated["owner"], self.test_user.id)
self.assertEqual(update["members"], []) self.assertEqual(updated["members"], [])
self.assertEqual(update["access"], "MANUAL") self.assertEqual(updated["access"], "MANUAL")
check2 = self.retrieve_unwrap("gamma") retrieved2 = self.retrieve_unwrap("gamma")
self.assertEqual(check2["slug"], "gamma") self.assertEqual(retrieved2["slug"], "gamma")
self.assertEqual(check2["name"], "Gamma") self.assertEqual(retrieved2["name"], "Gamma")
self.assertEqual(check2["description"], "An updated test group.") self.assertEqual(retrieved2["description"], "An updated test group.")
self.assertEqual(check2["owner"], self.test_user.id) self.assertEqual(retrieved2["owner"], self.test_user.id)
self.assertEqual(check2["members"], []) self.assertEqual(retrieved2["members"], [])
self.assertEqual(check2["access"], "MANUAL") self.assertEqual(retrieved2["access"], "MANUAL")
def test_update_not_logged_in(self): def test_update_not_logged_in(self):
result = self.update_unwrap("alpha", { self.update_fail("alpha", {
"slug": "alpha", "slug": "alpha",
"name": "AAAAA", "name": "AAAAA",
"description": "An hacker has updated the Alpha group without permissions!", "description": "An hacker has updated the Alpha group without permissions!",
"members": [], "members": [],
"access": "MANUAL", "access": "MANUAL",
}) }, 401)
self.assertIn("slug", result) def test_update_unauthorized(self):
self.assertIn("name", result) self.client.login(username="TAST", password="TheGreatDjangoTast")
self.assertIn("description", result)
self.assertIn("owner", result)
self.assertIn("members", result)
self.assertIn("access", result)
self.assertEqual(result["slug"], "alpha") self.update_fail("alpha", {
self.assertEqual(result["name"], "Alpha") "slug": "alpha",
self.assertEqual(result["description"], "First test group.") "name": "AAAAA",
self.assertEqual(result["owner"], self.test_user.id) "description": "An hacker has updated the Alpha group without permissions!",
self.assertEqual(result["members"], []) "members": [],
self.assertEqual(result["access"], "MANUAL") "access": "MANUAL",
}, 403)
def test_update_invalid_schema(self): def test_update_invalid_schema(self):
result = self.update_unwrap("alpha", { self.client.login(username="TEST", password="TheGreatDjangoTest")
self.update_fail("alpha", {
"hahaha": "soccer", "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.destroy_unwrap("boom")
self.assertIn("name", result) self.retrieve_fail("boom", 404)
self.assertIn("description", result)
self.assertIn("owner", result)
self.assertIn("members", result)
self.assertIn("access", result)
self.assertEqual(result["slug"], "alpha") def test_destroy_not_logged_in(self):
self.assertEqual(result["name"], "Alpha") self.client.login(username="TEST", password="TheGreatDjangoTest")
self.assertEqual(result["description"], "First test group.")
self.assertEqual(result["owner"], self.test_user.id)
self.assertEqual(result["members"], [])
self.assertEqual(result["access"], "MANUAL")
# 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")

4
backend/sophon/core/urls.py
View file

@ -5,8 +5,8 @@ from . import views
router = rest_framework.routers.DefaultRouter() router = rest_framework.routers.DefaultRouter()
router.register("groups", views.ResearchGroupViewSet, basename="research-group") 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 = [ urlpatterns = [
path("", include(router.urls)), path("", include(router.urls)),

132
backend/sophon/core/views.py
View file

@ -3,6 +3,7 @@ import typing as t
import deprecation import deprecation
from django.contrib.auth.models import User from django.contrib.auth.models import User
from django.db.models import QuerySet
from rest_framework import status as s from rest_framework import status as s
from rest_framework.decorators import action from rest_framework.decorators import action
from rest_framework.response import Response 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.views import APIView
from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet
from . import errors
from . import models from . import models
from . import permissions from . import permissions
from . import serializers 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): 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 @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() raise NotImplementedError()
# Override the permission_classes property with this hack, as ModelViewSet doesn't have the get_permission_classes method yet # 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 # noinspection PyMethodMayBeStatic
def get_permission_classes(self) -> t.Collection[t.Type[permissions.BasePermission]]: def get_permission_classes(self) -> t.Collection[t.Type[permissions.BasePermission]]:
""" return [permissions.AllowAny]
The "method" version of the :attr:`~rest_framework.viewsets.ModelViewSet.permission_classes` property.
: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"]: if self.action in ["list"]:
return self.get_queryset().model.get_view_serializer() return self.get_queryset().model.get_view_serializer()
if self.action in ["create", "metadata"]: if self.action in ["create", "metadata"]:
@ -61,9 +58,9 @@ class ReadSophonViewSet(ReadOnlyModelViewSet, metaclass=abc.ABCMeta):
return self.get_custom_serializer_classes() return self.get_custom_serializer_classes()
# noinspection PyMethodMayBeStatic # 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 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. 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. :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. :raises errors.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. :return: A :class:`dict` of fields to be merged to the object saved by the serializer.
""" """
return {} return {}
@ -88,16 +85,21 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta)
def perform_create(self, serializer): def perform_create(self, serializer):
""" """
.. warning:: This function does nothing and may not be called on :class:`SophonViewSet`\\ s. .. 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.") 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: Serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True) serializer.is_valid(raise_exception=True)
try: try:
hook = self.hook_create(serializer) hook = self.hook_create(serializer)
except HTTPException as e: except errors.HTTPException as e:
return e.as_response() return e.as_response()
serializer.save(**hook) serializer.save(**hook)
@ -119,10 +121,16 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta)
def perform_update(self, serializer): def perform_update(self, serializer):
""" """
.. warning:: This function does nothing and may not be called on :class:`SophonViewSet`\\ s. .. 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.") raise RuntimeError(f"`perform_update` may not be called on `SophonViewSet`s.")
def update(self, request, *args, **kwargs): 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) partial = kwargs.pop('partial', False)
instance = self.get_object() instance = self.get_object()
serializer = self.get_serializer(instance, data=request.data, partial=partial) serializer = self.get_serializer(instance, data=request.data, partial=partial)
@ -130,7 +138,7 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta)
try: try:
hook = self.hook_update(serializer) hook = self.hook_update(serializer)
except HTTPException as e: except errors.HTTPException as e:
return e.as_response() return e.as_response()
serializer.save(**hook) serializer.save(**hook)
@ -155,61 +163,91 @@ class WriteSophonViewSet(ModelViewSet, ReadSophonViewSet, metaclass=abc.ABCMeta)
def perform_destroy(self, serializer): def perform_destroy(self, serializer):
""" """
.. warning:: This function does nothing and may not be called on :class:`SophonViewSet`\\ s. .. 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.") raise RuntimeError(f"`perform_destroy` may not be called on `SophonViewSet`s.")
def destroy(self, request, *args, **kwargs): 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() instance = self.get_object()
try: try:
self.hook_destroy() self.hook_destroy()
except HTTPException as e: except errors.HTTPException as e:
return e.as_response() return e.as_response()
instance.delete() instance.delete()
return Response(status=s.HTTP_204_NO_CONTENT) 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): def get_queryset(self):
return User.objects.order_by("username").all() return User.objects.order_by("username").all()
def get_serializer_class(self): def get_serializer_class(self):
return serializers.UserSerializer 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): def get_object(self):
pk = self.kwargs["pk"] pk = self.kwargs["pk"]
try: return User.objects.filter(username=pk).get()
pk = int(pk)
except ValueError:
return User.objects.filter(username=pk).get()
else:
return User.objects.filter(id=pk).get()
class ResearchGroupViewSet(WriteSophonViewSet): class ResearchGroupViewSet(WriteSophonViewSet):
""" """
The viewset for :class:`~.models.ResearchGroup`\\ s. A ViewSet that allows interactions with the Sophon Research Group.
""" """
def get_queryset(self): def get_queryset(self):
# All research groups are public, so it's fine to do this
return models.ResearchGroup.objects.order_by("slug").all() return models.ResearchGroup.objects.order_by("slug").all()
def hook_create(self, serializer) -> dict[str, t.Any]: def hook_create(self, serializer) -> dict[str, t.Any]:
# Disallow group creation from anonymous users
if self.request.user.is_anonymous: if self.request.user.is_anonymous:
raise HTTPException(status=403) raise errors.HTTPException(status=401)
# Add the owner field to the serializer # Add the owner field to the serializer
return { return {
"owner": self.request.user, "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): def get_custom_serializer_classes(self):
if self.action in ["join", "leave"]: if self.action in ["join", "leave"]:
return self.get_object().get_access_serializer(self.request.user) return self.get_object().get_access_serializer(self.request.user)
@ -231,9 +269,8 @@ class ResearchGroupViewSet(WriteSophonViewSet):
# Add the user to the group # Add the user to the group
group.members.add(self.request.user) group.members.add(self.request.user)
# noinspection PyPep8Naming serializer_class = group.get_access_serializer(self.request.user)
Serializer = group.get_access_serializer(self.request.user) serializer = serializer_class(instance=group)
serializer = Serializer(instance=group)
return Response(data=serializer.data, status=s.HTTP_200_OK) return Response(data=serializer.data, status=s.HTTP_200_OK)
@ -252,9 +289,8 @@ class ResearchGroupViewSet(WriteSophonViewSet):
# Add the user to the group # Add the user to the group
group.members.remove(self.request.user) group.members.remove(self.request.user)
# noinspection PyPep8Naming serializer_class = group.get_access_serializer(self.request.user)
Serializer = group.get_access_serializer(self.request.user) serializer = serializer_class(instance=group)
serializer = Serializer(instance=group)
return Response(data=serializer.data, status=s.HTTP_200_OK) 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 # Allow creation of objects only on groups the user has Edit access on
group = self.get_group_from_serializer(serializer) group = self.get_group_from_serializer(serializer)
if not group.can_edit(self.request.user): if not group.can_edit(self.request.user):
raise HTTPException(s.HTTP_403_FORBIDDEN) raise errors.HTTPException(s.HTTP_403_FORBIDDEN)
return {} return {}
@ -284,7 +320,7 @@ class SophonGroupViewSet(WriteSophonViewSet, metaclass=abc.ABCMeta):
# Allow group transfers only to groups the user has Edit access on # Allow group transfers only to groups the user has Edit access on
group: models.ResearchGroup = self.get_group_from_serializer(serializer) group: models.ResearchGroup = self.get_group_from_serializer(serializer)
if not group.can_edit(self.request.user): if not group.can_edit(self.request.user):
raise HTTPException(s.HTTP_403_FORBIDDEN) raise errors.HTTPException(s.HTTP_403_FORBIDDEN)
return {} return {}