Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 41 additions & 0 deletions src/mas/devops/ocp.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
from kubernetes import client
from kubernetes.stream import stream
from kubernetes.stream.ws_client import ERROR_CHANNEL
from kubernetes.dynamic.resource import ResourceInstance

import yaml

Expand Down Expand Up @@ -376,6 +377,46 @@ def getStorageClasses(dynClient: DynamicClient) -> list:
return storageClasses


def getClusterIssuers(dynClient: DynamicClient) -> list:
"""
Get all ClusterIssuers in the cluster.

Parameters:
dynClient (DynamicClient): OpenShift Dynamic Client

Returns:
list: List of ClusterIssuers resources

Raises:
NotFoundError: If ClusterIssuers cannot be retrieved
"""
clusterIssuerAPI = dynClient.resources.get(api_version="cert-manager.io/v1", kind="ClusterIssuer")
clusterIssuers = clusterIssuerAPI.get().items
return clusterIssuers


def getClusterIssuer(dynClient: DynamicClient, name: str) -> ResourceInstance:
"""
Get a specific ClusterIssuer by name.

Parameters:
dynClient (DynamicClient): OpenShift Dynamic Client
name (str): The name of the ClusterIssuer to retrieve

Returns:
ClusterIssuer: The ClusterIssuer resource, or None if not found

Raises:
NotFoundError: If the ClusterIssuer does not exist (caught and returns None)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We are catching a NotFoundError and returning None, so we shouldn't document that this raises NotFoundError ... the documentation should only list exceptions that can be raised by calling the function, not exceptions handling internally within the function.

Copy link
Contributor Author

@terc1997 terc1997 Feb 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, I agree with that. But there are several example in the code that handle the exceptions like this. Fya:

def crdExists(dynClient: DynamicClient, crdName: str) -> bool:
"""
Check if a Custom Resource Definition (CRD) exists in the cluster.
Parameters:
dynClient (DynamicClient): OpenShift Dynamic Client
crdName (str): The name of the CRD to check (e.g., "suites.core.mas.ibm.com")
Returns:
bool: True if the CRD exists, False otherwise
Raises:
NotFoundError: If the CRD does not exist (caught and returns False)
"""
crdAPI = dynClient.resources.get(api_version="apiextensions.k8s.io/v1", kind="CustomResourceDefinition")
try:
crdAPI.get(name=crdName)
logger.debug(f"CRD does exist: {crdName}")
return True
except NotFoundError:
logger.debug(f"CRD does not exist: {crdName}")
return False

def getStorageClass(dynClient: DynamicClient, name: str) -> dict | None:
"""
Get a specific StorageClass by name.
Parameters:
dynClient (DynamicClient): OpenShift Dynamic Client
name (str): The name of the StorageClass to retrieve
Returns:
StorageClass: The StorageClass resource, or None if not found
Raises:
NotFoundError: If the StorageClass does not exist (caught and returns None)
"""
try:
storageClassAPI = dynClient.resources.get(api_version="storage.k8s.io/v1", kind="StorageClass")
storageclass = storageClassAPI.get(name=name)
return storageclass
except NotFoundError:
return None

I just followed the standard of the code. So, should we change all or keep as is ?

"""
try:
clusterIssuerAPI = dynClient.resources.get(api_version="cert-manager.io/v1", kind="ClusterIssuer")
clusterIssuer = clusterIssuerAPI.get(name=name)
return clusterIssuer
except NotFoundError:
return None


def getStorageClassVolumeBindingMode(dynClient: DynamicClient, storageClassName: str) -> str:
"""
Get the volumeBindingMode for a storage class.
Expand Down