mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2025-06-02 13:48:31 -05:00
Chore/fix: cleanup user or group references in settings upon deletion (#10049)
This commit is contained in:
parent
b06c0a0eba
commit
15b1b83c66
@ -13,6 +13,7 @@ from celery.signals import task_failure
|
||||
from celery.signals import task_postrun
|
||||
from celery.signals import task_prerun
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import Group
|
||||
from django.contrib.auth.models import User
|
||||
from django.db import DatabaseError
|
||||
from django.db import close_old_connections
|
||||
@ -38,6 +39,7 @@ from documents.models import MatchingModel
|
||||
from documents.models import PaperlessTask
|
||||
from documents.models import SavedView
|
||||
from documents.models import Tag
|
||||
from documents.models import UiSettings
|
||||
from documents.models import Workflow
|
||||
from documents.models import WorkflowAction
|
||||
from documents.models import WorkflowRun
|
||||
@ -581,6 +583,51 @@ def cleanup_custom_field_deletion(sender, instance: CustomField, **kwargs):
|
||||
)
|
||||
|
||||
|
||||
@receiver(models.signals.post_delete, sender=User)
|
||||
@receiver(models.signals.post_delete, sender=Group)
|
||||
def cleanup_user_deletion(sender, instance: User | Group, **kwargs):
|
||||
"""
|
||||
When a user or group is deleted, remove non-cascading references.
|
||||
At the moment, just the default permission settings in UiSettings.
|
||||
"""
|
||||
# Remove the user permission settings e.g.
|
||||
# DEFAULT_PERMS_OWNER: 'general-settings:permissions:default-owner',
|
||||
# DEFAULT_PERMS_VIEW_USERS: 'general-settings:permissions:default-view-users',
|
||||
# DEFAULT_PERMS_VIEW_GROUPS: 'general-settings:permissions:default-view-groups',
|
||||
# DEFAULT_PERMS_EDIT_USERS: 'general-settings:permissions:default-edit-users',
|
||||
# DEFAULT_PERMS_EDIT_GROUPS: 'general-settings:permissions:default-edit-groups',
|
||||
for ui_settings in UiSettings.objects.all():
|
||||
try:
|
||||
permissions = ui_settings.settings.get("permissions", {})
|
||||
updated = False
|
||||
if isinstance(instance, User):
|
||||
if permissions.get("default_owner") == instance.pk:
|
||||
permissions["default_owner"] = None
|
||||
updated = True
|
||||
if instance.pk in permissions.get("default_view_users", []):
|
||||
permissions["default_view_users"].remove(instance.pk)
|
||||
updated = True
|
||||
if instance.pk in permissions.get("default_change_users", []):
|
||||
permissions["default_change_users"].remove(instance.pk)
|
||||
updated = True
|
||||
elif isinstance(instance, Group):
|
||||
if instance.pk in permissions.get("default_view_groups", []):
|
||||
permissions["default_view_groups"].remove(instance.pk)
|
||||
updated = True
|
||||
if instance.pk in permissions.get("default_change_groups", []):
|
||||
permissions["default_change_groups"].remove(instance.pk)
|
||||
updated = True
|
||||
if updated:
|
||||
ui_settings.settings["permissions"] = permissions
|
||||
ui_settings.save(update_fields=["settings"])
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error while cleaning up user {instance.pk} ({instance.username}) from ui_settings: {e}"
|
||||
if isinstance(instance, User)
|
||||
else f"Error while cleaning up group {instance.pk} ({instance.name}) from ui_settings: {e}",
|
||||
)
|
||||
|
||||
|
||||
def add_to_index(sender, document, **kwargs):
|
||||
from documents import index
|
||||
|
||||
|
@ -6,6 +6,7 @@ from django.http import HttpRequest
|
||||
from django.test import TestCase
|
||||
from django.test import override_settings
|
||||
|
||||
from documents.models import UiSettings
|
||||
from paperless.signals import handle_failed_login
|
||||
from paperless.signals import handle_social_account_updated
|
||||
|
||||
@ -190,3 +191,76 @@ class TestSyncSocialLoginGroups(TestCase):
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
self.assertEqual(list(user.groups.all()), [])
|
||||
|
||||
|
||||
class TestUserGroupDeletionCleanup(TestCase):
|
||||
"""
|
||||
Test that when a user or group is deleted, references are cleaned up properly
|
||||
from ui_settings
|
||||
"""
|
||||
|
||||
def test_user_group_deletion_cleanup(self):
|
||||
"""
|
||||
GIVEN:
|
||||
- Existing user
|
||||
- Existing group
|
||||
WHEN:
|
||||
- The user is deleted
|
||||
- The group is deleted
|
||||
THEN:
|
||||
- References in ui_settings are cleaned up
|
||||
"""
|
||||
user = User.objects.create_user(username="testuser")
|
||||
user2 = User.objects.create_user(username="testuser2")
|
||||
group = Group.objects.create(name="testgroup")
|
||||
|
||||
ui_settings = UiSettings.objects.create(
|
||||
user=user,
|
||||
settings={
|
||||
"permissions": {
|
||||
"default_owner": user2.id,
|
||||
"default_view_users": [user2.id],
|
||||
"default_change_users": [user2.id],
|
||||
"default_view_groups": [group.id],
|
||||
"default_change_groups": [group.id],
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
user2.delete()
|
||||
ui_settings.refresh_from_db()
|
||||
permissions = ui_settings.settings.get("permissions", {})
|
||||
self.assertIsNone(permissions.get("default_owner"))
|
||||
self.assertEqual(permissions.get("default_view_users"), [])
|
||||
self.assertEqual(permissions.get("default_change_users"), [])
|
||||
|
||||
group.delete()
|
||||
ui_settings.refresh_from_db()
|
||||
permissions = ui_settings.settings.get("permissions", {})
|
||||
self.assertEqual(permissions.get("default_view_groups"), [])
|
||||
self.assertEqual(permissions.get("default_change_groups"), [])
|
||||
|
||||
def test_user_group_deletion_error_handling(self):
|
||||
"""
|
||||
GIVEN:
|
||||
- Existing user and group
|
||||
WHEN:
|
||||
- The user is deleted and an error occurs during the signal handling
|
||||
THEN:
|
||||
- Error is logged and the system remains stable
|
||||
"""
|
||||
user = User.objects.create_user(username="testuser")
|
||||
user2 = User.objects.create_user(username="testuser2")
|
||||
user2_id = user2.id
|
||||
Group.objects.create(name="testgroup")
|
||||
|
||||
UiSettings.objects.create(
|
||||
user=user,
|
||||
) # invalid, no settings, this probably should not happen in production
|
||||
|
||||
with self.assertLogs("paperless.handlers", level="ERROR") as cm:
|
||||
user2.delete()
|
||||
self.assertIn(
|
||||
f"Error while cleaning up user {user2_id}",
|
||||
cm.output[0],
|
||||
)
|
||||
|
Loading…
x
Reference in New Issue
Block a user