Commit 59b9c8ab authored by Mickaël Desfrênes's avatar Mickaël Desfrênes
Browse files

Reject nonrecursive deletion of collections with active resources 

Collection deletion checked only for active child collections, allowing
nonrecursive requests to archive populated leaf collections while their
resources remained active and linked to the archived collection.

Reject nonrecursive deletion when any active resource remains, returning
"collection has resources" before changing collection or resource state.
Ignore deleted content when determining emptiness and preserve recursive
handling of shared and unshared resources.

Use exists() for both content checks so detecting children or resources
does not load their complete result sets. Update the docstring and comments
to explain active-content requirements and recursive shared-resource behavior.

Add seven regression tests covering ordinary resources and files without
resource deletion permission, shared resources, active children, empty
collections, archived content, and recursive resource deletion. Verify that
rejected requests preserve memberships, timestamps, and queued tasks.

Validation: Ruff and diff whitespace checks passed. All 234 resource, RPC,
and annotation tests passed using an isolated in-memory SQLite database.
parent 21e9147b

src/rpc/methods.py

+11 −5
Changes for src/rpc/methods.py: 11 added lines, 5 removed lines.
Original line number Diff line number Diff line
@@ -825,9 +825,9 @@ def delete_collection(user: User, collection_id: int, recursive: bool = False) - 
    """

    Delete collection given its id.



    Collection MUST be empty of any content (no children collections and no resources).

    unless the 'recursive' parameter is set to True, in which case ALL descendants will be

    deleted.

    Nonrecursive deletion requires no active child collections or resources.

    Recursive deletion also processes descendants according to their permissions

    and preserves resources shared with other active collections.

    """

    query_set: Union[Iterator[Collection], QuerySet] = Collection.objects.filter(

        pk=collection_id, deleted_at__isnull=True
@@ -860,9 +860,15 @@ def delete_collection(user: User, collection_id: int, recursive: bool = False) - 
            if acl.can_delete(descendant_collection_instance):

                descendant_collection_instance.soft_delete()



    # do not allow to delete collection with content

    if len(collection_instance.children()) > 0:

    if collection_instance.children().exists():

        return {"success": False, "status": "collection has descendants"}

    # Recursive deletion can leave memberships to shared resources behind.

    # For nonrecursive deletion, active resources make the collection nonempty.

    if (

        not recursive

        and collection_instance.resources.filter(deleted_at__isnull=True).exists()

    ):

        return {"success": False, "status": "collection has resources"}

    collection_instance.soft_delete()

    return {"success": True, "status": "all good"}

src/rpc/tests.py

+168 −0
Changes for src/rpc/tests.py: 168 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -2518,6 +2518,174 @@ class ServiceTestCase(TestCase): 
        self.test_project_root_collection.refresh_from_db()

        self.assertIsNone(self.test_project_root_collection.deleted_at)



    def test_delete_collection_nonrecursive_rejects_active_resources(self):

        resources = [

            models.Resource.objects.create(

                ptr_project=self.test_project, title="Active resource"

            ),

            self._public_download_test_file(),

        ]

        for resource in resources:

            with self.subTest(resource_type=resource._meta.model_name):

                collection = models.Collection.objects.create(

                    project=self.test_project,

                    parent=self.test_project_root_collection,

                    title=f"Contains {resource._meta.model_name}",

                )

                collection.resources.add(resource)

                set_permission_on_role(self.admin_role, resource, can_read=True)

                queued_tasks = DBTaskResult.objects.count()

                original_updated_at = resource.updated_at



                result = methods.delete_collection(self.test_user, collection.pk)



                self.assertEqual(

                    result, {"success": False, "status": "collection has resources"}

                )

                collection.refresh_from_db()

                resource.refresh_from_db()

                self.assertIsNone(collection.deleted_at)

                self.assertIsNone(resource.deleted_at)

                self.assertEqual(resource.updated_at, original_updated_at)

                self.assertTrue(collection.resources.filter(pk=resource.pk).exists())

                self.assertEqual(DBTaskResult.objects.count(), queued_tasks)



    def test_delete_collection_nonrecursive_rejects_shared_resources(self):

        collection = models.Collection.objects.create(

            project=self.test_project,

            parent=self.test_project_root_collection,

            title="Shared resource collection",

        )

        resource = models.Resource.objects.create(

            ptr_project=self.test_project, title="Shared resource"

        )

        collection.resources.add(resource)

        self.test_project_root_collection.resources.add(resource)



        result = methods.delete_collection(self.test_user, collection.pk)



        self.assertEqual(

            result, {"success": False, "status": "collection has resources"}

        )

        collection.refresh_from_db()

        resource.refresh_from_db()

        self.assertIsNone(collection.deleted_at)

        self.assertIsNone(resource.deleted_at)

        self.assertEqual(resource.collections.count(), 2)



    def test_delete_collection_nonrecursive_rejects_active_children(self):

        collection = models.Collection.objects.create(

            project=self.test_project,

            parent=self.test_project_root_collection,

            title="Parent collection",

        )

        child = models.Collection.objects.create(

            project=self.test_project, parent=collection, title="Active child"

        )

        queued_tasks = DBTaskResult.objects.count()



        result = methods.delete_collection(self.test_user, collection.pk)



        self.assertEqual(

            result, {"success": False, "status": "collection has descendants"}

        )

        collection.refresh_from_db()

        child.refresh_from_db()

        self.assertIsNone(collection.deleted_at)

        self.assertIsNone(child.deleted_at)

        self.assertEqual(DBTaskResult.objects.count(), queued_tasks)



    def test_delete_collection_nonrecursive_allows_empty_collection(self):

        collection = models.Collection.objects.create(

            project=self.test_project,

            parent=self.test_project_root_collection,

            title="Empty collection",

        )



        result = methods.delete_collection(self.test_user, collection.pk)



        self.assertEqual(result, {"success": True, "status": "all good"})

        collection.refresh_from_db()

        self.assertIsNotNone(collection.deleted_at)



    def test_delete_collection_nonrecursive_ignores_deleted_content(self):

        collection = models.Collection.objects.create(

            project=self.test_project,

            parent=self.test_project_root_collection,

            title="Archived content collection",

        )

        child = models.Collection.objects.create(

            project=self.test_project,

            parent=collection,

            title="Deleted child",

            deleted_at=timezone.now(),

        )

        resource = models.Resource.objects.create(

            ptr_project=self.test_project,

            title="Deleted resource",

            deleted_at=timezone.now(),

        )

        collection.resources.add(resource)

        original_child_deleted_at = child.deleted_at

        original_resource_deleted_at = resource.deleted_at



        result = methods.delete_collection(self.test_user, collection.pk)



        self.assertEqual(result, {"success": True, "status": "all good"})

        collection.refresh_from_db()

        child.refresh_from_db()

        resource.refresh_from_db()

        self.assertIsNotNone(collection.deleted_at)

        self.assertEqual(child.deleted_at, original_child_deleted_at)

        self.assertEqual(resource.deleted_at, original_resource_deleted_at)

        self.assertTrue(collection.resources.filter(pk=resource.pk).exists())



    def test_delete_collection_recursive_preserves_shared_resources(self):

        collection = models.Collection.objects.create(

            project=self.test_project,

            parent=self.test_project_root_collection,

            title="Recursive shared resource collection",

        )

        resource = models.Resource.objects.create(

            ptr_project=self.test_project, title="Shared recursive resource"

        )

        collection.resources.add(resource)

        self.test_project_root_collection.resources.add(resource)



        result = methods.delete_collection(

            self.test_user, collection.pk, recursive=True

        )



        self.assertEqual(result, {"success": True, "status": "all good"})

        collection.refresh_from_db()

        resource.refresh_from_db()

        self.assertIsNotNone(collection.deleted_at)

        self.assertIsNone(resource.deleted_at)

        self.assertTrue(

            self.test_project_root_collection.resources.filter(pk=resource.pk).exists()

        )



    def test_delete_collection_recursive_deletes_unshared_resources(self):

        collection = models.Collection.objects.create(

            project=self.test_project,

            parent=self.test_project_root_collection,

            title="Recursive unshared resource collection",

        )

        resource = models.Resource.objects.create(

            ptr_project=self.test_project, title="Unshared recursive resource"

        )

        collection.resources.add(resource)



        result = methods.delete_collection(

            self.test_user, collection.pk, recursive=True

        )



        self.assertEqual(result, {"success": True, "status": "all good"})

        collection.refresh_from_db()

        resource.refresh_from_db()

        self.assertIsNotNone(collection.deleted_at)

        self.assertIsNotNone(resource.deleted_at)



    def test_move_collection_rejects_project_root(self):

        destination = models.Collection.objects.create(

            title="root move destination",