pycodestyle fixes: kraken/kubernetes/client.py

Signed-off-by: Sandro Bonazzola <sbonazzo@redhat.com>
2026-08-25 09:27:36 +00:00 · 2022-09-01 12:32:52 -04:00
parent d5615ac470
commit 8a68e1cc9b
1 changed files with 211 additions and 63 deletions
@@ -2,7 +2,7 @@ from kubernetes import client, config, utils, watch
 from kubernetes.dynamic.client import DynamicClient
 from kubernetes.stream import stream
 from kubernetes.client.rest import ApiException
-from kraken.kubernetes.resources import *
+from ..kubernetes.resources import *
 import logging
 import sys
 import re
@@ -39,7 +39,10 @@ def get_host() -> str:


 def get_clusterversion_string() -> str:
-    """Returns clusterversion status text on OpenShift, empty string on other distributions"""
+    """
+    Returns clusterversion status text on OpenShift, empty string
+    on other distributions
+    """
    try:
        cvs = custom_object_client.list_cluster_custom_object(
            "config.openshift.io",
@@ -63,11 +66,16 @@ def list_namespaces(label_selector=None):
    namespaces = []
    try:
        if label_selector:
-            ret = cli.list_namespace(pretty=True, label_selector=label_selector)
+            ret = cli.list_namespace(
+                pretty=True,
+                label_selector=label_selector
+                )
        else:
            ret = cli.list_namespace(pretty=True)
    except ApiException as e:
-        logging.error("Exception when calling CoreV1Api->list_namespaced_pod: %s\n" % e)
+        logging.error(
+            "Exception when calling CoreV1Api->list_namespaced_pod: %s\n" % e
+        )
        raise e
    for namespace in ret.items:
        namespaces.append(namespace.metadata.name)
@@ -80,7 +88,9 @@ def get_namespace_status(namespace_name):
    try:
        ret = cli.read_namespace_status(namespace_name)
    except ApiException as e:
-        logging.error("Exception when calling CoreV1Api->read_namespace_status: %s\n" % e)
+        logging.error(
+            "Exception when calling CoreV1Api->read_namespace_status: %s\n" % e
+        )
    return ret.status.phase


@@ -88,7 +98,9 @@ def delete_namespace(namespace):
    """Deletes a given namespace using kubernetes python client"""
    try:
        api_response = cli.delete_namespace(namespace)
-        logging.debug("Namespace deleted. status='%s'" % str(api_response.status))
+        logging.debug(
+            "Namespace deleted. status='%s'" % str(api_response.status)
+        )
        return api_response
    except Exception as e:
        logging.error(
@@ -114,7 +126,10 @@ def check_namespaces(namespaces, label_selectors=None):
                        break
        invalid_namespaces = regex_namespaces - valid_regex
        if invalid_namespaces:
-            raise Exception("There exists no namespaces matching: %s" % (invalid_namespaces))
+            raise Exception(
+                "There exists no namespaces matching: %s" %
+                (invalid_namespaces)
+            )
        return list(final_namespaces)
    except Exception as e:
        logging.info("%s" % (e))
@@ -161,7 +176,11 @@ def list_pods(namespace, label_selector=None):
    pods = []
    try:
        if label_selector:
-            ret = cli.list_namespaced_pod(namespace, pretty=True, label_selector=label_selector)
+            ret = cli.list_namespaced_pod(
+                namespace,
+                pretty=True,
+                label_selector=label_selector
+            )
        else:
            ret = cli.list_namespaced_pod(namespace, pretty=True)
    except ApiException as e:
@@ -179,7 +198,10 @@ def list_pods(namespace, label_selector=None):
 def get_all_pods(label_selector=None):
    pods = []
    if label_selector:
-        ret = cli.list_pod_for_all_namespaces(pretty=True, label_selector=label_selector)
+        ret = cli.list_pod_for_all_namespaces(
+            pretty=True,
+            label_selector=label_selector
+        )
    else:
        ret = cli.list_pod_for_all_namespaces(pretty=True)
    for pod in ret.items:
@@ -188,7 +210,13 @@ def get_all_pods(label_selector=None):


 # Execute command in pod
-def exec_cmd_in_pod(command, pod_name, namespace, container=None, base_command="bash"):
+def exec_cmd_in_pod(
+    command,
+    pod_name,
+    namespace,
+    container=None,
+    base_command="bash"
+):

    exec_command = [base_command, "-c", command]
    try:
@@ -239,7 +267,10 @@ def create_pod(body, namespace, timeout=120):
        pod_stat = cli.create_namespaced_pod(body=body, namespace=namespace)
        end_time = time.time() + timeout
        while True:
-            pod_stat = cli.read_namespaced_pod(name=body["metadata"]["name"], namespace=namespace)
+            pod_stat = cli.read_namespaced_pod(
+                name=body["metadata"]["name"],
+                namespace=namespace
+            )
            if pod_stat.status.phase == "Running":
                break
            if time.time() > end_time:
@@ -259,7 +290,10 @@ def read_pod(name, namespace="default"):

 def get_pod_log(name, namespace="default"):
    return cli.read_namespaced_pod_log(
-        name=name, namespace=namespace, _return_http_data_only=True, _preload_content=False
+        name=name,
+        namespace=namespace,
+        _return_http_data_only=True,
+        _preload_content=False
    )


@@ -277,7 +311,10 @@ def delete_job(name, namespace="default"):
        api_response = batch_cli.delete_namespaced_job(
            name=name,
            namespace=namespace,
-            body=client.V1DeleteOptions(propagation_policy="Foreground", grace_period_seconds=0),
+            body=client.V1DeleteOptions(
+                propagation_policy="Foreground",
+                grace_period_seconds=0
+            ),
        )
        logging.debug("Job deleted. status='%s'" % str(api_response.status))
        return api_response
@@ -299,7 +336,10 @@ def delete_job(name, namespace="default"):

 def create_job(body, namespace="default"):
    try:
-        api_response = batch_cli.create_namespaced_job(body=body, namespace=namespace)
+        api_response = batch_cli.create_namespaced_job(
+            body=body,
+            namespace=namespace
+        )
        return api_response
    except ApiException as api:
        logging.warn(
@@ -320,7 +360,10 @@ def create_job(body, namespace="default"):

 def get_job_status(name, namespace="default"):
    try:
-        return batch_cli.read_namespaced_job_status(name=name, namespace=namespace)
+        return batch_cli.read_namespaced_job_status(
+            name=name,
+            namespace=namespace
+        )
    except Exception as e:
        logging.error(
            "Exception when calling \
@@ -368,7 +411,11 @@ def monitor_namespace(namespace):
    notready_pods = []
    for pod in pods:
        try:
-            pod_info = cli.read_namespaced_pod_status(pod, namespace, pretty=True)
+            pod_info = cli.read_namespaced_pod_status(
+                pod,
+                namespace,
+                pretty=True
+            )
        except ApiException as e:
            logging.error(
                "Exception when calling \
@@ -377,7 +424,11 @@ def monitor_namespace(namespace):
            )
            raise e
        pod_status = pod_info.status.phase
-        if pod_status != "Running" and pod_status != "Completed" and pod_status != "Succeeded":
+        if (
+            pod_status != "Running" and
+            pod_status != "Completed" and
+            pod_status != "Succeeded"
+        ):
            notready_pods.append(pod)
    if len(notready_pods) != 0:
        status = False
@@ -388,8 +439,15 @@ def monitor_namespace(namespace):

 # Monitor component namespace
 def monitor_component(iteration, component_namespace):
-    watch_component_status, failed_component_pods = monitor_namespace(component_namespace)
-    logging.info("Iteration %s: %s: %s" % (iteration, component_namespace, watch_component_status))
+    watch_component_status, failed_component_pods = \
+        monitor_namespace(component_namespace)
+    logging.info(
+        "Iteration %s: %s: %s" % (
+            iteration,
+            component_namespace,
+            watch_component_status
+        )
+    )
    return watch_component_status, failed_component_pods


@@ -407,7 +465,11 @@ def apply_yaml(path, namespace='default'):
        The object created
    """

-    return utils.create_from_yaml(api_client, yaml_file=path, namespace=namespace)
+    return utils.create_from_yaml(
+        api_client,
+        yaml_file=path,
+        namespace=namespace
+    )


 def get_pod_info(name: str, namespace: str = 'default') -> Pod:
@@ -424,21 +486,36 @@ def get_pod_info(name: str, namespace: str = 'default') -> Pod:
            - Namespace to look for the pod

    Returns:
-        - Data class object of type Pod with the output of the above kubectl command
-        in the given format if the pod exists
+        - Data class object of type Pod with the output of the above
+          kubectl command in the given format if the pod exists
        - Returns None if the pod doesn't exist
    """
    pod_exists = check_if_pod_exists(name=name, namespace=namespace)
    if pod_exists:
-        response = cli.read_namespaced_pod(name=name, namespace=namespace, pretty='true')
+        response = cli.read_namespaced_pod(
+            name=name,
+            namespace=namespace,
+            pretty='true'
+        )
        container_list = []

        # Create a list of containers present in the pod
        for container in response.spec.containers:
            volume_mount_list = []
            for volume_mount in container.volume_mounts:
-                volume_mount_list.append(VolumeMount(name=volume_mount.name, mountPath=volume_mount.mount_path))
-            container_list.append(Container(name=container.name, image=container.image, volumeMounts= volume_mount_list))
+                volume_mount_list.append(
+                    VolumeMount(
+                        name=volume_mount.name,
+                        mountPath=volume_mount.mount_path
+                    )
+                )
+            container_list.append(
+                Container(
+                    name=container.name,
+                    image=container.image,
+                    volumeMounts=volume_mount_list
+                )
+            )

        for i, container in enumerate(response.status.container_statuses):
            container_list[i].ready = container.ready
@@ -447,25 +524,42 @@ def get_pod_info(name: str, namespace: str = 'default') -> Pod:
        volume_list = []
        for volume in response.spec.volumes:
            volume_name = volume.name
-            pvc_name = volume.persistent_volume_claim.claim_name if volume.persistent_volume_claim is not None else None
+            pvc_name = (
+                volume.persistent_volume_claim.claim_name
+                if volume.persistent_volume_claim is not None
+                else None
+            )
            volume_list.append(Volume(name=volume_name, pvcName=pvc_name))

        # Create the Pod data class object
-        pod_info = Pod(name=response.metadata.name, podIP=response.status.pod_ip, namespace=response.metadata.namespace,
-                    containers=container_list, nodeName=response.spec.node_name, volumes = volume_list)
+        pod_info = Pod(
+            name=response.metadata.name,
+            podIP=response.status.pod_ip,
+            namespace=response.metadata.namespace,
+            containers=container_list,
+            nodeName=response.spec.node_name,
+            volumes=volume_list
+        )
        return pod_info
    else:
        logging.error(
-                "Pod '%s' doesn't exist in namespace '%s'" % (str(name), str(namespace))
+                "Pod '%s' doesn't exist in namespace '%s'" % (
+                    str(name),
+                    str(namespace)
+                )
        )
        return None
-    


-def get_litmus_chaos_object(kind: str, name: str, namespace: str) -> LitmusChaosObject:
+def get_litmus_chaos_object(
+        kind: str,
+        name: str,
+        namespace: str
+) -> LitmusChaosObject:
    """
-    Function that returns an object of a custom resource type of the litmus project. Currently, only
-    ChaosEngine and ChaosResult objects are supported.
+    Function that returns an object of a custom resource type of
+    the litmus project. Currently, only ChaosEngine and ChaosResult
+    objects are supported.

    Args:
        kind (string)
@@ -483,29 +577,54 @@ def get_litmus_chaos_object(kind: str, name: str, namespace: str) -> LitmusChaos

    if kind.lower() == 'chaosengine':
        plural = 'chaosengines'
-        response = custom_object_client.get_namespaced_custom_object(group=group, plural=plural,
-                                            version=version, namespace=namespace, name=name)
+        response = custom_object_client.get_namespaced_custom_object(
+            group=group,
+            plural=plural,
+            version=version,
+            namespace=namespace,
+            name=name
+        )
        try:
            engine_status = response['status']['engineStatus']
            exp_status = response['status']['experiments'][0]['status']
-        except:
+        except Exception:
            engine_status = 'Not Initialized'
            exp_status = 'Not Initialized'
-        custom_object = ChaosEngine(kind='ChaosEngine', group=group, namespace=namespace, name=name,
-                                    plural=plural, version=version, engineStatus=engine_status,
-                                    expStatus=exp_status)
+        custom_object = ChaosEngine(
+            kind='ChaosEngine',
+            group=group,
+            namespace=namespace,
+            name=name,
+            plural=plural,
+            version=version,
+            engineStatus=engine_status,
+            expStatus=exp_status
+        )
    elif kind.lower() == 'chaosresult':
        plural = 'chaosresults'
-        response = custom_object_client.get_namespaced_custom_object(group=group, plural=plural,
-                                            version=version, namespace=namespace, name=name)
+        response = custom_object_client.get_namespaced_custom_object(
+            group=group,
+            plural=plural,
+            version=version,
+            namespace=namespace,
+            name=name
+        )
        try:
            verdict = response['status']['experimentStatus']['verdict']
            fail_step = response['status']['experimentStatus']['failStep']
-        except:
+        except Exception:
            verdict = 'N/A'
            fail_step = 'N/A'
-        custom_object = ChaosResult(kind='ChaosResult', group=group, namespace=namespace, name=name,
-                                    plural=plural, version=version, verdict=verdict, failStep=fail_step)
+        custom_object = ChaosResult(
+            kind='ChaosResult',
+            group=group,
+            namespace=namespace,
+            name=name,
+            plural=plural,
+            version=version,
+            verdict=verdict,
+            failStep=fail_step
+        )
    else:
        logging.error("Invalid litmus chaos custom resource name")
        custom_object = None
@@ -514,7 +633,8 @@ def get_litmus_chaos_object(kind: str, name: str, namespace: str) -> LitmusChaos

 def check_if_namespace_exists(name: str) -> bool:
    """
-    Function that checks if a namespace exists by parsing through the list of projects
+    Function that checks if a namespace exists by parsing through
+    the list of projects.
    Args:
        name (string)
            - Namespace name
@@ -523,7 +643,10 @@ def check_if_namespace_exists(name: str) -> bool:
        Boolean value indicating whether the namespace exists or not
    """

-    v1_projects = dyn_client.resources.get(api_version='project.openshift.io/v1', kind='Project')
+    v1_projects = dyn_client.resources.get(
+        api_version='project.openshift.io/v1',
+        kind='Project'
+    )
    project_list = v1_projects.get()
    return True if name in str(project_list) else False

@@ -554,7 +677,8 @@ def check_if_pod_exists(name: str, namespace: str) -> bool:

 def check_if_pvc_exists(name: str, namespace: str) -> bool:
    """
-    Function that checks if a namespace exists by parsing through the list of projects
+    Function that checks if a namespace exists by parsing through
+    the list of projects.
    Args:
        name (string)
            - PVC name
@@ -563,12 +687,14 @@ def check_if_pvc_exists(name: str, namespace: str) -> bool:
            - Namespace name

    Returns:
-        Boolean value indicating whether the Persistent Volume Claim exists or not
+        Boolean value indicating whether the Persistent Volume Claim
+        exists or not.
    """
-    
    namespace_exists = check_if_namespace_exists(namespace)
    if namespace_exists:
-        response = cli.list_namespaced_persistent_volume_claim(namespace=namespace)
+        response = cli.list_namespaced_persistent_volume_claim(
+            namespace=namespace
+        )
        pvc_list = [pvc.metadata.name for pvc in response.items]
        if name in pvc_list:
            return True
@@ -585,19 +711,23 @@ def get_pvc_info(name: str, namespace: str) -> PVC:
    Args:
        name (string)
            - Name of the persistent volume claim
-        
+
        namespace (string)
            - Namespace where the persistent volume claim is present
-    
+
    Returns:
-        - A PVC data class containing the name, capacity, volume name, namespace
-        and associated pod names of the PVC if the PVC exists
-        - Returns None if the PVC doesn't exist 
+        - A PVC data class containing the name, capacity, volume name,
+          namespace and associated pod names of the PVC if the PVC exists
+        - Returns None if the PVC doesn't exist
    """

    pvc_exists = check_if_pvc_exists(name=name, namespace=namespace)
    if pvc_exists:
-        pvc_info_response = cli.read_namespaced_persistent_volume_claim(name=name, namespace=namespace, pretty=True)
+        pvc_info_response = cli.read_namespaced_persistent_volume_claim(
+            name=name,
+            namespace=namespace,
+            pretty=True
+        )
        pod_list_response = cli.list_namespaced_pod(namespace=namespace)

        capacity = pvc_info_response.status.capacity['storage']
@@ -607,18 +737,30 @@ def get_pvc_info(name: str, namespace: str) -> PVC:
        pvc_pod_list = []
        for pod in pod_list_response.items:
            for volume in pod.spec.volumes:
-                if volume.persistent_volume_claim is not None and volume.persistent_volume_claim.claim_name == name:
+                if (
+                    volume.persistent_volume_claim is not None
+                    and volume.persistent_volume_claim.claim_name == name
+                ):
                    pvc_pod_list.append(pod.metadata.name)

-        pvc_info = PVC(name=name, capacity=capacity, volumeName=volume_name, podNames=pvc_pod_list, namespace=namespace)
+        pvc_info = PVC(
+            name=name,
+            capacity=capacity,
+            volumeName=volume_name,
+            podNames=pvc_pod_list,
+            namespace=namespace
+        )
        return pvc_info
    else:
        logging.error(
-                "PVC '%s' doesn't exist in namespace '%s'" % (str(name), str(namespace))
+            "PVC '%s' doesn't exist in namespace '%s'" % (
+                str(name),
+                str(namespace)
+            )
        )
        return None
-    
-    
+
+
 # Find the node kraken is deployed on
 # Set global kraken node to not delete
 def find_kraken_node():
@@ -651,13 +793,19 @@ def watch_node_status(node, status, timeout, resource_version):
        timeout_seconds=timeout,
        resource_version=f"{resource_version}"
    ):
-        conditions = [status for status in event["object"].status.conditions if status.type == "Ready"]
+        conditions = [
+            status
+            for status in event["object"].status.conditions
+            if status.type == "Ready"
+        ]
        if conditions[0].status == status:
            watch_resource.stop()
            break
        else:
            count -= 1
-            logging.info("Status of node " + node + ": " + str(conditions[0].status))
+            logging.info(
+                "Status of node " + node + ": " + str(conditions[0].status)
+            )
        if not count:
            watch_resource.stop()