[patch] moved functions to mas.py

Author: Sanjay Prabhakar

src/mas/devops/mas.py

@@ -12,6 +12,7 @@
 import re
 import yaml
 from os import path
+from time import sleep
 from types import SimpleNamespace
 from kubernetes.dynamic.resource import ResourceInstance
 from openshift.dynamic import DynamicClient
@@ -194,3 +195,62 @@ def updateIBMEntitlementKey(dynClient: DynamicClient, namespace: str, icrUsernam
 
     secret = secretsAPI.apply(body=secret, namespace=namespace)
     return secret
+
+def waitForPVC(dynClient: DynamicClient, namespace: str, pvcName: str) -> bool:
+    pvcAPI = dynClient.resources.get(api_version="v1", kind="PersistentVolumeClaim")
+    maxRetries = 60
+    foundReadyPVC = False
+    retries = 0
+    while not foundReadyPVC and retries < maxRetries:
+        retries += 1
+        try:
+            pvc = pvcAPI.get(name=pvcName, namespace=namespace)
+            if pvc.status.phase == "Bound":
+                foundReadyPVC = True
+            else:
+                logger.debug("Waiting 5s for PVC {pvcName} to be ready before checking again ...")
+                sleep(5)
+        except NotFoundError:
+            logger.debug("Waiting 5s for PVC {pvcName} to be created before checking again ...")
+            sleep(5)
+    return foundReadyPVC
+
+
+def patchPendingPVC(dynClient: DynamicClient, namespace: str, pvcName: str, storageClassName: str = None) -> bool:
+    pvcAPI = dynClient.resources.get(api_version="v1", kind="PersistentVolumeClaim")
+    try:
+        pvc = pvcAPI.get(name=pvcName, namespace=namespace)
+        if pvc.status.phase == "Pending" and pvc.spec.storageClassName is None:
+            if storageClassName is not None and storageClassName(dynClient, name=storageClassName) is not None:
+                pvc.spec.storageClassName = storageClassName
+            else:
+                defaultStorageClasses = getDefaultStorageClasses(dynClient)
+                if defaultStorageClasses.provider is not None:
+                    pvc.spec.storageClassName = defaultStorageClasses.rwo
+                else:
+                    logger.error("Unable to set storageClassName in PVC {pvcName}.")
+                    return False
+
+            pvcAPI.patch(body=pvc, namespace=namespace)
+
+            maxRetries = 60
+            foundReadyPVC = False
+            retries = 0
+            while not foundReadyPVC and retries < maxRetries:
+                retries += 1
+                try:
+                    patchedPVC = pvcAPI.get(name=pvcName, namespace=namespace)
+                    if patchedPVC.status.phase == "Bound":
+                        foundReadyPVC = True
+                    else:
+                        logger.debug("Waiting 5s for PVC {pvcName} to be bound before checking again ...")
+                        sleep(5)
+                except NotFoundError:
+                    logger.error("The patched PVC {pvcName} does not exist.")
+                    return False
+            
+            return foundReadyPVC
+
+    except NotFoundError:
+        logger.error("PVC {pvcName} does not exist")
+        return False

src/mas/devops/ocp.py

@@ -20,8 +20,6 @@
 from kubernetes.stream import stream
 from kubernetes.stream.ws_client import ERROR_CHANNEL
 
-from .mas import getDefaultStorageClasses
-
 import yaml
 
 logger = logging.getLogger(__name__)
@@ -200,65 +198,6 @@ def crdExists(dynClient: DynamicClient, crdName: str) -> bool:
         logger.debug(f"CRD does not exist: {crdName}")
         return False
 
-def waitForPVC(dynClient: DynamicClient, namespace: str, pvcName: str) -> bool:
-    pvcAPI = dynClient.resources.get(api_version="v1", kind="PersistentVolumeClaim")
-    maxRetries = 60
-    foundReadyPVC = False
-    retries = 0
-    while not foundReadyPVC and retries < maxRetries:
-        retries += 1
-        try:
-            pvc = pvcAPI.get(name=pvcName, namespace=namespace)
-            if pvc.status.phase == "Bound":
-                foundReadyPVC = True
-            else:
-                logger.debug("Waiting 5s for PVC {pvcName} to be ready before checking again ...")
-                sleep(5)
-        except NotFoundError:
-            logger.debug("Waiting 5s for PVC {pvcName} to be created before checking again ...")
-            sleep(5)
-    return foundReadyPVC
-
-
-def patchPendingPVC(dynClient: DynamicClient, namespace: str, pvcName: str, storageClassName: str = None) -> bool:
-    pvcAPI = dynClient.resources.get(api_version="v1", kind="PersistentVolumeClaim")
-    try:
-        pvc = pvcAPI.get(name=pvcName, namespace=namespace)
-        if pvc.status.phase == "Pending" and pvc.spec.storageClassName is None:
-            if storageClassName is not None and storageClassName(dynClient, name=storageClassName) is not None:
-                pvc.spec.storageClassName = storageClassName
-            else:
-                defaultStorageClasses = getDefaultStorageClasses(dynClient)
-                if defaultStorageClasses.provider is not None:
-                    pvc.spec.storageClassName = defaultStorageClasses.rwo
-                else:
-                    logger.error("Unable to set storageClassName in PVC {pvcName}.")
-                    return False
-
-            pvcAPI.patch(body=pvc, namespace=namespace)
-
-            maxRetries = 60
-            foundReadyPVC = False
-            retries = 0
-            while not foundReadyPVC and retries < maxRetries:
-                retries += 1
-                try:
-                    patchedPVC = pvcAPI.get(name=pvcName, namespace=namespace)
-                    if patchedPVC.status.phase == "Bound":
-                        foundReadyPVC = True
-                    else:
-                        logger.debug("Waiting 5s for PVC {pvcName} to be bound before checking again ...")
-                        sleep(5)
-                except NotFoundError:
-                    logger.error("The patched PVC {pvcName} does not exist.")
-                    return False
-            
-            return foundReadyPVC
-
-    except NotFoundError:
-        logger.error("PVC {pvcName} does not exist")
-        return False
-
 
 # Assisted by WCA@IBM
 # Latest GenAI contribution: ibm/granite-8b-code-instruct

src/mas/devops/tekton.py

@@ -22,7 +22,8 @@
 
 from jinja2 import Environment, FileSystemLoader
 
-from .ocp import getConsoleURL, waitForCRD, waitForDeployment, waitForPVC, patchPendingPVC, crdExists
+from .ocp import getConsoleURL, waitForCRD, waitForDeployment, crdExists
+from .mas import waitForPVC, patchPendingPVC
 
 logger = logging.getLogger(__name__)