Loading src/jama/__init__.py +1 −1 Changes for src/jama/__init__.py: 1 added line, 1 removed line. Original line number Diff line number Diff line __version__ = "2026.10.6.0" __version__ = "2026.10.13.0" src/rpc/methods.py +41 −23 Changes for src/rpc/methods.py: 41 added lines, 23 removed lines. Original line number Diff line number Diff line Loading @@ -2165,39 +2165,57 @@ def ancestors_from_resource(user: User, resource_id: int) -> List[List[dict]]: return ancestors @_rpc_groups(["Collections"]) def publish_collection(user: User, collection_id: int) -> bool: """ Mark a collection as public """ query_set: Union[Iterator[Collection], QuerySet] = Collection.objects.filter( pk=collection_id, deleted_at__isnull=True @transaction.atomic def _set_collection_public_access( user: User, collection_id: int, public_access: bool ) -> bool: collection_instance = ( Collection.objects.filter(pk=collection_id, deleted_at__isnull=True) .select_related("project") .first() ) collection_instance: Union[Collection, None] = query_set.first() if not collection_instance: raise ServiceException(NO_SUCH_COLLECTION) UserAccess(user, collection_instance.project).check_update(collection_instance) collection_instance.public_access = True collection_instance.published_at = timezone.now() collection_instance.save() acl = UserAccess(user, collection_instance.project) acl.check_update(collection_instance) 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 child in children: 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] published_at = timezone.now() if public_access else None for collection in collections_by_id.values(): collection.public_access = public_access if public_access: collection.published_at = published_at collection.save() return True @_rpc_groups(["Collections"]) def publish_collection(user: User, collection_id: int) -> bool: """ Mark a collection and its active descendants as public. """ return _set_collection_public_access(user, collection_id, True) @_rpc_groups(["Collections"]) def unpublish_collection(user: User, collection_id: int) -> bool: """ Mark a collection as private Mark a collection and its active descendants as private. """ query_set: Union[Iterator[Collection], QuerySet] = Collection.objects.filter( pk=collection_id, deleted_at__isnull=True ) collection_instance: Union[Collection, None] = query_set.first() if not collection_instance: raise ServiceException(NO_SUCH_COLLECTION) UserAccess(user, collection_instance.project).check_update(collection_instance) collection_instance.public_access = False collection_instance.save() return True return _set_collection_public_access(user, collection_id, False) @_rpc_groups(["Collections"]) Loading src/rpc/tests.py +142 −0 Changes for src/rpc/tests.py: 142 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -40,6 +40,148 @@ object_classes = [ crud = ["create", "read", "update", "delete"] class CollectionPublicationTestCase(TestCase): @classmethod def setUpTestData(cls): cls.user = User.objects.create(username="collection_publication_user") cls.project = models.Project.objects.create(label="collection publication") cls.root = models.Collection.objects.create(project=cls.project, title="Root") cls.collection = models.Collection.objects.create( project=cls.project, parent=cls.root, title="Collection" ) 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.sibling = models.Collection.objects.create( project=cls.project, parent=cls.root, title="Sibling" ) cls.deleted_child = models.Collection.objects.create( project=cls.project, parent=cls.collection, title="Deleted child", deleted_at=timezone.now(), ) cls.hidden_grandchild = models.Collection.objects.create( project=cls.project, parent=cls.deleted_child, title="Hidden grandchild" ) 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) def _publication_state(self) -> list[tuple]: return list( models.Collection.objects.order_by("pk").values_list( "pk", "public_access", "published_at", "updated_at" ) ) def test_publication_status_applies_to_the_active_subtree(self): subtree_ids = [self.collection.pk, self.child.pk, self.grandchild.pk] for public_access, method in ( (True, methods.publish_collection), (False, methods.unpublish_collection), ): with self.subTest(public_access=public_access): old_published_at = timezone.now() - timezone.timedelta(days=1) models.Collection.objects.update( public_access=not public_access, published_at=old_published_at ) # Already-matching parents must still propagate to descendants. models.Collection.objects.filter(pk=self.collection.pk).update( public_access=public_access ) outside_state = [ state for state in self._publication_state() if state[0] not in subtree_ids ] self.assertTrue(method(self.user, self.collection.pk)) subtree = models.Collection.objects.filter(pk__in=subtree_ids) self.assertEqual( list(subtree.values_list("public_access", flat=True)), [public_access] * 3, ) published_dates = set(subtree.values_list("published_at", flat=True)) if public_access: self.assertEqual(len(published_dates), 1) self.assertGreater(published_dates.pop(), old_published_at) else: self.assertEqual(published_dates, {old_published_at}) self.assertEqual( [ state for state in self._publication_state() if state[0] not in subtree_ids ], outside_state, ) def test_denied_collection_prevents_all_publication_changes(self): for denied_collection in (self.collection, self.grandchild): set_permission_on_role(self.role, denied_collection, can_read=True) for method in (methods.publish_collection, methods.unpublish_collection): with self.subTest(method=method.__name__, denied=denied_collection.pk): models.Collection.objects.update( public_access=method == methods.unpublish_collection ) initial_state = self._publication_state() with self.assertRaises(ServiceException): method(self.user, self.collection.pk) self.assertEqual(self._publication_state(), initial_state) set_permission_on_role( self.role, denied_collection, can_read=True, can_update=True ) def test_inherited_update_access_allows_publication(self): set_permission_on_role(self.role, "collection", can_read=True) set_permission_on_role( self.role, self.collection, can_read=True, can_update=True ) self.assertTrue(methods.publish_collection(self.user, self.collection.pk)) self.assertEqual( models.Collection.objects.filter(public_access=True).count(), 3 ) def test_missing_or_deleted_collection_is_rejected(self): initial_state = self._publication_state() for collection_id in (99999, self.deleted_child.pk): for method in (methods.publish_collection, methods.unpublish_collection): with self.subTest(method=method.__name__, collection_id=collection_id): with self.assertRaisesMessage( ServiceException, "no such collection" ): method(self.user, collection_id) self.assertEqual(self._publication_state(), initial_state) def test_foreign_project_descendant_prevents_all_publication_changes(self): foreign_project = models.Project.objects.create(label="foreign publication") models.Collection.objects.create( project=foreign_project, parent=self.child, title="Foreign child" ) for method in (methods.publish_collection, methods.unpublish_collection): with self.subTest(method=method.__name__): models.Collection.objects.update( public_access=method == methods.unpublish_collection ) initial_state = self._publication_state() with self.assertRaises(ServiceException): method(self.user, self.collection.pk) self.assertEqual(self._publication_state(), initial_state) class RecursiveMetadataRemovalTestCase(TestCase): @classmethod def setUpTestData(cls): Loading Loading
src/jama/__init__.py +1 −1 Changes for src/jama/__init__.py: 1 added line, 1 removed line. Original line number Diff line number Diff line __version__ = "2026.10.6.0" __version__ = "2026.10.13.0"
src/rpc/methods.py +41 −23 Changes for src/rpc/methods.py: 41 added lines, 23 removed lines. Original line number Diff line number Diff line Loading @@ -2165,39 +2165,57 @@ def ancestors_from_resource(user: User, resource_id: int) -> List[List[dict]]: return ancestors @_rpc_groups(["Collections"]) def publish_collection(user: User, collection_id: int) -> bool: """ Mark a collection as public """ query_set: Union[Iterator[Collection], QuerySet] = Collection.objects.filter( pk=collection_id, deleted_at__isnull=True @transaction.atomic def _set_collection_public_access( user: User, collection_id: int, public_access: bool ) -> bool: collection_instance = ( Collection.objects.filter(pk=collection_id, deleted_at__isnull=True) .select_related("project") .first() ) collection_instance: Union[Collection, None] = query_set.first() if not collection_instance: raise ServiceException(NO_SUCH_COLLECTION) UserAccess(user, collection_instance.project).check_update(collection_instance) collection_instance.public_access = True collection_instance.published_at = timezone.now() collection_instance.save() acl = UserAccess(user, collection_instance.project) acl.check_update(collection_instance) 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 child in children: 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] published_at = timezone.now() if public_access else None for collection in collections_by_id.values(): collection.public_access = public_access if public_access: collection.published_at = published_at collection.save() return True @_rpc_groups(["Collections"]) def publish_collection(user: User, collection_id: int) -> bool: """ Mark a collection and its active descendants as public. """ return _set_collection_public_access(user, collection_id, True) @_rpc_groups(["Collections"]) def unpublish_collection(user: User, collection_id: int) -> bool: """ Mark a collection as private Mark a collection and its active descendants as private. """ query_set: Union[Iterator[Collection], QuerySet] = Collection.objects.filter( pk=collection_id, deleted_at__isnull=True ) collection_instance: Union[Collection, None] = query_set.first() if not collection_instance: raise ServiceException(NO_SUCH_COLLECTION) UserAccess(user, collection_instance.project).check_update(collection_instance) collection_instance.public_access = False collection_instance.save() return True return _set_collection_public_access(user, collection_id, False) @_rpc_groups(["Collections"]) Loading
src/rpc/tests.py +142 −0 Changes for src/rpc/tests.py: 142 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -40,6 +40,148 @@ object_classes = [ crud = ["create", "read", "update", "delete"] class CollectionPublicationTestCase(TestCase): @classmethod def setUpTestData(cls): cls.user = User.objects.create(username="collection_publication_user") cls.project = models.Project.objects.create(label="collection publication") cls.root = models.Collection.objects.create(project=cls.project, title="Root") cls.collection = models.Collection.objects.create( project=cls.project, parent=cls.root, title="Collection" ) 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.sibling = models.Collection.objects.create( project=cls.project, parent=cls.root, title="Sibling" ) cls.deleted_child = models.Collection.objects.create( project=cls.project, parent=cls.collection, title="Deleted child", deleted_at=timezone.now(), ) cls.hidden_grandchild = models.Collection.objects.create( project=cls.project, parent=cls.deleted_child, title="Hidden grandchild" ) 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) def _publication_state(self) -> list[tuple]: return list( models.Collection.objects.order_by("pk").values_list( "pk", "public_access", "published_at", "updated_at" ) ) def test_publication_status_applies_to_the_active_subtree(self): subtree_ids = [self.collection.pk, self.child.pk, self.grandchild.pk] for public_access, method in ( (True, methods.publish_collection), (False, methods.unpublish_collection), ): with self.subTest(public_access=public_access): old_published_at = timezone.now() - timezone.timedelta(days=1) models.Collection.objects.update( public_access=not public_access, published_at=old_published_at ) # Already-matching parents must still propagate to descendants. models.Collection.objects.filter(pk=self.collection.pk).update( public_access=public_access ) outside_state = [ state for state in self._publication_state() if state[0] not in subtree_ids ] self.assertTrue(method(self.user, self.collection.pk)) subtree = models.Collection.objects.filter(pk__in=subtree_ids) self.assertEqual( list(subtree.values_list("public_access", flat=True)), [public_access] * 3, ) published_dates = set(subtree.values_list("published_at", flat=True)) if public_access: self.assertEqual(len(published_dates), 1) self.assertGreater(published_dates.pop(), old_published_at) else: self.assertEqual(published_dates, {old_published_at}) self.assertEqual( [ state for state in self._publication_state() if state[0] not in subtree_ids ], outside_state, ) def test_denied_collection_prevents_all_publication_changes(self): for denied_collection in (self.collection, self.grandchild): set_permission_on_role(self.role, denied_collection, can_read=True) for method in (methods.publish_collection, methods.unpublish_collection): with self.subTest(method=method.__name__, denied=denied_collection.pk): models.Collection.objects.update( public_access=method == methods.unpublish_collection ) initial_state = self._publication_state() with self.assertRaises(ServiceException): method(self.user, self.collection.pk) self.assertEqual(self._publication_state(), initial_state) set_permission_on_role( self.role, denied_collection, can_read=True, can_update=True ) def test_inherited_update_access_allows_publication(self): set_permission_on_role(self.role, "collection", can_read=True) set_permission_on_role( self.role, self.collection, can_read=True, can_update=True ) self.assertTrue(methods.publish_collection(self.user, self.collection.pk)) self.assertEqual( models.Collection.objects.filter(public_access=True).count(), 3 ) def test_missing_or_deleted_collection_is_rejected(self): initial_state = self._publication_state() for collection_id in (99999, self.deleted_child.pk): for method in (methods.publish_collection, methods.unpublish_collection): with self.subTest(method=method.__name__, collection_id=collection_id): with self.assertRaisesMessage( ServiceException, "no such collection" ): method(self.user, collection_id) self.assertEqual(self._publication_state(), initial_state) def test_foreign_project_descendant_prevents_all_publication_changes(self): foreign_project = models.Project.objects.create(label="foreign publication") models.Collection.objects.create( project=foreign_project, parent=self.child, title="Foreign child" ) for method in (methods.publish_collection, methods.unpublish_collection): with self.subTest(method=method.__name__): models.Collection.objects.update( public_access=method == methods.unpublish_collection ) initial_state = self._publication_state() with self.assertRaises(ServiceException): method(self.user, self.collection.pk) self.assertEqual(self._publication_state(), initial_state) class RecursiveMetadataRemovalTestCase(TestCase): @classmethod def setUpTestData(cls): Loading
