Loading src/rpc/methods.py +1 −0 Original line number Diff line number Diff line Loading @@ -560,6 +560,7 @@ def collections( recursive=recursive, only_published=only_published, cache=serializer_cache, acl=acl if recursive else None, ) ) return data Loading src/rpc/serializers.py +25 −7 Original line number Diff line number Diff line from resources import models from resources.acl import UserAccess from jama import settings import os Loading Loading @@ -155,7 +156,18 @@ def collection( recursive: bool = False, only_published: bool = False, cache: SerializerCache = None, acl: UserAccess | None = None, ) -> dict: if recursive and acl is None: raise ValueError("Recursive collection serialization requires UserAccess") visible_children = None if recursive: children_query = collection_instance.children().filter(project=acl.project) if only_published: children_query = children_query.filter(public_access=True) visible_children = [child for child in children_query if acl.can_read(child)] tags = [] for tag_instance in collection_instance.tags.all(): tags.append(tag(tag_instance, cache=cache)) Loading @@ -175,7 +187,9 @@ def collection( "rpc_children_count", None, ) if children_count is None: if visible_children is not None: children_count = len(visible_children) elif children_count is None: children_count = collection_instance.children().count() # Ancestors may be attached by the caller after batch hydration; otherwise # use the model method for standalone serialization. Loading Loading @@ -211,12 +225,16 @@ def collection( "ancestors": ancestors, } if recursive: payload["children"] = [] for child in collection_instance.children(): # filter private content if only_published and child.public_access is False: continue payload["children"].append(collection(child, recursive, cache=cache)) payload["children"] = [ collection( child, recursive=True, only_published=only_published, cache=cache, acl=acl, ) for child in visible_children ] for prop in collection_instance.metadatacollectionvalue_set.all(): payload["metas"].append(metadata_collection_value(prop, cache=cache)) Loading src/rpc/tests.py +104 −0 Original line number Diff line number Diff line Loading @@ -776,6 +776,110 @@ class ServiceTestCase(TestCase): "some_unknown_class" ) def test_recursive_collections_filter_scoped_acl_and_project(self): top = models.Collection.objects.create( title="recursive top", parent=self.test_project_root_collection, project=self.test_project, ) visible_child = models.Collection.objects.create( title="visible child", parent=top, project=self.test_project, ) hidden_child = models.Collection.objects.create( title="hidden child", parent=top, project=self.test_project, ) readable_hidden_descendant = models.Collection.objects.create( title="readable hidden descendant", parent=hidden_child, project=self.test_project, ) other_project = models.Project.objects.create(label="recursive other project") models.Collection.objects.create( title="cross-project child", parent=top, project=other_project, ) models.ObjectPermission.objects.create( object_class="collection", object_pk=hidden_child.pk, role=self.admin_role, object_read=False, ) models.ObjectPermission.objects.create( object_class="collection", object_pk=readable_hidden_descendant.pk, role=self.admin_role, object_read=True, ) payload = methods.collections( self.test_user, parent_id=self.test_project_root_collection.pk, recursive=True, )[0] self.assertTrue( UserAccess(self.test_user, self.test_project).can_read( readable_hidden_descendant ) ) self.assertEqual(payload["children_count"], 1) self.assertEqual( [child["id"] for child in payload["children"]], [visible_child.pk] ) def test_recursive_collections_apply_only_published_at_every_depth(self): top = models.Collection.objects.create( title="published top", parent=self.test_project_root_collection, project=self.test_project, public_access=True, ) child = models.Collection.objects.create( title="published child", parent=top, project=self.test_project, public_access=True, ) private_grandchild = models.Collection.objects.create( title="private grandchild", parent=child, project=self.test_project, public_access=False, ) models.Collection.objects.create( title="published below private", parent=private_grandchild, project=self.test_project, public_access=True, ) payload = methods.collections( self.test_user, parent_id=self.test_project_root_collection.pk, recursive=True, only_published=True, )[0] child_payload = payload["children"][0] self.assertEqual(child_payload["id"], child.pk) self.assertEqual(child_payload["children_count"], 0) self.assertEqual(child_payload["children"], []) def test_recursive_collection_serializer_requires_acl(self): collection = models.Collection.objects.create( title="recursive serialization guard", parent=self.test_project_root_collection, project=self.test_project, ) with self.assertRaisesRegex(ValueError, "requires UserAccess"): serializers.collection(collection, recursive=True) def test_add_collection(self): methods.add_collection( self.test_user, Loading Loading
src/rpc/methods.py +1 −0 Original line number Diff line number Diff line Loading @@ -560,6 +560,7 @@ def collections( recursive=recursive, only_published=only_published, cache=serializer_cache, acl=acl if recursive else None, ) ) return data Loading
src/rpc/serializers.py +25 −7 Original line number Diff line number Diff line from resources import models from resources.acl import UserAccess from jama import settings import os Loading Loading @@ -155,7 +156,18 @@ def collection( recursive: bool = False, only_published: bool = False, cache: SerializerCache = None, acl: UserAccess | None = None, ) -> dict: if recursive and acl is None: raise ValueError("Recursive collection serialization requires UserAccess") visible_children = None if recursive: children_query = collection_instance.children().filter(project=acl.project) if only_published: children_query = children_query.filter(public_access=True) visible_children = [child for child in children_query if acl.can_read(child)] tags = [] for tag_instance in collection_instance.tags.all(): tags.append(tag(tag_instance, cache=cache)) Loading @@ -175,7 +187,9 @@ def collection( "rpc_children_count", None, ) if children_count is None: if visible_children is not None: children_count = len(visible_children) elif children_count is None: children_count = collection_instance.children().count() # Ancestors may be attached by the caller after batch hydration; otherwise # use the model method for standalone serialization. Loading Loading @@ -211,12 +225,16 @@ def collection( "ancestors": ancestors, } if recursive: payload["children"] = [] for child in collection_instance.children(): # filter private content if only_published and child.public_access is False: continue payload["children"].append(collection(child, recursive, cache=cache)) payload["children"] = [ collection( child, recursive=True, only_published=only_published, cache=cache, acl=acl, ) for child in visible_children ] for prop in collection_instance.metadatacollectionvalue_set.all(): payload["metas"].append(metadata_collection_value(prop, cache=cache)) Loading
src/rpc/tests.py +104 −0 Original line number Diff line number Diff line Loading @@ -776,6 +776,110 @@ class ServiceTestCase(TestCase): "some_unknown_class" ) def test_recursive_collections_filter_scoped_acl_and_project(self): top = models.Collection.objects.create( title="recursive top", parent=self.test_project_root_collection, project=self.test_project, ) visible_child = models.Collection.objects.create( title="visible child", parent=top, project=self.test_project, ) hidden_child = models.Collection.objects.create( title="hidden child", parent=top, project=self.test_project, ) readable_hidden_descendant = models.Collection.objects.create( title="readable hidden descendant", parent=hidden_child, project=self.test_project, ) other_project = models.Project.objects.create(label="recursive other project") models.Collection.objects.create( title="cross-project child", parent=top, project=other_project, ) models.ObjectPermission.objects.create( object_class="collection", object_pk=hidden_child.pk, role=self.admin_role, object_read=False, ) models.ObjectPermission.objects.create( object_class="collection", object_pk=readable_hidden_descendant.pk, role=self.admin_role, object_read=True, ) payload = methods.collections( self.test_user, parent_id=self.test_project_root_collection.pk, recursive=True, )[0] self.assertTrue( UserAccess(self.test_user, self.test_project).can_read( readable_hidden_descendant ) ) self.assertEqual(payload["children_count"], 1) self.assertEqual( [child["id"] for child in payload["children"]], [visible_child.pk] ) def test_recursive_collections_apply_only_published_at_every_depth(self): top = models.Collection.objects.create( title="published top", parent=self.test_project_root_collection, project=self.test_project, public_access=True, ) child = models.Collection.objects.create( title="published child", parent=top, project=self.test_project, public_access=True, ) private_grandchild = models.Collection.objects.create( title="private grandchild", parent=child, project=self.test_project, public_access=False, ) models.Collection.objects.create( title="published below private", parent=private_grandchild, project=self.test_project, public_access=True, ) payload = methods.collections( self.test_user, parent_id=self.test_project_root_collection.pk, recursive=True, only_published=True, )[0] child_payload = payload["children"][0] self.assertEqual(child_payload["id"], child.pk) self.assertEqual(child_payload["children_count"], 0) self.assertEqual(child_payload["children"], []) def test_recursive_collection_serializer_requires_acl(self): collection = models.Collection.objects.create( title="recursive serialization guard", parent=self.test_project_root_collection, project=self.test_project, ) with self.assertRaisesRegex(ValueError, "requires UserAccess"): serializers.collection(collection, recursive=True) def test_add_collection(self): methods.add_collection( self.test_user, Loading