e99616ed6d
All checks were successful
State Corp Backend CI/CD / Quality gate (push) Successful in 2m42s
State Corp Backend CI/CD / Build linux/amd64 images once (push) Successful in 5m29s
State Corp Backend CI/CD / Refresh and release internal main (push) Has been skipped
State Corp Backend CI/CD / Release customer main (push) Has been skipped
State Corp Backend CI/CD / Release dev (push) Successful in 36s
259 lines
11 KiB
Python
259 lines
11 KiB
Python
"""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/<int:user_id>/",
|
|
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()
|