Coverage for baibu/users/account_deletion_requests.py: 90%

1import logging 

2 

3from django.conf import settings 

4from django.core.mail import EmailMessage 

5from django.db import IntegrityError 

6from django.db import transaction 

7 

8from config.email_recipients import resolve_operations_recipients 

9 

10from .models import AccountDeletionRequest 

11from .models import User 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16def capture_account_deletion_request( # noqa: PLR0913 

17 *, 

18 source: str, 

19 user: User, 

20 requester_name: str = "", 

21 requester_email: str = "", 

22 request_details: str = "", 

23 metadata: dict | None = None, 

24) -> tuple[AccountDeletionRequest, bool]: 

25 """Record a deletion request and deactivate the account at once. 

26 

27 Returns the open request and whether it was newly created. A user has at 

28 most one open request; asking again returns the existing one. 

29 """ 

30 normalized = { 

31 "requester_name": " ".join((requester_name or "").split()), 

32 "requester_email": User.objects.normalize_email_or_none(requester_email) or "", 

33 "request_details": (request_details or "").strip(), 

34 "metadata": dict(metadata or {}), 

35 } 

36 try: 

37 with transaction.atomic(): 

38 user = User.objects.select_for_update().get(pk=user.pk) 

39 if user.is_active: 

40 user.is_active = False 

41 user.save(update_fields=["is_active"]) 

42 existing = ( 

43 AccountDeletionRequest.objects.select_for_update() 

44 .filter(user=user, status__in=AccountDeletionRequest.OPEN_STATUSES) 

45 .order_by("-requested_at") 

46 .first() 

47 ) 

48 if existing is not None: 

49 return existing, False 

50 deletion_request = AccountDeletionRequest.objects.create(source=source, user=user, **normalized) 

51 transaction.on_commit( 

52 lambda request_id=deletion_request.pk: _notify_operations_safely(request_id), 

53 ) 

54 except IntegrityError: 

55 # A concurrent request won the race; return the one that exists. 

56 User.objects.filter(pk=user.pk, is_active=True).update(is_active=False) 

57 existing = AccountDeletionRequest.objects.get(user=user, status__in=AccountDeletionRequest.OPEN_STATUSES) 

58 return existing, False 

59 return deletion_request, True 

60 

61 

62def _notify_operations_safely(request_id) -> None: 

63 try: 

64 _notify_operations(request_id) 

65 except Exception as exc: # noqa: BLE001 

66 logger.error( 

67 "Account deletion notification failed; request_id=%s error_type=%s", 

68 request_id, 

69 type(exc).__name__, 

70 ) 

71 

72 

73def _notify_operations(request_id) -> None: 

74 deletion_request = AccountDeletionRequest.objects.only("id", "source", "requested_at").get(pk=request_id) 

75 recipients = resolve_operations_recipients() 

76 if not recipients: 

77 return 

78 # Personal details stay out of email; staff read them in the admin. 

79 body = "\n".join( 

80 [ 

81 "An account deletion request was received.", 

82 f"Receipt ID: {deletion_request.pk}", 

83 f"Source: {deletion_request.source}", 

84 f"Requested at: {deletion_request.requested_at.isoformat()}", 

85 "", 

86 "Review the request in the admin. Contact details and request text are left out of this email on purpose.", 

87 ], 

88 ) 

89 EmailMessage( 

90 subject=f"{settings.EMAIL_SUBJECT_PREFIX}Account deletion request {deletion_request.pk}", 

91 body=body, 

92 from_email=settings.DEFAULT_FROM_EMAIL, 

93 to=recipients, 

94 ).send(fail_silently=False)