Merge pull request #98 from ImperialCollegeLondon/permissions

Add and use custom permissions.

main/migrations/0002_alter_user_options.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.1.1 on 2024-09-23 16:11
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('main', '0001_initial'),
+ ]
+
+ operations = [
+ migrations.AlterModelOptions(
+ name='user',
+ options={'permissions': [('can_modify_processes', 'Can modify processes'), ('can_view_process_logs', 'Can view process logs')]},
+ ),
+ ]

main/models.py

@@ -1,8 +1,18 @@
 """Models module for the main app."""
 
+from typing import ClassVar
+
 from django.contrib.auth.models import AbstractUser
 from django.db import models # noqa: F401
 
 
 class User(AbstractUser):
  """Custom user model for this project."""
+
+ class Meta:
+ """Meta class for the User model."""
+
+ permissions: ClassVar = [
+ ("can_modify_processes", "Can modify processes"),
+ ("can_view_process_logs", "Can view process logs"),
+ ]

main/views.py

@@ -7,8 +7,8 @@
 
 import django_tables2
 from django.conf import settings
-from django.contrib.auth.decorators import login_required
-from django.contrib.auth.mixins import LoginRequiredMixin
+from django.contrib.auth.decorators import login_required, permission_required
+from django.contrib.auth.mixins import PermissionRequiredMixin
 from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
 from django.shortcuts import render
 from django.urls import reverse, reverse_lazy
@@ -117,6 +117,7 @@ async def _process_call(uuids: list[str], action: ProcessAction) -> None:
 
 
 @login_required
+@permission_required("main.can_modify_processes", raise_exception=True)
 def process_action(request: HttpRequest) -> HttpResponse:
  """Perform an action on the selected processes.
 
@@ -155,6 +156,7 @@ async def _get_process_logs(uuid: str) -> list[DecodedResponse]:
 
 
 @login_required
+@permission_required("main.can_view_process_logs", raise_exception=True)
 def logs(request: HttpRequest, uuid: uuid.UUID) -> HttpResponse:
  """Display the logs of a process.
 
@@ -182,12 +184,13 @@ async def _boot_process(user: str, data: dict[str, str | int]) -> None:
  pass
 
 
-class BootProcessView(LoginRequiredMixin, FormView): # type: ignore [type-arg]
+class BootProcessView(PermissionRequiredMixin, FormView): # type: ignore [type-arg]
  """View for the BootProcess form."""
 
  template_name = "main/boot_process.html"
  form_class = BootProcessForm
  success_url = reverse_lazy("main:index")
+ permission_required = "main.can_modify_processes"
 
  def form_valid(self, form: BootProcessForm) -> HttpResponse:
  """Boot a Process when valid form data has been POSTed.

tests/conftest.py

@@ -1,13 +1,36 @@
 """Configuration for pytest."""
 
 import pytest
+from django.contrib.auth.models import Permission
 from django.test import Client
 
 
 @pytest.fixture
 def auth_client(django_user_model) -> Client:
  """Return an authenticated client."""
- user = django_user_model.objects.create(username="testuser")
+ user = django_user_model.objects.create(username="user")
+ client = Client()
+ client.force_login(user)
+ return client
+
+
+@pytest.fixture
+def auth_process_client(django_user_model) -> Client:
+ """Return a authenticated client with modify process privilege."""
+ user = django_user_model.objects.create(username="process_user")
+ permission = Permission.objects.get(codename="can_modify_processes")
+ user.user_permissions.add(permission)
+ client = Client()
+ client.force_login(user)
+ return client
+
+
+@pytest.fixture
+def auth_logs_client(django_user_model) -> Client:
+ """Return a authenticated client with view logs privilege."""
+ user = django_user_model.objects.create(username="logs_user")
+ permission = Permission.objects.get(codename="can_view_process_logs")
+ user.user_permissions.add(permission)
  client = Client()
  client.force_login(user)
  return client

tests/test_views.py

@@ -49,11 +49,16 @@ class TestLogsView(LoginRequiredTest):
  uuid = uuid4()
  endpoint = reverse("main:logs", kwargs=dict(uuid=uuid))
 
- def test_logs_view_authenticated(self, auth_client, mocker):
- """Test the logs view for an authenticated user."""
+ def test_logs_view_unprivileged(self, auth_client):
+ """Test the logs view for an unprivileged user."""
+ response = auth_client.get(self.endpoint)
+ assert response.status_code == HTTPStatus.FORBIDDEN
+
+ def test_logs_view_privileged(self, auth_logs_client, mocker):
+ """Test the logs view for a privileged user."""
  mock = mocker.patch("main.views._get_process_logs")
  with assertTemplateUsed(template_name="main/logs.html"):
- response = auth_client.get(self.endpoint)
+ response = auth_logs_client.get(self.endpoint)
  assert response.status_code == HTTPStatus.OK
 
  mock.assert_called_once_with(str(self.uuid))
@@ -65,59 +70,75 @@ class TestProcessActionView(LoginRequiredTest):
 
  endpoint = reverse("main:process_action")
 
- def test_process_action_no_action(self, auth_client):
+ def test_process_action_no_action(self, auth_process_client):
  """Test process_action view with no action provided."""
- response = auth_client.post(self.endpoint, data={})
+ response = auth_process_client.post(self.endpoint, data={})
  assert response.status_code == HTTPStatus.FOUND
  assert response.url == reverse("main:index")
 
- def test_process_action_invalid_action(self, auth_client):
+ def test_process_action_invalid_action(self, auth_process_client):
  """Test process_action view with an invalid action."""
- response = auth_client.post(self.endpoint, data={"action": "invalid_action"})
+ response = auth_process_client.post(
+ self.endpoint, data={"action": "invalid_action"}
+ )
  assert response.status_code == HTTPStatus.FOUND
  assert response.url == reverse("main:index")
 
  @pytest.mark.parametrize("action", ["kill", "restart", "flush"])
- def test_process_action_valid_action(self, action, auth_client, mocker):
+ def test_process_action_valid_action(self, action, auth_process_client, mocker):
  """Test process_action view with a valid action."""
  mock = mocker.patch("main.views._process_call")
  uuids_ = [str(uuid4()), str(uuid4())]
- response = auth_client.post(
+ response = auth_process_client.post(
  self.endpoint, data={"action": action, "select": uuids_}
  )
  assert response.status_code == HTTPStatus.FOUND
  assert response.url == reverse("main:index")
 
  mock.assert_called_once_with(uuids_, ProcessAction(action))
 
+ def test_process_action_get_unprivileged(self, auth_client):
+ """Test the GET request for the process_action view (unprivileged)."""
+ response = auth_client.get(reverse("main:boot_process"))
+ assert response.status_code == HTTPStatus.FORBIDDEN
+
 
 class TestBootProcess(LoginRequiredTest):
  """Grouping the tests for the BootProcess view."""
 
  template_name = "main/boot_process.html"
  endpoint = reverse("main:boot_process")
 
- def test_boot_process_get(self, auth_client):
- """Test the GET request for the BootProcess view."""
+ def test_boot_process_get_unprivileged(self, auth_client):
+ """Test the GET request for the BootProcess view (unprivileged)."""
+ response = auth_client.get(reverse("main:boot_process"))
+ assert response.status_code == HTTPStatus.FORBIDDEN
+
+ def test_boot_process_get_privileged(self, auth_process_client):
+ """Test the GET request for the BootProcess view (privileged)."""
  with assertTemplateUsed(template_name=self.template_name):
- response = auth_client.get(reverse("main:boot_process"))
+ response = auth_process_client.get(reverse("main:boot_process"))
  assert response.status_code == HTTPStatus.OK
 
  assert "form" in response.context
  assertContains(response, f'form action="{reverse("main:boot_process")}"')
 
- def test_boot_process_post_invalid(self, auth_client):
+ def test_boot_process_post_invalid(self, auth_process_client):
  """Test the POST request for the BootProcess view with invalid data."""
  with assertTemplateUsed(template_name=self.template_name):
- response = auth_client.post(reverse("main:boot_process"), data=dict())
+ response = auth_process_client.post(
+ reverse("main:boot_process"), data=dict()
+ )
  assert response.status_code == HTTPStatus.OK
 
  assert "form" in response.context
 
- def test_boot_process_post_valid(self, auth_client, mocker, dummy_session_data):
+ def test_boot_process_post_valid(
+ self, auth_process_client, mocker, dummy_session_data
+ ):
  """Test the POST request for the BootProcess view."""
  mock = mocker.patch("main.views._boot_process")
- response = auth_client.post(
+ response = auth_process_client.post(
  reverse("main:boot_process"), data=dummy_session_data
  )
  assert response.status_code == HTTPStatus.FOUND