Merge pull request galaxyproject#16904 from jdavcs/dev_munge_lists_bug

Fix subtle bug in listify function + simplify list munging

lib/galaxy/managers/base.py

@@ -194,24 +194,6 @@ def get_object(trans, id, class_name, check_ownership=False, check_accessible=Fa
 
 
 # =============================================================================
-def munge_lists(listA, listB):
-    """
-    Combine two lists into a single list.
-
-    (While allowing them to be None, non-lists, or lists.)
-    """
-    # TODO: there's nothing specifically filter or model-related here - move to util
-    if listA is None:
-        return listB
-    if listB is None:
-        return listA
-    if not isinstance(listA, list):
-        listA = [listA]
-    if not isinstance(listB, list):
-        listB = [listB]
-    return listA + listB
-
-
 U = TypeVar("U", bound=model._HasTable)
 
 
@@ -288,14 +270,6 @@ def _apply_orm_filters(self, query: Query, filters) -> Query:
             query = query.filter(filter)
         return query
 
-    def _munge_filters(self, filtersA, filtersB):
-        """
-        Combine two lists into a single list.
-
-        (While allowing them to be None, non-lists, or lists.)
-        """
-        return munge_lists(filtersA, filtersB)
-
     # .... order, limit, and offset
     def _apply_order_by(self, query: Query, order_by) -> Query:
         """
@@ -487,7 +461,7 @@ def by_ids(self, ids, filters=None, **kwargs):
         if not ids:
             return []
         ids_filter = parsed_filter("orm", self.model_class.__table__.c.id.in_(ids))
-        found = self.list(filters=self._munge_filters(ids_filter, filters), **kwargs)
+        found = self.list(filters=combine_lists(ids_filter, filters), **kwargs)
         # TODO: this does not order by the original 'ids' array
 
         # ...could use get (supposedly since found are in the session, the db won't be hit twice)
@@ -1389,3 +1363,24 @@ def get_archived(
     def cleanup_items(self, user: model.User, item_ids: Set[int]) -> StorageItemsCleanupResult:
         """Purges the given list of items by ID. The items must be owned by the user."""
         raise NotImplementedError
+
+
+def combine_lists(listA: Any, listB: Any) -> List:
+    """
+    Combine two lists into a single list.
+
+    Arguments can be None, non-lists, or lists. If an argument is None, it will
+    not be included in the returned list. If both arguments are None, an empty
+    list will be returned.
+    """
+
+    def make_list(item):
+        # Check for None explicitly: __bool__ may be overwritten.
+        if item is None:
+            return []
+        elif isinstance(item, list):
+            return item
+        else:
+            return [item]
+
+    return make_list(listA) + make_list(listB)

lib/galaxy/managers/histories.py

@@ -37,6 +37,7 @@
     sharable,
 )
 from galaxy.managers.base import (
+    combine_lists,
     ModelDeserializingError,
     Serializer,
     SortableManager,
@@ -135,7 +136,7 @@ def most_recent(self, user, filters=None, current_history=None, **kwargs):
         if self.user_manager.is_anonymous(user):
             return None if (not current_history or current_history.deleted) else current_history
         desc_update_time = desc(self.model_class.update_time)
-        filters = self._munge_filters(filters, self.model_class.user_id == user.id)
+        filters = combine_lists(filters, self.model_class.user_id == user.id)
         # TODO: normalize this return value
         return self.query(filters=filters, order_by=desc_update_time, limit=1, **kwargs).first()
 

lib/galaxy/managers/sharable.py

@@ -34,6 +34,7 @@
     taggable,
     users,
 )
+from galaxy.managers.base import combine_lists
 from galaxy.model import (
     User,
     UserShareAssociation,
@@ -83,7 +84,7 @@ def by_user(self, user: User, **kwargs: Any) -> List[Any]:
         `user`.
         """
         user_filter = self.model_class.table.c.user_id == user.id
-        filters = self._munge_filters(user_filter, kwargs.get("filters", None))
+        filters = combine_lists(user_filter, kwargs.get("filters", None))
         return self.list(filters=filters, **kwargs)
 
     # .... owned/accessible interfaces
@@ -154,15 +155,15 @@ def _query_published(self, filters=None, **kwargs):
         Return a query for all published items.
         """
         published_filter = self.model_class.table.c.published == true()
-        filters = self._munge_filters(published_filter, filters)
+        filters = combine_lists(published_filter, filters)
         return self.query(filters=filters, **kwargs)
 
     def list_published(self, filters=None, **kwargs):
         """
         Return a list of all published items.
         """
         published_filter = self.model_class.table.c.published == true()
-        filters = self._munge_filters(published_filter, filters)
+        filters = combine_lists(published_filter, filters)
         return self.list(filters=filters, **kwargs)
 
     # .... user sharing

lib/galaxy/managers/users.py

@@ -37,6 +37,7 @@
     base,
     deletable,
 )
+from galaxy.managers.base import combine_lists
 from galaxy.model import (
     User,
     UserAddress,
@@ -267,7 +268,7 @@ def by_email(self, email: str, filters=None, **kwargs) -> Optional[model.User]:
         """
         Find a user by their email.
         """
-        filters = self._munge_filters(self.model_class.email == email, filters)
+        filters = combine_lists(self.model_class.email == email, filters)
         try:
             # TODO: use one_or_none
             return super().one(filters=filters, **kwargs)
@@ -321,7 +322,7 @@ def admins(self, filters=None, **kwargs):
         Return a list of admin Users.
         """
         admin_emails = self.app.config.admin_users_list
-        filters = self._munge_filters(self.model_class.email.in_(admin_emails), filters)
+        filters = combine_lists(self.model_class.email.in_(admin_emails), filters)
         return super().list(filters=filters, **kwargs)
 
     def error_unless_admin(self, user, msg="Administrators only", **kwargs):

test/unit/app/managers/test_base.py

@@ -0,0 +1,18 @@
+from galaxy.managers.base import combine_lists
+
+
+class NotFalsy:
+    """Class requires explicit check for None"""
+
+    def __bool__(self):
+        raise Exception("not implemented")
+
+
+def test_combine_lists():
+    foo, bar = NotFalsy(), NotFalsy()
+    assert combine_lists(foo, None) == [foo]
+    assert combine_lists(None, foo) == [foo]
+    assert combine_lists(foo, bar) == [foo, bar]
+    assert combine_lists([foo, bar], None) == [foo, bar]
+    assert combine_lists(None, [foo, bar]) == [foo, bar]
+    assert combine_lists(None, None) == []