[WIP] SA2.0, model/data-access edits, unit testing

lib/galaxy/managers/hdcas.py

@@ -8,6 +8,8 @@
 import logging
 from typing import Dict
 
+from sqlalchemy import select
+
 from galaxy import model
 from galaxy.managers import (
     annotatable,
@@ -334,3 +336,12 @@ def serialize_job_state_summary(self, item, key, **context):
     def serialize_elements_datatypes(self, item, key, **context):
         extensions_set = item.dataset_dbkeys_and_extensions_summary[1]
         return list(extensions_set)
+
+
+def get_hdca_by_name(session, name):
+    stmt = (
+        select(model.HistoryDatasetCollectionAssociation)
+        .where(model.HistoryDatasetCollectionAssociation.name == name)
+        .limit(1)
+    )
+    return session.scalars(stmt).first()

lib/galaxy/managers/users.py

@@ -20,7 +20,9 @@
 from sqlalchemy import (
     and_,
     exc,
+    false,
     func,
+    or_,
     select,
     true,
 )
@@ -893,3 +895,35 @@ def get_user_by_email(session, email: str, model_class=User, case_sensitive=True
 def get_user_by_username(session, username: str, model_class=User):
     stmt = select(model_class).filter(model_class.username == username).limit(1)
     return session.scalars(stmt).first()
+
+
+def get_users_for_index(
+    session,
+    deleted: bool,
+    f_email: Optional[str] = None,
+    f_name: Optional[str] = None,
+    f_any: Optional[str] = None,
+    is_admin: bool = False,
+    expose_user_email: bool = False,
+    expose_user_name: bool = False,
+):
+    stmt = select(User)
+    if f_email and (is_admin or expose_user_email):
+        stmt = stmt.where(User.email.like(f"%{f_email}%"))
+    if f_name and (is_admin or expose_user_name):
+        stmt = stmt.where(User.username.like(f"%{f_name}%"))
+    if f_any:
+        if is_admin:
+            stmt = stmt.where(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
+        else:
+            if expose_user_email and expose_user_name:
+                stmt = stmt.where(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
+            elif expose_user_email:
+                stmt = stmt.where(User.email.like(f"%{f_any}%"))
+            elif expose_user_name:
+                stmt = stmt.where(User.username.like(f"%{f_any}%"))
+    if deleted:
+        stmt = stmt.where(User.deleted == true())
+    else:
+        stmt = stmt.where(User.deleted == false())
+    return session.scalars(stmt).all()

lib/galaxy/model/security.py

@@ -80,22 +80,11 @@ def sort_by_attr(self, seq, attr):
         intermed.sort()
         return [_[-1] for _ in intermed]
 
-    def _get_npns_roles(self, trans):
-        """
-        non-private, non-sharing roles
-        """
-        stmt = (
-            select(Role)
-            .where(and_(Role.deleted == false(), Role.type != Role.types.PRIVATE, Role.type != Role.types.SHARING))
-            .order_by(Role.name)
-        )
-        return trans.sa_session.scalars(stmt)
-
     def get_all_roles(self, trans, cntrller):
         admin_controller = cntrller in ["library_admin"]
         roles = set()
         if not trans.user:
-            return self._get_npns_roles(trans)
+            return get_npns_roles(trans.sa_session)
         if admin_controller:
             # The library is public and the user is an admin, so all roles are legitimate
             stmt = select(Role).where(Role.deleted == false()).order_by(Role.name)
@@ -108,7 +97,7 @@ def get_all_roles(self, trans, cntrller):
             for role in self.get_sharing_roles(trans.user):
                 roles.add(role)
             # Add all remaining non-private, non-sharing roles
-            for role in self._get_npns_roles(trans):
+            for role in get_npns_roles(trans.sa_session):
                 roles.add(role)
         return self.sort_by_attr(list(roles), "name")
 
@@ -189,7 +178,7 @@ def get_valid_roles(self, trans, item, query=None, page=None, page_limit=None, i
             for role in self.get_sharing_roles(trans.user):
                 roles.append(role)
             # Add all remaining non-private, non-sharing roles
-            for role in self._get_npns_roles(trans):
+            for role in get_npns_roles(trans.sa_session):
                 roles.append(role)
         # User will see all the roles derived from the access roles on the item
         else:
@@ -765,23 +754,9 @@ def create_private_user_role(self, user):
         return self.get_private_user_role(user)
 
     def get_private_user_role(self, user, auto_create=False):
-        stmt = (
-            select(Role)
-            .where(
-                and_(
-                    UserRoleAssociation.user_id == user.id,
-                    Role.id == UserRoleAssociation.role_id,
-                    Role.type == Role.types.PRIVATE,
-                )
-            )
-            .distinct()
-        )
-        role = self.sa_session.execute(stmt).scalar_one_or_none()
-        if not role:
-            if auto_create:
-                return self.create_private_user_role(user)
-            else:
-                return None
+        role = get_private_user_role(user, self.sa_session)
+        if not role and auto_create:
+            role = self.create_private_user_role(user)
         return role
 
     def get_role(self, name, type=None):
@@ -1695,3 +1670,27 @@ def _walk_action_roles(permissions, query_action):
                 yield action, roles
         elif action == query_action.action and roles:
             yield action, roles
+
+
+def get_npns_roles(session):
+    """
+    non-private, non-sharing roles
+    """
+    stmt = (
+        select(Role)
+        .where(and_(Role.deleted == false(), Role.type != Role.types.PRIVATE, Role.type != Role.types.SHARING))
+        .order_by(Role.name)
+    )
+    return session.scalars(stmt)
+
+
+def get_private_user_role(user, session):
+    stmt = select(Role).where(
+        and_(
+            UserRoleAssociation.user_id == user.id,
+            Role.id == UserRoleAssociation.role_id,
+            Role.type == Role.types.PRIVATE,
+        )
+    )
+    .distinct()
+    return session.execute(stmt).scalar_one_or_none()

lib/galaxy/webapps/galaxy/services/users.py

@@ -4,13 +4,6 @@
     Union,
 )
 
-from sqlalchemy import (
-    false,
-    or_,
-    select,
-    true,
-)
-
 import galaxy.managers.base as managers_base
 from galaxy import (
     exceptions as glx_exceptions,
@@ -22,6 +15,7 @@
     ProvidesUserContext,
 )
 from galaxy.managers.users import (
+    get_users_for_index,
     UserDeserializer,
     UserManager,
     UserSerializer,
@@ -205,31 +199,11 @@ def get_index(
         f_name: Optional[str],
         f_any: Optional[str],
     ) -> List[Union[UserModel, LimitedUserModel]]:
-        rval = []
-        stmt = select(User)
-
-        if f_email and (trans.user_is_admin or trans.app.config.expose_user_email):
-            stmt = stmt.filter(User.email.like(f"%{f_email}%"))
-
-        if f_name and (trans.user_is_admin or trans.app.config.expose_user_name):
-            stmt = stmt.filter(User.username.like(f"%{f_name}%"))
-
-        if f_any:
-            if trans.user_is_admin:
-                stmt = stmt.filter(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
-            else:
-                if trans.app.config.expose_user_email and trans.app.config.expose_user_name:
-                    stmt = stmt.filter(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
-                elif trans.app.config.expose_user_email:
-                    stmt = stmt.filter(User.email.like(f"%{f_any}%"))
-                elif trans.app.config.expose_user_name:
-                    stmt = stmt.filter(User.username.like(f"%{f_any}%"))
-
+        # check for early return conditions
         if deleted:
-            # only admins can see deleted users
             if not trans.user_is_admin:
+                # only admins can see deleted users
                 return []
-            stmt = stmt.filter(User.deleted == true())
         else:
             # special case: user can see only their own user
             # special case2: if the galaxy admin has specified that other user email/names are
@@ -244,8 +218,19 @@ def get_index(
                     return [item]
                 else:
                     return []
-            stmt = stmt.filter(User.deleted == false())
-        for user in trans.sa_session.scalars(stmt).all():
+
+        users = get_users_for_index(
+            trans.sa_session,
+            deleted,
+            f_email,
+            f_name,
+            f_any,
+            trans.user_is_admin,
+            trans.app.config.expose_user_email,
+            trans.app.config.expose_user_name,
+        )
+        rval = []
+        for user in users:
             item = user.to_dict()
             # If NOT configured to expose_email, do not expose email UNLESS the user is self, or
             # the user is an admin

test/unit/data/data_access/__init__.py

@@ -0,0 +1,20 @@
+from collections import namedtuple
+
+PRIVATE_OBJECT_STORE_ID = "my_private_data"
+
+MockTransaction = namedtuple("MockTransaction", "user")
+
+
+class MockObjectStore:
+
+    def is_private(self, object):
+        if object.object_store_id == PRIVATE_OBJECT_STORE_ID:
+            return True
+        else:
+            return False
+
+
+def verify_items(items1, length, items2=None):
+    assert len(items1) == length
+    if items2:
+        assert set(items2) == set(items1)