MODULE

PYTEST_HELM_CHARTS.UTILS

Different utilities required over the whole testing lib.

Functions
  • delete_and_wait_for_objects(kube_client, obj_type, objects_to_del, timeout_sec) For each object in objects_to_delete, make an API call to delete it, then wait until the object is gone from k8s API server.</>
  • wait_for_objects_condition(kube_client, obj_type, obj_names, objs_namespace, obj_condition_func, timeout_sec, missing_ok, failure_condition_func) (list) Block until all the kubernetes objects of type obj_type pass obj_condition_fun or timeout is reached. If objs_namespace is None, objects are treated as cluster-scope, otherwise as namespace-scope. If optional failure_condition_func is passed, it is executed when objects are refreshed and if it evaluates to True, it throws ObjectStatusError exception.</>
function

pytest_helm_charts.utils.wait_for_objects_condition(kube_client, obj_type, obj_names, objs_namespace, obj_condition_func, timeout_sec, missing_ok, failure_condition_func=None)

Block until all the kubernetes objects of type obj_type pass obj_condition_fun or timeout is reached. If objs_namespace is None, objects are treated as cluster-scope, otherwise as namespace-scope. If optional failure_condition_func is passed, it is executed when objects are refreshed and if it evaluates to True, it throws ObjectStatusError exception.

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 APIObject
  • obj_names (list of str) a list of object resource names to check; all the objects must pass obj_condition_fun for this function to end with success
  • objs_namespace (str, optional) namespace where all the resources should be present (single namespace for all resources). If 'None', then it is assumed objects passed are cluster-scope
  • 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 objects listed in the obj_names don't exist in k8s API and waits for them to show up; when False, an ObjectNotExist exception is raised.
  • failure_condition_func (callable(T: bool), optional) if not None, then each monitored object is passed to this function. If it returns True, the ObjectStatusError is raised.
Returns (list)

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

Raises
  • ObjectStatusError when failure_condition_func is not None and any of the objects in obj_names returned True from this function.
  • 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.delete_and_wait_for_objects(kube_client, obj_type, objects_to_del, timeout_sec=120)

For each object in objects_to_delete, make an API call to delete it, then wait until the object is gone from k8s API server.

Parameters
  • kube_client (HTTPClient) client to use to connect to the k8s cluster
  • obj_type (type) type of the objects to check; they should be derived from APIObject
  • objects_to_del (iterable) iterable of APIObject to delete. All objects must be of the same specific type.
  • timeout_sec (int, optional) timeout for all the objects in the list to be gone from API server.

Returns: None