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 inobjects_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 typeobj_type
passobj_condition_fun
or timeout is reached. Ifobjs_namespace
is None, objects are treated as cluster-scope, otherwise as namespace-scope. If optionalfailure_condition_func
is passed, it is executed when objects are refreshed and if it evaluates toTrue
, it throwsObjectStatusError
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 clusterobj_type
(type) — type of the objects to check; they most be derived from APIObjectobj_names
(list of str) — a list of object resource names to check; all the objects must passobj_condition_fun
for this function to end with successobjs_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-scopeobj_condition_func
(callable(T: bool)) — a function that gets one instance of the resource object of typeobj_type
and returns boolean showing whether the object meets the condition or not. The call succeeds only if thisobj_condition_fun
returnsTrue
for every object passed inobj_names
.timeout_sec
(int) — timeout for the callmissing_ok
(bool) — whenTrue
, the function ignores that some objects listed in theobj_names
don't exist in k8s API and waits for them to show up; whenFalse
, 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 returnsTrue
, theObjectStatusError
is raised.
Returns (list)
The list of object resources with all the objects listed in obj_names
included in the list.
Raises
ObjectStatusError
— whenfailure_condition_func
is not None and any of the objects inobj_names
returnedTrue
from this function.TimeoutError
— when timeout is reached.pykube.exceptions.ObjectDoesNotExist
— whenmissing_ok == False
and one of the objects listed inobj_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 clusterobj_type
(type) — type of the objects to check; they should be derived from APIObjectobjects_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