Match metadata field and value during recursive removal (21e9147b) · Commits · certic / Jama

src/rpc/methods.py

+66 −11

Changes for src/rpc/methods.py: 66 added lines, 11 removed lines.
Original line number	Diff line number	Diff line
		@@ -1819,32 +1819,87 @@ def add_meta_to_resource(


		@_rpc_groups(["Metadatas", "Collections"])
		@transaction.atomic
		def remove_meta_value_from_collection(
		user: User, collection_id: int, meta_value_id: int, recursive: bool = False
		) -> bool:
		"""
		Remove a meta value from a collection given their ids.

		Recursive removal matches the metadata field and exact value on active
		descendant collections and resources, including direct resources.
		"""
		query_set: Union[Iterator[Collection], QuerySet] = Collection.objects.filter(
		pk=collection_id, deleted_at__isnull=True
		)
		).select_related("project")
		collection_instance = query_set.first()
		if not collection_instance:
		raise ServiceException(NO_SUCH_COLLECTION)
		UserAccess(user, collection_instance.project).check_update(collection_instance)
		for meta_value in collection_instance.metadatacollectionvalue_set.all():
		if meta_value.id == meta_value_id:
		acl = UserAccess(user, collection_instance.project)
		acl.check_update(collection_instance)
		# Resolve the selected row on this collection first. A missing row or an
		# ID belonging to another collection must not trigger descendant deletion.
		meta_value = (
		collection_instance.metadatacollectionvalue_set.filter(pk=meta_value_id)
		.select_related("metadata__set")
		.first()
		)
		if meta_value is None:
		return True
		# Every recursive match uses this same metadata field, so its EXIF guard
		# protects all matches. The metadata set was loaded with the selected row.
		_prevent_exiftool_metadata_modification(meta_value)
		if not recursive:
		meta_value.delete()
		if recursive:
		for descendant_collection_instance in collection_instance.descendants():
		remove_meta_value_from_collection(
		user, descendant_collection_instance.pk, meta_value_id
		return True

		# Fetch one tree level per query instead of querying children individually.
		# Deleted collections also exclude their descendants from this traversal.
		collections_by_id = {collection_instance.pk: collection_instance}
		parent_ids = [collection_instance.pk]
		while parent_ids:
		children = list(
		Collection.objects.filter(
		parent_id__in=parent_ids, deleted_at__isnull=True
		).exclude(pk__in=collections_by_id)
		)
		for descendant_resource_instance in collection_instance.descendants_resources():
		remove_meta_value_from_resource(
		user, descendant_resource_instance.pk, meta_value_id
		for child in children:
		# Reuse hydrated parents so inherited ACL checks avoid lazy queries.
		child.parent = collections_by_id[child.parent_id]
		acl.check_update(child)
		collections_by_id[child.pk] = child
		parent_ids = [child.pk for child in children]

		# Include direct resources of the source collection. Shared resources can
		# appear in several subtree collections; DISTINCT checks each only once.
		resources = list(
		Resource.objects.filter(
		collections__in=collections_by_id, deleted_at__isnull=True
		).distinct()
		)
		# Memberships outside this subtree can also affect inherited permissions.
		# Prefetch ancestors of all active memberships, including those outside it.
		acl.prefetch_resource_ancestors(resources)
		for resource in resources:
		acl.check_update(resource)

		# All collection/resource permissions are checked before either deletion.
		# The transaction also rolls back collection deletions if resource deletion
		# fails, including when this function is called directly outside the RPC view.
		#
		# Value-row IDs are unrelated across objects and metadata value tables.
		# Match both the field and exact text to preserve other values of that field
		# and values with the same text in different fields.
		MetadataCollectionValue.objects.filter(
		collection_id__in=collections_by_id,
		metadata_id=meta_value.metadata_id,
		value=meta_value.value,
		).delete()
		MetadataResourceValue.objects.filter(
		resource_id__in=[resource.pk for resource in resources],
		metadata_id=meta_value.metadata_id,
		value=meta_value.value,
		).delete()
		return True

src/rpc/tests.py

+305 −0

Changes for src/rpc/tests.py: 305 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -40,6 +40,311 @@ object_classes = [
		crud = ["create", "read", "update", "delete"]


		class RecursiveMetadataRemovalTestCase(TestCase):
		@classmethod
		def setUpTestData(cls):
		cls.user = User.objects.create(username="metadata_removal_user")
		cls.project = models.Project.objects.create(label="metadata removal")
		cls.root = models.Collection.objects.create(project=cls.project, title="Root")
		cls.collection = models.Collection.objects.create(
		project=cls.project, parent=cls.root, title="Source"
		)
		cls.child = models.Collection.objects.create(
		project=cls.project, parent=cls.collection, title="Child"
		)
		cls.grandchild = models.Collection.objects.create(
		project=cls.project, parent=cls.child, title="Grandchild"
		)
		cls.outside_collection = models.Collection.objects.create(
		project=cls.project, parent=cls.root, title="Outside"
		)
		cls.direct_resource = models.Resource.objects.create(
		ptr_project=cls.project, title="Direct resource"
		)
		cls.shared_resource = models.Resource.objects.create(
		ptr_project=cls.project, title="Shared resource"
		)
		cls.outside_resource = models.Resource.objects.create(
		ptr_project=cls.project, title="Outside resource"
		)
		cls.collection.resources.add(cls.direct_resource)
		cls.child.resources.add(cls.shared_resource)
		cls.grandchild.resources.add(cls.shared_resource)
		cls.outside_collection.resources.add(cls.outside_resource)
		cls.metadata_set = models.MetadataSet.objects.create(
		project=cls.project, title="Description"
		)
		cls.author = models.Metadata.objects.create(
		project=cls.project, set=cls.metadata_set, title="Author"
		)
		cls.copyright = models.Metadata.objects.create(
		project=cls.project, set=cls.metadata_set, title="Copyright"
		)
		cls.source_value = models.MetadataCollectionValue.objects.create(
		pk=42, collection=cls.collection, metadata=cls.author, value="Alice"
		)
		cls.child_value = models.MetadataCollectionValue.objects.create(
		collection=cls.child, metadata=cls.author, value="Alice"
		)
		cls.grandchild_value = models.MetadataCollectionValue.objects.create(
		collection=cls.grandchild, metadata=cls.author, value="Alice"
		)
		cls.other_collection_value = models.MetadataCollectionValue.objects.create(
		collection=cls.outside_collection, metadata=cls.author, value="Alice"
		)
		models.MetadataCollectionValue.objects.create(
		collection=cls.child, metadata=cls.author, value="Bob"
		)
		models.MetadataCollectionValue.objects.create(
		collection=cls.child, metadata=cls.copyright, value="Alice"
		)
		cls.colliding_resource_value = models.MetadataResourceValue.objects.create(
		pk=42, resource=cls.shared_resource, metadata=cls.copyright, value="Museum"
		)
		cls.direct_value = models.MetadataResourceValue.objects.create(
		resource=cls.direct_resource, metadata=cls.author, value="Alice"
		)
		cls.shared_value = models.MetadataResourceValue.objects.create(
		resource=cls.shared_resource, metadata=cls.author, value="Alice"
		)
		models.MetadataResourceValue.objects.create(
		resource=cls.outside_resource, metadata=cls.author, value="Alice"
		)
		models.MetadataResourceValue.objects.create(
		resource=cls.direct_resource, metadata=cls.author, value="Bob"
		)
		models.MetadataResourceValue.objects.create(
		resource=cls.direct_resource, metadata=cls.copyright, value="Alice"
		)
		cls.role = models.Role.objects.create(project=cls.project, label="editor")
		models.ProjectAccess.objects.create(
		project=cls.project, user=cls.user, role=cls.role
		)
		set_permission_on_role(cls.role, "collection", can_read=True, can_update=True)
		set_permission_on_role(cls.role, "resource", can_read=True, can_update=True)

		def _value_ids(self) -> tuple[set[int], set[int]]:
		return (
		set(models.MetadataCollectionValue.objects.values_list("pk", flat=True)),
		set(models.MetadataResourceValue.objects.values_list("pk", flat=True)),
		)

		def _remove(self, *, recursive: bool = True) -> bool:
		return methods.remove_meta_value_from_collection(
		self.user, self.collection.pk, self.source_value.pk, recursive
		)

		def test_recursive_removal_matches_field_and_value_instead_of_row_id(self):
		collection_ids, resource_ids = self._value_ids()

		self.assertTrue(self._remove())

		self.assertEqual(
		self._value_ids(),
		(
		collection_ids
		- {self.source_value.pk, self.child_value.pk, self.grandchild_value.pk},
		resource_ids - {self.direct_value.pk, self.shared_value.pk},
		),
		)
		self.assertTrue(
		models.MetadataResourceValue.objects.filter(
		pk=self.colliding_resource_value.pk, value="Museum"
		).exists()
		)

		def test_nonrecursive_removal_only_deletes_the_selected_collection_value(self):
		collection_ids, resource_ids = self._value_ids()

		self.assertTrue(self._remove(recursive=False))

		self.assertEqual(
		self._value_ids(), (collection_ids - {self.source_value.pk}, resource_ids)
		)

		def test_missing_or_other_collection_value_does_not_delete_any_descendant_value(
		self,
		):
		initial_ids = self._value_ids()
		for value_id in (99999, self.other_collection_value.pk):
		with self.subTest(value_id=value_id):
		self.assertTrue(
		methods.remove_meta_value_from_collection(
		self.user, self.collection.pk, value_id, True
		)
		)
		self.assertEqual(self._value_ids(), initial_ids)

		def test_denied_source_prevents_removal(self):
		initial_ids = self._value_ids()
		set_permission_on_role(self.role, self.collection, can_read=True)

		with self.assertRaises(ServiceException):
		self._remove()

		self.assertEqual(self._value_ids(), initial_ids)

		def test_denied_descendant_collection_prevents_all_removal(self):
		initial_ids = self._value_ids()
		set_permission_on_role(self.role, self.grandchild, can_read=True)

		with self.assertRaises(ServiceException):
		self._remove()

		self.assertEqual(self._value_ids(), initial_ids)

		def test_denied_descendant_resource_prevents_all_removal(self):
		initial_ids = self._value_ids()
		set_permission_on_role(self.role, self.shared_resource, can_read=True)

		with self.assertRaises(ServiceException):
		self._remove()

		self.assertEqual(self._value_ids(), initial_ids)

		def test_inherited_resource_update_access_allows_removal(self):
		set_permission_on_role(self.role, "resource", can_read=True)
		set_permission_on_role(
		self.role, self.collection, can_read=True, can_update=True
		)

		self.assertTrue(self._remove())

		self.assertFalse(
		models.MetadataResourceValue.objects.filter(
		pk__in=[self.direct_value.pk, self.shared_value.pk]
		).exists()
		)

		def test_exif_protection_prevents_all_removal(self):
		initial_ids = self._value_ids()
		self.metadata_set.title = "ExifTool"
		self.metadata_set.save()

		with self.assertRaisesMessage(ServiceException, "can't modify exif"):
		self._remove()

		self.assertEqual(self._value_ids(), initial_ids)

		def test_deleted_collections_and_resources_are_preserved(self):
		deleted_collection = models.Collection.objects.create(
		project=self.project,
		parent=self.child,
		title="Deleted collection",
		deleted_at=timezone.now(),
		)
		hidden_child = models.Collection.objects.create(
		project=self.project, parent=deleted_collection, title="Hidden child"
		)
		deleted_resource = models.Resource.objects.create(
		ptr_project=self.project,
		title="Deleted resource",
		deleted_at=timezone.now(),
		)
		hidden_resource = models.Resource.objects.create(
		ptr_project=self.project, title="Hidden resource"
		)
		self.collection.resources.add(deleted_resource)
		hidden_child.resources.add(hidden_resource)
		preserved_collection_ids = {
		models.MetadataCollectionValue.objects.create(
		collection=collection, metadata=self.author, value="Alice"
		).pk
		for collection in (deleted_collection, hidden_child)
		}
		preserved_resource_ids = {
		models.MetadataResourceValue.objects.create(
		resource=resource, metadata=self.author, value="Alice"
		).pk
		for resource in (deleted_resource, hidden_resource)
		}
		set_permission_on_role(self.role, deleted_collection, can_read=True)
		set_permission_on_role(self.role, deleted_resource, can_read=True)

		self.assertTrue(self._remove())

		collection_ids, resource_ids = self._value_ids()
		self.assertTrue(preserved_collection_ids <= collection_ids)
		self.assertTrue(preserved_resource_ids <= resource_ids)

		def test_failure_during_resource_deletion_rolls_back_collection_deletions(self):
		initial_ids = self._value_ids()
		original_delete = methods.QuerySet.delete

		def fail_resource_deletion(queryset):
		if queryset.model is models.MetadataResourceValue:
		raise RuntimeError("resource deletion failed")
		return original_delete(queryset)

		with (
		patch.object(
		methods.QuerySet,
		"delete",
		autospec=True,
		side_effect=fail_resource_deletion,
		),
		self.assertRaisesMessage(RuntimeError, "resource deletion failed"),
		):
		self._remove()

		self.assertEqual(self._value_ids(), initial_ids)

		def test_queries_are_batched_for_wide_subtrees(self):
		set_permission_on_role(self.role, "resource", can_read=True)
		select_counts = []
		for width in (1, 21):
		source = models.Collection.objects.create(
		project=self.project, parent=self.root, title=f"Wide subtree {width}"
		)
		source_value = models.MetadataCollectionValue.objects.create(
		collection=source, metadata=self.author, value="Alice"
		)
		set_permission_on_role(self.role, source, can_read=True, can_update=True)
		resource_value_ids = []
		collection_value_ids = [source_value.pk]
		for index in range(width):
		child = models.Collection.objects.create(
		project=self.project, parent=source, title=f"Child {index}"
		)
		resource = models.Resource.objects.create(
		ptr_project=self.project, title=f"Resource {index}"
		)
		child.resources.add(resource)
		collection_value_ids.append(
		models.MetadataCollectionValue.objects.create(
		collection=child, metadata=self.author, value="Alice"
		).pk
		)
		resource_value_ids.append(
		models.MetadataResourceValue.objects.create(
		resource=resource, metadata=self.author, value="Alice"
		).pk
		)

		with CaptureQueriesContext(connection) as queries:
		methods.remove_meta_value_from_collection(
		self.user, source.pk, source_value.pk, True
		)

		select_counts.append(
		sum(
		query["sql"].lstrip().upper().startswith("SELECT")
		for query in queries
		)
		)
		self.assertFalse(
		models.MetadataCollectionValue.objects.filter(
		pk__in=collection_value_ids
		).exists()
		)
		self.assertFalse(
		models.MetadataResourceValue.objects.filter(
		pk__in=resource_value_ids
		).exists()
		)
		self.assertEqual(select_counts[0], select_counts[1])


		class RestoreCollectionAccessTestCase(TestCase):
		@classmethod
		def setUpTestData(cls):