[minor] Add functions to retrieve a subscription and mas channel

src/mas/devops/mas.py

@@ -19,6 +19,7 @@
 from jinja2 import Environment, FileSystemLoader
 
 from .ocp import getStorageClasses
+from .olm import getSubscription
 
 logger = logging.getLogger(__name__)
 
@@ -141,6 +142,20 @@ def verifyMasInstance(dynClient: DynamicClient, instanceId: str) -> bool:
         return False
 
 
+def getMasChannel(dynClient: DynamicClient, instanceId: str) -> str:
+    """
+    Get the MAS channel from the subscription
+    """
+    try:
+        masSubscription = getSubscription(dynClient, f"mas-{instanceId}-core", "ibm-mas")
+        return masSubscription.items[0]["spec"]["channel"]
+    except NotFoundError:
+        return False
+    except UnauthorizedError:
+        logger.error("Error: Unable to verify MAS instance due to failed authorization: {e}")
+        return False
+
+
 def updateIBMEntitlementKey(dynClient: DynamicClient, namespace: str, icrUsername: str, icrPassword: str, artifactoryUsername: str = None, artifactoryPassword: str = None, secretName: str = "ibm-entitlement") -> ResourceInstance:
     if secretName is None:
         secretName = "ibm-entitlement"

src/mas/devops/olm.py

@@ -57,6 +57,24 @@ def ensureOperatorGroupExists(dynClient: DynamicClient, env: Environment, namesp
         logger.debug(f"An OperatorGroup already exists in namespace {namespace}")
 
 
+def getSubscription(dynClient: DynamicClient, namespace: str, packageName: str):
+    try:
+        labelSelector = f"operators.coreos.com/{packageName}.{namespace}"
+        logger.debug(f"Get Subscription for {packageName} in {namespace}")
+        subscriptionsAPI = dynClient.resources.get(
+            api_version="operators.coreos.com/v1alpha1", kind="Subscription"
+        )
+        subscription = subscriptionsAPI.get(
+            label_selector=labelSelector, namespace=namespace
+        )
+        if subscription.items == 0:
+            raise NotFoundError
+    except NotFoundError:
+        logger.info(f"No matching Subscription found for {packageName} in {namespace}")
+        subscription = None
+    return subscription
+
+
 def applySubscription(dynClient: DynamicClient, namespace: str, packageName: str, packageChannel: str = None, catalogSource: str = None, catalogSourceNamespace: str = "openshift-marketplace", config: dict = None):
     """
     Usage: