Merge pull request #15987 from jmchilton/badge_refactor

Refactor badge parsing/serialization/typing for reuse.

client/src/schema/schema.ts

@@ -1494,24 +1494,20 @@ export interface components {
              * @enum {string}
              */
             source: "admin" | "galaxy";
-            /**
-             * Type
-             * @enum {string}
-             */
+            /** Type */
             type:
-                | "faster"
-                | "slower"
-                | "short_term"
-                | "cloud"
-                | "backed_up"
-                | "not_backed_up"
-                | "more_secure"
-                | "less_secure"
-                | "more_stable"
-                | "less_stable"
-                | "quota"
-                | "no_quota"
-                | "restricted";
+                | (
+                      | "faster"
+                      | "slower"
+                      | "short_term"
+                      | "backed_up"
+                      | "not_backed_up"
+                      | "more_secure"
+                      | "less_secure"
+                      | "more_stable"
+                      | "less_stable"
+                  )
+                | ("cloud" | "quota" | "no_quota" | "restricted");
         };
         /**
          * BasicRoleModel

lib/galaxy/objectstore/__init__.py

@@ -18,17 +18,12 @@ from typing import (
     List,
     NamedTuple,
     Optional,
-    Set,
     Tuple,
     Type,
 )
 
 import yaml
 from pydantic import BaseModel
-from typing_extensions import (
-    Literal,
-    TypedDict,
-)
 
 from galaxy.exceptions import (
     ObjectInvalid,
@@ -47,6 +42,12 @@ from galaxy.util.path import (
     safe_relpath,
 )
 from galaxy.util.sleeper import Sleeper
+from .badges import (
+    BadgeDict,
+    read_badges,
+    serialize_badges,
+    StoredBadgeDict,
+)
 
 NO_SESSION_ERROR_MESSAGE = (
     "Attempted to 'create' object store entity in configuration with no database session present."
@@ -57,44 +58,6 @@ DEFAULT_QUOTA_ENABLED = True  # enable quota tracking in object stores by defaul
 
 log = logging.getLogger(__name__)
 
-BadgeSourceT = Literal["admin", "galaxy"]
-BadgeT = Literal[
-    "faster",
-    "slower",
-    "short_term",
-    "cloud",
-    "backed_up",
-    "not_backed_up",
-    "more_secure",
-    "less_secure",
-    "more_stable",
-    "less_stable",
-    "quota",
-    "no_quota",
-    "restricted",
-]
-
-BADGE_SPECIFICATION = [
-    {"type": "faster", "conflicts": ["slower"]},
-    {"type": "slower", "conflicts": ["faster"]},
-    {"type": "short_term", "conflicts": []},
-    {"type": "cloud", "conflicts": []},
-    {"type": "backed_up", "conflicts": ["not_backed_up"]},
-    {"type": "not_backed_up", "conflicts": ["backed_up"]},
-    {"type": "more_secure", "conflicts": ["less_secure"]},
-    {"type": "less_secure", "conflicts": ["more_secure"]},
-    {"type": "more_stable", "conflicts": ["less_stable"]},
-    {"type": "less_stable", "conflicts": ["more_stable"]},
-]
-KNOWN_BADGE_TYPES = [s["type"] for s in BADGE_SPECIFICATION]
-BADGE_SPECIFCATION_BY_TYPE = {s["type"]: s for s in BADGE_SPECIFICATION}
-
-
-class BadgeDict(TypedDict):
-    type: BadgeT
-    message: Optional[str]
-    source: BadgeSourceT
-
 
 class ObjectStore(metaclass=abc.ABCMeta):
 
@@ -511,6 +474,8 @@ class ConcreteObjectStore(BaseObjectStore):
     persisted, not how a file is routed to a persistence source.
     """
 
+    badges: List[StoredBadgeDict]
+
     def __init__(self, config, config_dict=None, **kwargs):
         """
         :type config: object
@@ -537,31 +502,7 @@ class ConcreteObjectStore(BaseObjectStore):
         quota_config = config_dict.get("quota", {})
         self.quota_source = quota_config.get("source", DEFAULT_QUOTA_SOURCE)
         self.quota_enabled = quota_config.get("enabled", DEFAULT_QUOTA_ENABLED)
-        raw_badges = config_dict.get("badges", [])
-        badges = []
-        badge_types: Set[str] = set()
-        badge_conflicts: Dict[str, str] = {}
-        for badge in raw_badges:
-            # when recovering serialized badges, skip ones that are set by Galaxy
-            badge_source = badge.get("source")
-            if badge_source and badge_source != "admin":
-                continue
-            assert "type" in badge
-            badge_type = badge["type"]
-            if badge_type not in KNOWN_BADGE_TYPES:
-                raise Exception(f"badge_type {badge_type} unimplemented/unknown {badge}")
-            message = badge.get("message", None)
-            badges.append({"type": badge_type, "message": message})
-            badge_types.add(badge_type)
-            if badge_type in badge_conflicts:
-                conflicting_badge_type = badge_conflicts[badge_type]
-                raise Exception(
-                    f"Conflicting badge to [{badge_type}] defined on the object store [{conflicting_badge_type}]."
-                )
-            conflicts = BADGE_SPECIFCATION_BY_TYPE[badge_type]["conflicts"]
-            for conflict in conflicts:
-                badge_conflicts[conflict] = badge_type
-        self.badges = badges
+        self.badges = read_badges(config_dict)
 
     def to_dict(self):
         rval = super().to_dict()
@@ -587,37 +528,11 @@ class ConcreteObjectStore(BaseObjectStore):
         )
 
     def _get_concrete_store_badges(self, obj) -> List[BadgeDict]:
-        badge_dicts: List[BadgeDict] = []
-        for badge in self.badges:
-            badge_dict: BadgeDict = {
-                "source": "admin",
-                "type": badge["type"],
-                "message": badge["message"],
-            }
-            badge_dicts.append(badge_dict)
-
-        quota_badge_dict: BadgeDict
-        if self.galaxy_enable_quotas and self.quota_enabled:
-            quota_badge_dict = {
-                "type": "quota",
-                "message": None,
-                "source": "galaxy",
-            }
-        else:
-            quota_badge_dict = {
-                "type": "no_quota",
-                "message": None,
-                "source": "galaxy",
-            }
-        badge_dicts.append(quota_badge_dict)
-        if self.private:
-            restricted_badge_dict: BadgeDict = {
-                "type": "restricted",
-                "message": None,
-                "source": "galaxy",
-            }
-            badge_dicts.append(restricted_badge_dict)
-        return badge_dicts
+        return serialize_badges(
+            self.badges,
+            self.galaxy_enable_quotas and self.quota_enabled,
+            self.private,
+        )
 
     def _get_concrete_store_name(self, obj):
         return self.name

lib/galaxy/objectstore/badges.py

@@ -0,0 +1,141 @@
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+    Set,
+    Union,
+)
+
+from typing_extensions import (
+    Literal,
+    TypedDict,
+)
+
+BadgeSourceT = Literal["admin", "galaxy"]
+# Badges that can be explicitly set by admins.
+AdminBadgeT = Literal[
+    "faster",
+    "slower",
+    "short_term",
+    "backed_up",
+    "not_backed_up",
+    "more_secure",
+    "less_secure",
+    "more_stable",
+    "less_stable",
+]
+
+# All badges - so AdminBadgeT plus Galaxy specifiable badges.
+BadgeT = Union[
+    AdminBadgeT,
+    Literal[
+        "cloud",
+        "quota",
+        "no_quota",
+        "restricted",
+    ],
+]
+
+
+class BadgeSpecDict(TypedDict):
+    """Describe badges that can be set by Galaxy admins."""
+
+    type: AdminBadgeT
+    conflicts: List[AdminBadgeT]
+
+
+BADGE_SPECIFICATION: List[BadgeSpecDict] = [
+    {"type": "faster", "conflicts": ["slower"]},
+    {"type": "slower", "conflicts": ["faster"]},
+    {"type": "short_term", "conflicts": []},
+    {"type": "backed_up", "conflicts": ["not_backed_up"]},
+    {"type": "not_backed_up", "conflicts": ["backed_up"]},
+    {"type": "more_secure", "conflicts": ["less_secure"]},
+    {"type": "less_secure", "conflicts": ["more_secure"]},
+    {"type": "more_stable", "conflicts": ["less_stable"]},
+    {"type": "less_stable", "conflicts": ["more_stable"]},
+]
+
+KNOWN_BADGE_TYPES: List[AdminBadgeT] = [s["type"] for s in BADGE_SPECIFICATION]
+BADGE_SPECIFICATION_BY_TYPE: Dict[AdminBadgeT, BadgeSpecDict] = {s["type"]: s for s in BADGE_SPECIFICATION}
+
+
+class BadgeDict(TypedDict):
+    type: BadgeT
+    message: Optional[str]
+    source: BadgeSourceT
+
+
+class StoredBadgeDict(TypedDict):
+    type: AdminBadgeT
+    message: Optional[str]
+
+
+def read_badges(config_dict: Dict[str, Any]) -> List[StoredBadgeDict]:
+    raw_badges = config_dict.get("badges", [])
+    badges: List[StoredBadgeDict] = []
+    badge_types: Set[str] = set()
+    badge_conflicts: Dict[str, str] = {}
+    for badge in raw_badges:
+        # when recovering serialized badges, skip ones that are set by Galaxy
+        badge_source = badge.get("source")
+        if badge_source and badge_source != "admin":
+            continue
+
+        assert "type" in badge
+        badge_type = badge["type"]
+        if badge_type not in KNOWN_BADGE_TYPES:
+            raise Exception(f"badge_type {badge_type} unimplemented/unknown {badge}")
+        message = badge.get("message", None)
+        badges.append({"type": badge_type, "message": message})
+        badge_types.add(badge_type)
+        if badge_type in badge_conflicts:
+            conflicting_badge_type = badge_conflicts[badge_type]
+            raise Exception(
+                f"Conflicting badge to [{badge_type}] defined on the object store [{conflicting_badge_type}]."
+            )
+        conflicts = BADGE_SPECIFICATION_BY_TYPE[badge_type]["conflicts"]
+        for conflict in conflicts:
+            badge_conflicts[conflict] = badge_type
+    return badges
+
+
+def serialize_badges(stored_badges: List[StoredBadgeDict], quota_enabled: bool, private: bool) -> List[BadgeDict]:
+    """Produce blended, unified list of badges for target object store entity.
+
+    Combine more free form admin information stored about badges with Galaxy tracked
+    information (quota and access restriction information) to produce a unified list
+    of badges to be consumed via the API.
+    """
+    badge_dicts: List[BadgeDict] = []
+    for badge in stored_badges:
+        badge_dict: BadgeDict = {
+            "source": "admin",
+            "type": badge["type"],
+            "message": badge["message"],
+        }
+        badge_dicts.append(badge_dict)
+
+    quota_badge_dict: BadgeDict
+    if quota_enabled:
+        quota_badge_dict = {
+            "type": "quota",
+            "message": None,
+            "source": "galaxy",
+        }
+    else:
+        quota_badge_dict = {
+            "type": "no_quota",
+            "message": None,
+            "source": "galaxy",
+        }
+    badge_dicts.append(quota_badge_dict)
+    if private:
+        restricted_badge_dict: BadgeDict = {
+            "type": "restricted",
+            "message": None,
+            "source": "galaxy",
+        }
+        badge_dicts.append(restricted_badge_dict)
+    return badge_dicts

test/unit/objectstore/test_badges.py

@@ -0,0 +1,17 @@
+from galaxy.objectstore.badges import read_badges
+
+
+def test_read_config_ignores_galaxy_set_badges():
+    config = {"badges": [{"source": "galaxy", "type": "restricted", "message": "private_storage"}]}
+    res = read_badges(config)
+    assert len(res) == 0
+
+
+def test_exception_on_unknown_badge_type():
+    config = {"badges": [{"type": "cool_custom", "message": "private_storage"}]}
+    exc = None
+    try:
+        read_badges(config)
+    except Exception as e:
+        exc = e
+    assert exc is not None