"""Account deletion keeps business history and revokes authentication.""" from datetime import date from tempfile import TemporaryDirectory from unittest.mock import patch from apps.core.models import BackgroundJob, ReportUpload from apps.exchange.models import ExchangePackageImport from apps.registers.models import Register, RegisterUpload from apps.user.models import Profile, User from apps.user.services import UserService from apps.user.views import AdminUserDetailView from django.contrib.auth.models import Group from django.core.files.base import ContentFile from django.db import transaction from django.db.models.deletion import ProtectedError from django.test import override_settings from django.urls import path, reverse from drf_yasg import openapi from drf_yasg.generators import OpenAPISchemaGenerator from rest_framework.test import APITestCase from rest_framework_simplejwt.token_blacklist.models import ( BlacklistedToken, OutstandingToken, ) from rest_framework_simplejwt.tokens import RefreshToken from tests.apps.user.factories import UserFactory class UserDeletionTest(APITestCase): """Verify deletion through the public administrative API.""" def setUp(self): self.admin = UserFactory.create_superuser() self.user = UserFactory.create_user() self.user_id = self.user.pk self.url = reverse("api_v1:user:admin-user-detail", args=[self.user_id]) self.client.force_authenticate(self.admin) def test_delete_removes_account_and_credentials_but_preserves_history(self): group = Group.objects.create(name="deletion-test") self.user.groups.add(group) refresh = RefreshToken.for_user(self.user) refresh.blacklist() report = ReportUpload.objects.create( form="f1", original_file="reports/keep.xlsx", file_name="keep.xlsx", content_type="application/octet-stream", file_size=1, file_hash="hash", uploaded_by=self.user, ) upload = RegisterUpload.objects.create( registry=Register.objects.create(name="History"), actual_date=date.today(), file_name="keep.xlsx", file_hash="hash", uploaded_by=self.user, ) imported = ExchangePackageImport.objects.create( package_id="history-package", package_name="keep.zip", package_hash="hash", imported_by=self.user, ) job = BackgroundJob.objects.create( task_id="history-task", task_name="test", user_id=self.user_id, ) response = self.client.delete(self.url) self.assertEqual(response.status_code, 204) self.assertEqual(response.content, b"") self.assertFalse(User.objects.filter(pk=self.user_id).exists()) self.assertFalse(Profile.objects.filter(user_id=self.user_id).exists()) self.assertFalse(OutstandingToken.objects.filter(jti=refresh["jti"]).exists()) self.assertEqual(BlacklistedToken.objects.count(), 0) self.assertTrue(Group.objects.filter(pk=group.pk).exists()) self.assertEqual( User.groups.through.objects.filter(user_id=self.user_id).count(), 0 ) for record, field in [ (report, "uploaded_by_id"), (upload, "uploaded_by_id"), (imported, "imported_by_id"), (job, "user_id"), ]: record.refresh_from_db() self.assertIsNone(getattr(record, field)) self.assertEqual(report.original_file.name, "reports/keep.xlsx") self.assertEqual(job.status, "pending") self.assertEqual(self.client.get(self.url).status_code, 404) self.assertEqual(self.client.delete(self.url).status_code, 404) response = self.client.get( reverse("api_v1:user:admin_users"), {"search": self.user.username} ) self.assertEqual(response.data["count"], 0) replacement = UserFactory.create_user( username=self.user.username, email=self.user.email ) self.assertNotEqual(replacement.pk, self.user_id) def test_deletion_revokes_access_and_refresh_even_without_outstanding_row(self): tokens = UserService.get_tokens_for_user(self.user) self.assertEqual(self.client.delete(self.url).status_code, 204) self.client.force_authenticate(user=None) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {tokens['access']}") self.assertEqual( self.client.get(reverse("api_v1:user:current_user")).status_code, 401 ) self.client.credentials() response = self.client.post( reverse("api_v1:user:token_refresh"), {"refresh": tokens["refresh"]} ) self.assertEqual(response.status_code, 401) def test_inactive_user_cannot_refresh(self): tokens = UserService.get_tokens_for_user(self.user) User.objects.filter(pk=self.user_id).update(is_active=False) self.client.force_authenticate(user=None) response = self.client.post( reverse("api_v1:user:token_refresh"), {"refresh": tokens["refresh"]} ) self.assertEqual(response.status_code, 401) def test_invalid_user_claim_cannot_refresh(self): refresh = RefreshToken.for_user(self.user) refresh["user_id"] = "invalid-user-id" self.client.force_authenticate(user=None) response = self.client.post( reverse("api_v1:user:token_refresh"), {"refresh": str(refresh)} ) self.assertEqual(response.status_code, 401) def test_self_delete_is_rejected_and_account_remains(self): url = reverse("api_v1:user:admin-user-detail", args=[self.admin.pk]) response = self.client.delete(url) self.assertEqual(response.status_code, 400) self.assertEqual(response.data["errors"][0]["code"], "self_delete_forbidden") self.assertTrue( User.objects.filter( pk=self.admin.pk, is_active=True, is_staff=True ).exists() ) def test_another_admin_and_inactive_user_can_be_deleted(self): for attributes in ({"is_staff": True}, {"is_active": False}): user = UserFactory.create_user(**attributes) url = reverse("api_v1:user:admin-user-detail", args=[user.pk]) self.assertEqual(self.client.delete(url).status_code, 204) def test_openapi_describes_delete_without_a_success_body(self): generator = OpenAPISchemaGenerator( info=openapi.Info(title="Account management", default_version="v1"), patterns=[ path( "api/v1/users/admin/users//", AdminUserDetailView.as_view(), ) ], ) schema = generator.get_schema(request=None, public=True) operation = next( item["delete"] for item in schema.paths.values() if "delete" in item ) self.assertEqual( set(operation.responses), {"204", "400", "401", "403", "404", "409"} ) self.assertNotIn("schema", operation.responses["204"]) def test_permission_is_rechecked_for_stale_authenticated_actor(self): User.objects.filter(pk=self.admin.pk).update(is_staff=False) self.assertEqual(self.client.delete(self.url).status_code, 403) self.assertTrue(User.objects.filter(pk=self.user_id).exists()) def test_non_admin_and_anonymous_cannot_delete(self): self.client.force_authenticate(self.user) self.assertEqual(self.client.delete(self.url).status_code, 403) self.client.force_authenticate(user=None) self.assertEqual(self.client.delete(self.url).status_code, 401) def test_transaction_rollback_restores_credentials_and_author_links(self): refresh = RefreshToken.for_user(self.user) job = BackgroundJob.objects.create( task_id="rollback-job", task_name="test", user_id=self.user_id ) with patch.object( User, "delete", side_effect=RuntimeError("test rollback") ), self.assertRaises(RuntimeError): UserService.delete_user(self.user_id, actor_id=self.admin.pk) self.assertTrue(User.objects.filter(pk=self.user_id).exists()) self.assertTrue(Profile.objects.filter(user_id=self.user_id).exists()) self.assertTrue(OutstandingToken.objects.filter(jti=refresh["jti"]).exists()) job.refresh_from_db() self.assertEqual(job.user_id, self.user_id) def test_avatar_cleanup_waits_for_commit_and_preserves_shared_files(self): with TemporaryDirectory(prefix="statecorp-avatar-") as media, override_settings( MEDIA_ROOT=media ): avatar = self.user.profile.avatar avatar.save("test.png", ContentFile(b"avatar")) storage, name = avatar.storage, avatar.name with self.captureOnCommitCallbacks(execute=True): UserService.delete_user(self.user_id, actor_id=self.admin.pk) self.assertTrue(storage.exists(name)) self.assertFalse(storage.exists(name)) owner = UserFactory.create_user() owner.profile.avatar.save("shared.png", ContentFile(b"shared")) name = owner.profile.avatar.name self.admin.profile.avatar = name self.admin.profile.save() with self.captureOnCommitCallbacks(execute=True): UserService.delete_user(owner.pk, actor_id=self.admin.pk) self.assertTrue(storage.exists(name)) def test_avatar_survives_rollback(self): with TemporaryDirectory( prefix="statecorp-avatar-rollback-" ) as media, override_settings(MEDIA_ROOT=media): self.user.profile.avatar.save("test.png", ContentFile(b"avatar")) storage, name = ( self.user.profile.avatar.storage, self.user.profile.avatar.name, ) with self.captureOnCommitCallbacks(execute=True), self.assertRaises( RuntimeError ), transaction.atomic(): UserService.delete_user(self.user_id, actor_id=self.admin.pk) raise RuntimeError("rollback outer transaction") self.assertTrue(storage.exists(name)) self.assertTrue(User.objects.filter(pk=self.user_id).exists()) def test_protected_relation_returns_conflict_and_rolls_back(self): refresh = RefreshToken.for_user(self.user) with patch.object(User, "delete", side_effect=ProtectedError("protected", [])): response = self.client.delete(self.url) self.assertEqual(response.status_code, 409) self.assertEqual(response.data["errors"][0]["code"], "user_delete_conflict") self.assertTrue(User.objects.filter(pk=self.user_id).exists()) self.assertTrue(OutstandingToken.objects.filter(jti=refresh["jti"]).exists()) def test_avatar_storage_failure_does_not_undo_account_deletion(self): self.user.profile.avatar = "avatars/unavailable.png" self.user.profile.save() with patch("apps.user.management.logger.error") as log_error, patch.object( self.user.profile.avatar.storage, "delete", side_effect=OSError("storage unavailable"), ), self.captureOnCommitCallbacks(execute=True): self.assertEqual(self.client.delete(self.url).status_code, 204) self.assertFalse(User.objects.filter(pk=self.user_id).exists()) log_error.assert_called_once()