MODULE

PYTEST_HELM_CHARTS.K8S.JOB

</>
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.</>
  • make_job_object(kube_client, name_prefix, namespace, command, image, restart_policy, backoff_limit, extra_metadata, extra_spec) (Job) — Creates a Job object according to the parameters.</>
  • 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.</>
function

pytest_helm_charts.k8s.job.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.k8s.job.make_job_object(kube_client, name_prefix, namespace, command, image='quay.io/giantswarm/busybox:1.32.0', restart_policy='OnFailure', backoff_limit=6, extra_metadata=None, extra_spec=None)

</>

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)
  • extra_metadata (dict, optional) — optional dict that will be merged with the 'metadata:' section of the object
  • extra_spec (dict, optional) — optional dict that will be merged with the 'spec:' section of the object
Returns (Job)

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

function

pytest_helm_charts.k8s.job.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