MODULE

PYTEST_HELM_CHARTS.UTILS

Different utilities required over the whole testing lib.

Functions
  • create_job_and_run_to_completion(kube_client, namespace, job, timeout_sec, missing_ok) (Job) Creates Job object in k8s and blocks until it is completed.</>
  • ensure_namespace_exists(kube_client, namespace_name) (Namespace) Checks if the Namespace exists and creates it if it doesn't</>
  • make_job_object(kube_client, name_prefix, namespace, command, image, restart_policy, backoff_limit) (Job) Creates a Job object according to the parameters.</>
  • wait_for_daemon_sets_to_run(kube_client, daemon_set_names, daemon_sets_namespace, timeout_sec, missing_ok) (list of DaemonSet) Block until all the DaemonSets are running or timeout is reached.</>
  • wait_for_deployments_to_run(kube_client, deployment_names, deployments_namespace, timeout_sec, missing_ok) (list of Deployment) Block until all the Deployments are running or timeout is reached.</>
  • wait_for_jobs_to_complete(kube_client, job_names, jobs_namespace, timeout_sec, missing_ok) (list of Job) Block until all the Jobs are complete or timeout is reached.</>
  • wait_for_namespaced_objects_condition(kube_client, obj_type, obj_names, objs_namespace, obj_condition_func, timeout_sec, missing_ok) (list) Block until all the namespaced kubernetes objects of type obj_type pass obj_condition_fun or timeout is reached.</>
  • wait_for_stateful_sets_to_run(kube_client, stateful_set_names, stateful_sets_namespace, timeout_sec, missing_ok) (list of StatefulSet) Block until all the StatefulSets are running or timeout is reached.</>
function

pytest_helm_charts.utils.wait_for_namespaced_objects_condition(kube_client, obj_type, obj_names, objs_namespace, obj_condition_func, timeout_sec, missing_ok)

Block until all the namespaced kubernetes objects of type obj_type pass obj_condition_fun or timeout is reached.

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • obj_type (type) type of the objects to check; they most be derived from NamespacedAPIObject
  • obj_names (list of str) a list of object resource names to check; all of the objects must be pass obj_condition_fun for this function to succeed
  • objs_namespace (str) namespace where all the resources are created (single namespace for all resources)
  • obj_condition_func (callable(T: bool)) a function that gets one instance of the resource object of type obj_type and returns boolean showing whether the object meets the condition or not. The call succeeds only if this obj_condition_fun returns True for every object passed in obj_names.
  • timeout_sec (int) timeout for the call
  • missing_ok (bool) when True, the function ignores that some of the objects listed in the obj_names don't exist in k8s API and waits for them to show up; when False, an ObjectNotFound exception is raised.
Returns (list)

The list of namespaced object resources with all the objects listed in obj_names included.

Raises
  • TimeoutError when timeout is reached.
  • pykube.exceptions.ObjectDoesNotExist when missing_ok == False and one of the objects listed in obj_names can't be found in k8s API
function

pytest_helm_charts.utils.wait_for_jobs_to_complete(kube_client, job_names, jobs_namespace, timeout_sec, missing_ok=True)

Block until all the Jobs are complete or timeout is reached.

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • job_names (list of str) a list of Job names to check
  • jobs_namespace (str) namespace where all the Jobs are created (single namespace for all resources)
  • timeout_sec (int) timeout for the call
  • missing_ok (bool, optional) when True, the function ignores that some of the objects listed in the job_names don't exist in k8s API and waits for them to show up; when False, an ObjectNotFound exception is raised.
Returns (list of Job)

The list of Job resources with all the objects listed in job_names included.

Raises
  • TimeoutError when timeout is reached.
  • pykube.exceptions.ObjectDoesNotExist when missing_ok == False and one of the objects listed in job_names can't be found in k8s API
function

pytest_helm_charts.utils.wait_for_deployments_to_run(kube_client, deployment_names, deployments_namespace, timeout_sec, missing_ok=True)

Block until all the Deployments are running or timeout is reached.

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • deployment_names (list of str) a list of Deployment names to check
  • deployments_namespace (str) namespace where all the Deployments are created (single namespace for all resources)
  • timeout_sec (int) timeout for the call
  • missing_ok (bool, optional) when True, the function ignores that some of the objects listed in the deployment_names don't exist in k8s API and waits for them to show up; when False, an ObjectNotFound exception is raised.
Returns (list of Deployment)

The list of Deployment resources with all the objects listed in deployment_names included.

Raises
  • TimeoutError when timeout is reached.
  • pykube.exceptions.ObjectDoesNotExist when missing_ok == False and one of the objects listed in deployment_names can't be found in k8s API
function

pytest_helm_charts.utils.wait_for_stateful_sets_to_run(kube_client, stateful_set_names, stateful_sets_namespace, timeout_sec, missing_ok=False)

Block until all the StatefulSets are running or timeout is reached.

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • stateful_set_names (list of str) a list of StatefulSet names to check
  • stateful_sets_namespace (str) namespace where all the StatefulSets are created (single namespace for all resources)
  • timeout_sec (int) timeout for the call
  • missing_ok (bool, optional) when True, the function ignores that some of the objects listed in the stateful_set_names don't exist in k8s API and waits for them to show up; when False, an ObjectNotFound exception is raised.
Returns (list of StatefulSet)

The list of StatefulSet resources with all the objects listed in stateful_set_names included.

Raises
  • TimeoutError when timeout is reached.
  • pykube.exceptions.ObjectDoesNotExist when missing_ok == False and one of the objects listed in stateful_set_names can't be found in k8s API
function

pytest_helm_charts.utils.wait_for_daemon_sets_to_run(kube_client, daemon_set_names, daemon_sets_namespace, timeout_sec, missing_ok=False)

Block until all the DaemonSets are running or timeout is reached.

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • daemon_set_names (list of str) a list of DaemonSet names to check
  • daemon_sets_namespace (str) namespace where all the DaemonSets are created (single namespace for all resources)
  • timeout_sec (int) timeout for the call
  • missing_ok (bool, optional) when True, the function ignores that some of the objects listed in the daemon_set_names don't exist in k8s API and waits for them to show up; when False, an ObjectNotFound exception is raised.
Returns (list of DaemonSet)

The list of DaemonSet resources with all the objects listed in daemon_set_names included.

Raises
  • TimeoutError when timeout is reached.
  • pykube.exceptions.ObjectDoesNotExist when missing_ok == False and one of the objects listed in daemon_set_names can't be found in k8s API
function

pytest_helm_charts.utils.ensure_namespace_exists(kube_client, namespace_name)

Checks if the Namespace exists and creates it if it doesn't

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • namespace_name (str) a name of the Namespace to ensure
Returns (Namespace)

Namespace resource object

function

pytest_helm_charts.utils.make_job_object(kube_client, name_prefix, namespace, command, image='quay.io/giantswarm/busybox:1.32.0', restart_policy='OnFailure', backoff_limit=6)

Creates a Job object according to the parameters.

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • name_prefix (str) metadata.generateName Job's object property that is used to name Pods
  • namespace (str) Namespace to create the Job in
  • command (list of str) command to run inside the Job pod
  • image (str, optional) container image to use
  • restart_policy (str, optional) Job's restart policy (as in k8s API)
  • backoff_limit (int, optional) Job's backoff limit (as in k8s API)
Returns (Job)

Job object. The Job is not sent for creation to API server.

function

pytest_helm_charts.utils.create_job_and_run_to_completion(kube_client, namespace, job, timeout_sec=60, missing_ok=False)

Creates Job object in k8s and blocks until it is completed.

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • namespace (str) Namespace to create the Job in
  • job (Job) Job object to run and check.
  • timeout_sec (int, optional) timeout for the call
  • missing_ok (bool, optional) when True, the function ignores that the Job doesn't yet exist in k8s API and waits for it to show up; when False, an ObjectNotFound exception is raised.
Returns (Job)

The Job object with refreshed state.

Raises
  • TimeoutError when timeout is reached.
  • pykube.exceptions.ObjectDoesNotExist when missing_ok == False and the job can't be found in k8s API