feat(users): add transactional employee deletion and token revocation

tests/apps/user/test_management_admin.py

@@ -0,0 +1,122 @@
+"""Django admin must share the account deletion and access guards."""
+
+from unittest.mock import patch
+
+from apps.core.exceptions import BadRequestError, ConflictError, PermissionDeniedError
+from apps.user.admin import UserAdmin
+from apps.user.management import ensure_admins_remain, lock_managed_users
+from apps.user.models import User
+from django.contrib.admin import AdminSite
+from django.contrib.admin.models import LogEntry
+from django.test import RequestFactory, TestCase
+from django.urls import reverse
+from rest_framework_simplejwt.token_blacklist.models import OutstandingToken
+from rest_framework_simplejwt.tokens import RefreshToken
+
+from tests.apps.user.factories import UserFactory
+
+
+class ManagedUserAdminTest(TestCase):
+    def setUp(self):
+        self.actor = UserFactory.create_superuser()
+        self.user = UserFactory.create_user()
+        self.model_admin = UserAdmin(User, AdminSite())
+        self.request = RequestFactory().post("/admin/user/user/")
+        self.request.user = self.actor
+        self.initial_user_count = User.objects.count()
+
+    def test_admin_single_and_bulk_delete_clean_up_tokens(self):
+        second = UserFactory.create_user()
+        first_id, second_id = self.user.pk, second.pk
+        RefreshToken.for_user(self.user)
+        RefreshToken.for_user(second)
+        self.model_admin.delete_model(self.request, self.user)
+        self.model_admin.delete_queryset(
+            self.request, User.objects.filter(pk=second_id)
+        )
+        self.assertFalse(User.objects.filter(pk__in=[first_id, second_id]).exists())
+        self.assertFalse(
+            OutstandingToken.objects.filter(user_id__in=[first_id, second_id]).exists()
+        )
+
+    def test_bulk_delete_is_atomic_when_selection_includes_actor(self):
+        with self.assertRaises(BadRequestError):
+            self.model_admin.delete_queryset(self.request, User.objects.all())
+        self.assertEqual(User.objects.count(), self.initial_user_count)
+
+    def test_admin_bulk_deactivate_and_change_form_cannot_disable_self(self):
+        with self.assertRaises(BadRequestError):
+            self.model_admin.deactivate_users(self.request, User.objects.all())
+        self.assertTrue(User.objects.get(pk=self.user.pk).is_active)
+        edited_actor = User.objects.get(pk=self.actor.pk)
+        edited_actor.is_staff = False
+        with self.assertRaises(BadRequestError):
+            self.model_admin.save_model(self.request, edited_actor, None, True)
+        self.assertTrue(User.objects.get(pk=self.actor.pk).is_staff)
+
+    def test_admin_rechecks_actor_before_saving_or_bulk_deactivating(self):
+        User.objects.filter(pk=self.actor.pk).update(is_active=False)
+        with self.assertRaises(PermissionDeniedError):
+            self.model_admin.save_model(self.request, self.user, None, True)
+        with self.assertRaises(PermissionDeniedError):
+            self.model_admin.deactivate_users(
+                self.request, User.objects.filter(pk=self.user.pk)
+            )
+
+    def test_bulk_deactivation_preserves_other_administrator(self):
+        with patch.object(self.model_admin, "message_user"):
+            self.model_admin.deactivate_users(
+                self.request, User.objects.filter(pk=self.user.pk)
+            )
+        self.assertFalse(User.objects.get(pk=self.user.pk).is_active)
+        self.assertTrue(User.objects.get(pk=self.actor.pk).is_active)
+
+    def test_last_active_administrator_guard_excludes_inactive_staff(self):
+        User.objects.exclude(pk=self.actor.pk).update(is_active=False)
+        inactive = UserFactory.create_user(is_staff=True, is_active=False)
+        with lock_managed_users(
+            actor_id=self.actor.pk, user_ids=[inactive.pk]
+        ), self.assertRaises(ConflictError) as error:
+            ensure_admins_remain(removed_ids=[self.actor.pk])
+        self.assertEqual(error.exception.code, "last_active_admin")
+
+    def test_admin_selected_delete_failure_does_not_publish_false_audit_entries(self):
+        self.client.force_login(self.actor)
+        response = self.client.post(
+            reverse("admin:user_user_changelist"),
+            {
+                "action": "delete_selected",
+                "post": "yes",
+                "_selected_action": [str(self.actor.pk), str(self.user.pk)],
+            },
+        )
+        self.assertEqual(response.status_code, 302)
+        self.assertEqual(User.objects.count(), self.initial_user_count)
+        self.assertEqual(LogEntry.objects.count(), 0)
+
+    def test_admin_single_delete_error_is_presented_without_server_error(self):
+        self.client.force_login(self.actor)
+        response = self.client.post(
+            reverse("admin:user_user_delete", args=[self.actor.pk]), {"post": "yes"}
+        )
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue(User.objects.filter(pk=self.actor.pk).exists())
+        self.assertEqual(LogEntry.objects.count(), 0)
+
+    def test_separate_admin_password_route_rechecks_actor(self):
+        old_hash = self.user.password
+        User.objects.filter(pk=self.actor.pk).update(is_staff=False)
+        request = RequestFactory().post(
+            "/admin/user/user/password/",
+            {
+                "password1": "changed-test-password",
+                "password2": "changed-test-password",
+            },
+        )
+        request.user = self.actor
+        with patch.object(self.model_admin, "message_user") as message:
+            response = self.model_admin.user_change_password(request, str(self.user.pk))
+        self.assertEqual(response.status_code, 302)
+        message.assert_called_once()
+        self.user.refresh_from_db()
+        self.assertEqual(self.user.password, old_hash)