From debb253fa589df5c7390198bf29328e1ca479ed8 Mon Sep 17 00:00:00 2001 From: Manuel Raynaud Date: Fri, 3 Jul 2026 17:10:04 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8(backend)=20add=20management=20command?= =?UTF-8?q?=20to=20reset=20a=20Document?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We need a management command to reset a Document to an initial state and deletes everything related to it. This command can be usefull to reset a demo for example. --- CHANGELOG.md | 1 + .../management/commands/clean_document.py | 178 ++++++++ .../tests/commands/test_clean_document.py | 380 ++++++++++++++++++ 3 files changed, 559 insertions(+) create mode 100644 src/backend/core/management/commands/clean_document.py create mode 100644 src/backend/core/tests/commands/test_clean_document.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 810ddc02c..3e0046f64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to - ✨(y-provider) preserve callouts, PDFs, page breaks, interlinking links and commented text on HTML/markdown export #2296 - ✨(frontend) add a user menu #2463 +- ✨(backend) add management command to reset a Document #1882 ### Changed diff --git a/src/backend/core/management/commands/clean_document.py b/src/backend/core/management/commands/clean_document.py new file mode 100644 index 000000000..e7a006ad5 --- /dev/null +++ b/src/backend/core/management/commands/clean_document.py @@ -0,0 +1,178 @@ +"""Clean a document by resetting it (keeping its title) and deleting all descendants.""" + +import logging + +from django.conf import settings +from django.core.files.storage import default_storage +from django.core.management.base import BaseCommand, CommandError +from django.db import transaction +from django.db.models import Q + +from botocore.exceptions import ClientError + +from core.choices import LinkReachChoices, LinkRoleChoices, RoleChoices +from core.models import ( + Document, + DocumentAccess, + DocumentAskForAccess, + DocumentFavorite, + Invitation, + LinkTrace, + Thread, +) + +logger = logging.getLogger("impress.commands.clean_document") + + +class Command(BaseCommand): + """Reset a document (keeping its title) and delete all its descendants.""" + + help = __doc__ + + def add_arguments(self, parser): + """Define command arguments.""" + parser.add_argument( + "document_id", + type=str, + help="UUID of the document to clean", + ) + parser.add_argument( + "-f", + "--force", + action="store_true", + default=False, + help="Force command execution despite DEBUG is set to False", + ) + parser.add_argument( + "-t", + "--title", + type=str, + default=None, + help="Update the document title to this value", + ) + parser.add_argument( + "--link_reach", + type=str, + default=LinkReachChoices.RESTRICTED, + choices=LinkReachChoices, + help="Update the link_reach to this value", + ) + parser.add_argument( + "--link_role", + type=str, + default=LinkRoleChoices.READER, + choices=LinkRoleChoices, + help="update the link_role to this value", + ) + + def handle(self, *args, **options): + """Execute the clean_document command.""" + if not settings.DEBUG and not options["force"]: + raise CommandError( + "This command is not meant to be used in production environment " + "except you know what you are doing, if so use --force parameter" + ) + + document_id = options["document_id"] + + try: + document = Document.objects.get(pk=document_id) + except (Document.DoesNotExist, ValueError) as err: + raise CommandError(f"Document {document_id} does not exist.") from err + + descendants = list(document.get_descendants()) + descendant_ids = [doc.id for doc in descendants] + all_documents = [document, *descendants] + + # Collect all attachment keys before the transaction clears them + all_attachment_keys = [] + for doc in all_documents: + all_attachment_keys.extend(doc.attachments) + + self.stdout.write( + f"Cleaning document {document_id} and deleting " + f"{len(descendants)} descendant(s)..." + ) + + with transaction.atomic(): + self._clean_root_relations(document) + + # Reset root document fields. We use save() rather than a queryset + # update() so the post_save signal fires (search re-indexation) and + # `updated_at` is refreshed. All descendants are about to be deleted, + # so the document can no longer have any deleted child either. + document.excerpt = None + document.link_reach = options["link_reach"] + document.link_role = options["link_role"] + document.attachments = [] + document.has_deleted_children = False + if options["title"] is not None: + document.title = options["title"] + document.save() + + if options["title"] is not None: + self.stdout.write( + f'Reset fields on root document (title set to "{options["title"]}").' + ) + else: + self.stdout.write("Reset fields on root document (title kept).") + + # Delete all descendants (cascades accesses and invitations) + if descendants: + deleted_count, _ = Document.objects.filter( + id__in=descendant_ids + ).delete() + self.stdout.write(f"Deleted {deleted_count} descendant(s).") + + # Delete S3 content outside the transaction (S3 is not transactional) + bucket = default_storage.bucket + + for doc in all_documents: + try: + self._purge_object(bucket, doc.file_key) + except ClientError: + logger.warning("Failed to delete S3 file for document %s", doc.id) + + self.stdout.write(f"Deleted S3 content for {len(all_documents)} document(s).") + + for key in all_attachment_keys: + try: + self._purge_object(bucket, key) + except ClientError: + logger.warning("Failed to delete S3 attachment %s", key) + + self.stdout.write(f"Deleted {len(all_attachment_keys)} attachment(s) from S3.") + self.stdout.write("Done.") + + def _clean_root_relations(self, document): + """ + Delete the relations attached to the root document: accesses (except + owners), invitations, threads, favorites, link traces and pending + access requests. + """ + access_count, _ = DocumentAccess.objects.filter( + Q(document_id=document.id) & ~Q(role=RoleChoices.OWNER) + ).delete() + self.stdout.write(f"Deleted {access_count} access(es) on root document.") + + for model, label in ( + (Invitation, "invitation"), + (Thread, "thread"), + (DocumentFavorite, "favorite"), + (LinkTrace, "link trace"), + (DocumentAskForAccess, "pending access request"), + ): + count, _ = model.objects.filter(document_id=document.id).delete() + self.stdout.write(f"Deleted {count} {label}(s) on root document.") + + @staticmethod + def _purge_object(bucket, key): + """ + Permanently delete every version and delete-marker of a single object. + + The media bucket is versioned, so a plain ``delete`` would only add a + delete-marker and leave the content retrievable. Filtering the bucket's + ``object_versions`` collection on the key and deleting it removes all + stored versions at once. + """ + bucket.object_versions.filter(Prefix=key).delete() diff --git a/src/backend/core/tests/commands/test_clean_document.py b/src/backend/core/tests/commands/test_clean_document.py new file mode 100644 index 000000000..2f8468416 --- /dev/null +++ b/src/backend/core/tests/commands/test_clean_document.py @@ -0,0 +1,380 @@ +"""Unit tests for the `clean_document` management command.""" + +import random +from unittest import mock +from uuid import uuid4 + +from django.core.management import CommandError, call_command + +import pytest +from botocore.exceptions import ClientError + +from core import choices, factories, models +from core.choices import LinkReachChoices, LinkRoleChoices + +pytestmark = pytest.mark.django_db + + +def purged_keys(mock_storage): + """ + Return the set of object keys whose versions were purged from S3, i.e. the + keys passed to ``bucket.object_versions.filter(Prefix=...)``. + """ + return { + call.kwargs["Prefix"] + for call in mock_storage.bucket.object_versions.filter.call_args_list + } + + +def test_clean_document_with_descendants(settings): + """The command should reset the root (keeping title) and delete descendants.""" + settings.DEBUG = True + + # Create a root document with subdocuments + root = factories.DocumentFactory( + title="Root", + link_reach=LinkReachChoices.PUBLIC, + link_role=LinkRoleChoices.EDITOR, + ) + child = factories.DocumentFactory( + parent=root, + title="Child", + link_reach=LinkReachChoices.AUTHENTICATED, + link_role=LinkRoleChoices.EDITOR, + ) + grandchild = factories.DocumentFactory( + parent=child, + title="Grandchild", + ) + + # Create accesses and invitations + factories.UserDocumentAccessFactory.create_batch( + 5, + document=root, + role=random.choice( + [ + role + for role in choices.RoleChoices + if role not in choices.PRIVILEGED_ROLES + ], + ), + ) + # One owner role + factories.UserDocumentAccessFactory(document=root, role=choices.RoleChoices.OWNER) + factories.UserDocumentAccessFactory(document=child) + factories.InvitationFactory(document=root) + factories.InvitationFactory(document=child) + factories.ThreadFactory.create_batch(5, document=root) + + assert models.Invitation.objects.filter(document=root).exists() + assert models.Thread.objects.filter(document=root).exists() + assert models.DocumentAccess.objects.filter(document=root).exists() + + with mock.patch( + "core.management.commands.clean_document.default_storage" + ) as mock_storage: + call_command("clean_document", str(root.id), "--force") + + # Root document should still exist with title kept and other fields reset + root.refresh_from_db() + assert root.title == "Root" + assert root.excerpt is None + assert root.link_reach == LinkReachChoices.RESTRICTED + assert root.link_role == LinkRoleChoices.READER + assert root.attachments == [] + + # Accesses and invitations on root should be deleted. Only owner should be kept + keeping_accesses = list(models.DocumentAccess.objects.filter(document=root)) + assert len(keeping_accesses) == 1 + assert keeping_accesses[0].role == models.RoleChoices.OWNER + assert not models.Invitation.objects.filter(document=root).exists() + assert not models.Thread.objects.filter(document=root).exists() + + # Descendants should be deleted entirely + assert not models.Document.objects.filter(id__in=[child.id, grandchild.id]).exists() + + # Root should have no descendants + root.refresh_from_db() + assert root.get_descendants().count() == 0 + + # Every version of the 3 document files should have been purged from S3 + assert purged_keys(mock_storage) == { + root.file_key, + child.file_key, + grandchild.file_key, + } + + +def test_clean_document_invalid_uuid(settings): + """The command should raise an error for a non-existent document.""" + settings.DEBUG = True + + fake_id = str(uuid4()) + with pytest.raises(CommandError, match=f"Document {fake_id} does not exist."): + call_command("clean_document", fake_id, "--force") + + +def test_clean_document_no_force_in_production(settings): + """The command should require --force when DEBUG is False.""" + settings.DEBUG = False + + doc = factories.DocumentFactory() + with pytest.raises(CommandError, match="not meant to be used in production"): + call_command("clean_document", str(doc.id)) + + +def test_clean_document_resets_has_deleted_children(settings): + """ + Cleaning a document with a soft-deleted child must leave the root as a + leaf again: `has_deleted_children` reset and `numchild` back to zero. + """ + settings.DEBUG = True + + root = factories.DocumentFactory(title="Root") + child = factories.DocumentFactory(parent=root) + child.soft_delete() + + root.refresh_from_db() + assert root.has_deleted_children is True + assert root.is_leaf() is False + + with mock.patch("core.management.commands.clean_document.default_storage"): + call_command("clean_document", str(root.id), "--force") + + assert not models.Document.objects.filter(id=child.id).exists() + + root.refresh_from_db() + assert root.has_deleted_children is False + assert root.numchild == 0 + assert root.is_leaf() is True + + +def test_clean_document_clears_root_relations(settings): + """ + Cleaning a document should also remove its favorites, link traces and + pending access requests, not only accesses/invitations/threads. + """ + settings.DEBUG = True + + users = factories.UserFactory.create_batch(3) + root = factories.DocumentFactory( + favorited_by=users, + link_traces=users, + ) + factories.DocumentAskForAccessFactory.create_batch(2, document=root) + + assert models.DocumentFavorite.objects.filter(document=root).exists() + assert models.LinkTrace.objects.filter(document=root).exists() + assert models.DocumentAskForAccess.objects.filter(document=root).exists() + + with mock.patch("core.management.commands.clean_document.default_storage"): + call_command("clean_document", str(root.id), "--force") + + assert not models.DocumentFavorite.objects.filter(document=root).exists() + assert not models.LinkTrace.objects.filter(document=root).exists() + assert not models.DocumentAskForAccess.objects.filter(document=root).exists() + + +def test_clean_document_single_document(settings): + """The command should work on a single document without children.""" + settings.DEBUG = True + + doc = factories.DocumentFactory( + title="Single", + link_reach=LinkReachChoices.PUBLIC, + link_role=LinkRoleChoices.EDITOR, + ) + factories.UserDocumentAccessFactory.create_batch( + 5, + document=doc, + role=random.choice( + [ + role + for role in choices.RoleChoices + if role not in choices.PRIVILEGED_ROLES + ], + ), + ) + # One owner role + factories.UserDocumentAccessFactory(document=doc, role=choices.RoleChoices.OWNER) + factories.ThreadFactory.create_batch(5, document=doc) + factories.InvitationFactory(document=doc) + + with mock.patch( + "core.management.commands.clean_document.default_storage" + ) as mock_storage: + call_command("clean_document", str(doc.id), "--force") + + # Accesses and invitations on root should be deleted. Only owner should be kept + keeping_accesses = list(models.DocumentAccess.objects.filter(document=doc)) + assert len(keeping_accesses) == 1 + assert keeping_accesses[0].role == models.RoleChoices.OWNER + assert not models.Invitation.objects.filter(document=doc).exists() + assert not models.Thread.objects.filter(document=doc).exists() + + doc.refresh_from_db() + assert doc.title == "Single" + assert doc.excerpt is None + assert doc.link_reach == LinkReachChoices.RESTRICTED + assert doc.link_role == LinkRoleChoices.READER + assert doc.attachments == [] + + assert purged_keys(mock_storage) == {doc.file_key} + + +def test_clean_document_with_title_option(settings): + """The --title option should update the document title.""" + settings.DEBUG = True + + doc = factories.DocumentFactory( + title="Old Title", + link_reach=LinkReachChoices.PUBLIC, + link_role=LinkRoleChoices.EDITOR, + ) + + with mock.patch("core.management.commands.clean_document.default_storage"): + call_command("clean_document", str(doc.id), "--force", "--title", "New Title") + + doc.refresh_from_db() + assert doc.title == "New Title" + assert doc.excerpt is None + assert doc.link_reach == LinkReachChoices.RESTRICTED + assert doc.link_role == LinkRoleChoices.READER + assert doc.attachments == [] + + +def test_clean_document_deletes_attachments_from_s3(settings): + """The command should delete attachment files from S3.""" + settings.DEBUG = True + + root = factories.DocumentFactory( + attachments=["root-id/attachments/file1.png", "root-id/attachments/file2.pdf"], + ) + child = factories.DocumentFactory( + parent=root, + attachments=["child-id/attachments/file3.png"], + ) + + with mock.patch( + "core.management.commands.clean_document.default_storage" + ) as mock_storage: + call_command("clean_document", str(root.id), "--force") + + # Every document file and every attachment key should have been purged + # (all versions of each key are removed via object_versions.filter().delete()) + assert purged_keys(mock_storage) == { + root.file_key, + child.file_key, + "root-id/attachments/file1.png", + "root-id/attachments/file2.pdf", + "child-id/attachments/file3.png", + } + + +def test_clean_document_s3_errors_do_not_stop_command(settings): + """S3 deletion errors should be logged but not stop the command.""" + settings.DEBUG = True + + doc = factories.DocumentFactory( + attachments=["doc-id/attachments/file1.png"], + ) + + with mock.patch( + "core.management.commands.clean_document.default_storage" + ) as mock_storage: + mock_storage.bucket.object_versions.filter.return_value.delete.side_effect = ( + ClientError( + {"Error": {"Code": "500", "Message": "Internal Error"}}, + "DeleteObjects", + ) + ) + # Command should complete without raising + call_command("clean_document", str(doc.id), "--force") + + +def test_clean_document_with_options(settings): + """Run the command using optional argument link_reach and link_role.""" + + settings.DEBUG = True + + # Create a root document with subdocuments + root = factories.DocumentFactory( + title="Root", + link_reach=LinkReachChoices.PUBLIC, + link_role=LinkRoleChoices.READER, + ) + child = factories.DocumentFactory( + parent=root, + title="Child", + link_reach=LinkReachChoices.AUTHENTICATED, + link_role=LinkRoleChoices.EDITOR, + ) + grandchild = factories.DocumentFactory( + parent=child, + title="Grandchild", + ) + + # Create accesses and invitations + factories.UserDocumentAccessFactory.create_batch( + 5, + document=root, + role=random.choice( + [ + role + for role in choices.RoleChoices + if role not in choices.PRIVILEGED_ROLES + ], + ), + ) + # One owner role + factories.UserDocumentAccessFactory(document=root, role=choices.RoleChoices.OWNER) + factories.UserDocumentAccessFactory(document=child) + factories.InvitationFactory(document=root) + factories.InvitationFactory(document=child) + factories.ThreadFactory.create_batch(5, document=root) + + assert models.Invitation.objects.filter(document=root).exists() + assert models.Thread.objects.filter(document=root).exists() + assert models.DocumentAccess.objects.filter(document=root).exists() + + with mock.patch( + "core.management.commands.clean_document.default_storage" + ) as mock_storage: + call_command( + "clean_document", + str(root.id), + "--force", + "--link_reach", + "public", + "--link_role", + "editor", + ) + + # Root document should still exist with title kept and other fields reset + root.refresh_from_db() + assert root.title == "Root" + assert root.excerpt is None + assert root.link_reach == LinkReachChoices.PUBLIC + assert root.link_role == LinkRoleChoices.EDITOR + assert root.attachments == [] + + # Accesses and invitations on root should be deleted. Only owner should be kept + keeping_accesses = list(models.DocumentAccess.objects.filter(document=root)) + assert len(keeping_accesses) == 1 + assert keeping_accesses[0].role == models.RoleChoices.OWNER + assert not models.Invitation.objects.filter(document=root).exists() + assert not models.Thread.objects.filter(document=root).exists() + + # Descendants should be deleted entirely + assert not models.Document.objects.filter(id__in=[child.id, grandchild.id]).exists() + + # Root should have no descendants + root.refresh_from_db() + assert root.get_descendants().count() == 0 + + # Every version of the 3 document files should have been purged from S3 + assert purged_keys(mock_storage) == { + root.file_key, + child.file_key, + grandchild.file_key, + }