API Reference¶

This page contains the full API reference for kubetest.

Client
- TestClient
Objects
- ApiObject
- ClusterRoleBinding
- ConfigMap
- Container
- Deployment
- Namespace
- Node
- Pod
- RoleBinding
- Secret
- Service
Conditions
- Policy
- Condition
- Helpers

Client ¶

The test client for managing Kubernetes resources within test cases.

An instance of the TestClient defined in this module is automatically created for each test case that uses the kube fixture. The kube fixture provides the TestClient instance to the test case.

TestClient ¶

New in version 0.0.1.

class kubetest.client.TestClient(namespace: str)[source]

Test client for managing Kubernetes resources for a test case.

The namespace for the TestClient will be automatically generated and provided to the TestClient during the test setup process.

Parameters:	namespace – The namespace associated with the test client. Each test case will have its own namespace assigned.

create(obj: kubetest.objects.api_object.ApiObject) → None[source]

Create the provided ApiObject on the Kubernetes cluster.

If the object does not already have a namespace assigned to it, the client’s generated test case namespace will be used.

Parameters:	obj – A kubetest API Object wrapper.

delete(obj: kubetest.objects.api_object.ApiObject, options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → None[source]

Delete the provided ApiObject from the Kubernetes cluster.

If the object does not already have a namespace assigned to it, the client’s generated test case namespace will be used.

Parameters:	obj – A kubetest API Object wrapper. options – Additional options for deleting the resource from the cluster.

static refresh(obj: kubetest.objects.api_object.ApiObject) → None[source]

Refresh the underlying Kubernetes resource status and state.

Parameters:	obj – A kubetest API Object wrapper.

static load_clusterrolebinding(path: str, name: Optional[str] = None) → kubetest.objects.clusterrolebinding.ClusterRoleBinding[source]

Load a manifest YAML into a ClusterRoleBinding object.

Parameters:

path – The path to the ClusterRoleBinding manifest.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The ClusterRoleBinding for the specified manifest.

load_configmap(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.configmap.ConfigMap[source]

Load a manifest YAML into a ConfigMap object.

By default, this will augment the ConfigMap object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the ConfigMap manifest.
set_namespace – Enable/disable the automatic augmentation of the ConfigMap namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The ConfigMap for the specified manifest.

load_daemonset(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.daemonset.DaemonSet[source]

Load a manifest YAML into a DaemonSet object.

By default, this will augment the DaemonSet object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the DaemonSet manifest.
set_namespace – Enable/disable the automatic augmentation of the DaemonSet namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The DaemonSet for the specified manifest.

load_deployment(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.deployment.Deployment[source]

Load a manifest YAML into a Deployment object.

By default, this will augment the Deployment object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the Deployment manifest.
set_namespace – Enable/disable the automatic augmentation of the Deployment namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The Deployment for the specified manifest.

load_pod(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.pod.Pod[source]

Load a manifest YAML into a Pod object.

By default, this will augment the Pod object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the Pod manifest.
set_namespace – Enable/disable the automatic augmentation of the Pod namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The Pod for the specified manifest.

load_rolebinding(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.rolebinding.RoleBinding[source]

Load a manifest YAML into a RoleBinding object.

By default, this will augment the RoleBinding object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the RoleBinding manifest.
set_namespace – Enable/disable the automatic augmentation of the RoleBinding namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The RoleBinding for the specified manifest.

load_secret(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.secret.Secret[source]

Load a manifest YAML into a Secret object.

By default, this will augment the Secret object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the Secret manifest.
set_namespace – Enable/disable the automatic augmentation of the Secret namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The Secret for the specified manifest.

load_service(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.service.Service[source]

Load a manifest YAML into a Service object.

By default, this will augment the Service object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the Service manifest.
set_namespace – Enable/disable the automatic augmentation of the Service namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The Service for the specified manifest.

load_persistentvolumeclaim(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.persistentvolumeclaim.PersistentVolumeClaim[source]

Load a manifest YAML into a PersistentVolumeClaim object.

By default, this will augment the PersistentVolumeClaim object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:	path (str) – The path to the PersistentVolumeClaim manifest. set_namespace (bool) – Enable/disable the automatic augmentation of the PersistentVolumeClaim namespace. name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.
Returns:	The PersistentVolumeClaim for the specified manifest.
Return type:	objects.PersistentVolumeClaim

load_ingress(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.ingress.Ingress[source]

Load a manifest YAML into a Ingress object.

By default, this will augment the Ingress object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:	path (str) – The path to the Ingress manifest. set_namespace (bool) – Enable/disable the automatic augmentation of the Ingress namespace. name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.
Returns:	The ingress for the specified manifest.
Return type:	objects.Ingress

load_replicaset(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.replicaset.ReplicaSet[source]

Load a manifest YAML into a ReplicaSet object.

By default, this will augment the ReplicaSet object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the ReplicaSet manifest.
set_namespace – Enable/disable the automatic augmentation of the ReplicaSet namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The ReplicaSet for the specified manifest.

load_statefulset(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.statefulset.StatefulSet[source]

Load a manifest YAML into a StatefulSet object.

By default, this will augment the StatefulSet object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the StatefulSet manifest.
set_namespace – Enable/disable the automatic augmentation of the StatefulSet namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The StatefulSet for the specified manifest.

load_serviceaccount(path: str, set_namespace: bool = True, name: Optional[str] = None) → kubetest.objects.serviceaccount.ServiceAccount[source]

Load a manifest YAML into a ServiceAccount object.

By default, this will augment the ServiceAccount object with the generated test case namespace. This behavior can be disabled with the set_namespace flag.

Parameters:

path – The path to the ServiceAccount manifest.
set_namespace – Enable/disable the automatic augmentation of the ServiceAccount namespace.
name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.

Returns:

The ServiceAccount for the specified manifest.

get_configmaps(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.configmap.ConfigMap][source]

Get ConfigMaps from the cluster.

Parameters:

namespace – The namespace to get the ConfigMaps from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of ConfigMaps to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of ConfigMaps to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the ConfigMap name and the value is the ConfigMap itself.

get_daemonsets(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.daemonset.DaemonSet][source]

Get DaemonSets from the cluster.

Parameters:

namespace – The namespace to get the DaemonSets from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of DaemonSets to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of DaemonSets to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the DaemonSet name and the value is the DaemonSet itself.

get_deployments(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.deployment.Deployment][source]

Get Deployments from the cluster.

Parameters:

namespace – The namespace to get the Deployments from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of Deployments to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of Deployments to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the Deployment name and the value is the Deployment itself.

get_endpoints(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.endpoints.Endpoints][source]

Get Endpoints from the cluster.

Parameters:

namespace – The namespace to get the Endpoints from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of Endpoints to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of Endpoints to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the Endpoint name and the value is the Endpoint itself.

get_events(fields: Dict[str, str] = None, labels: Dict[str, str] = None, all_namespaces: bool = False) → Dict[str, kubetest.objects.event.Event][source]

Get the latest Events that occurred in the cluster.

Parameters:

fields – A dictionary of fields used to restrict the returned collection of Events to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of Events to only those which match these label selectors. By default, no restricting is done.
all_namespaces – If True, get the events across all namespaces.

Returns:

A dictionary where the key is the Event name and the value is the Event itself.

get_namespaces(fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.namespace.Namespace][source]

Get Namespaces from the cluster.

Parameters:	fields – A dictionary of fields used to restrict the returned collection of Namespaces to only those which match these field selectors. By default, no restricting is done. labels – A dictionary of labels used to restrict the returned collection of Namespaces to only those which match these label selectors. By default, no restricting is done.
Returns:	A dictionary where the key is the Namespace name and the value is the Namespace itself.

static get_nodes(fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.node.Node][source]

Get the Nodes that make up the cluster.

Parameters:	fields – A dictionary of fields used to restrict the returned collection of Nodes to only those which match these field selectors. By default, no restricting is done. labels – A dictionary of labels used to restrict the returned collection of Nodes to only those which match these label selectors. By default, no restricting is done.
Returns:	A dictionary where the key is the Node name and the value is the Node itself.

get_pods(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.pod.Pod][source]

Get Pods from the cluster.

Parameters:

namespace – The namespace to get the Pods from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of Pods to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of Pods to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the Pod name and the value is the Pod itself.

get_secrets(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.secret.Secret][source]

Get Secrets from the cluster.

Parameters:

namespace – The namespace to get the Secrets from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of Secrets to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of Secrets to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the Secret name and the value is the Secret itself.

get_services(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.service.Service][source]

Get Services under the test case namespace.

Parameters:

namespace – The namespace to get the Services from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of Services to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of Services to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the Service name and the value is the Service itself.

get_persistentvolumeclaims(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.persistentvolumeclaim.PersistentVolumeClaim][source]

Get PersistentVolumeClaims from the cluster.

Parameters:

namespace – The namespace to get the PersistentVolumeClaim from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of PersistentVolumeClaim to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of PersistentVolumeClaim to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the PersistentVolumeClaim name and the value is the PersistentVolumeClaim itself.

get_ingresses(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.ingress.Ingress][source]

Get Ingresses from the cluster.

Parameters:

namespace – The namespace to get the Ingress from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of Ingress to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of Ingress to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the Ingress name and the value is the Ingress itself.

get_replicasets(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.replicaset.ReplicaSet][source]

Get ReplicaSets from the cluster.

Parameters:

namespace – The namespace to get the ReplicaSets from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of ReplicaSets to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of ReplicaSets to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the ReplicaSet name and the value is the ReplicaSet itself.

get_statefulsets(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.statefulset.StatefulSet][source]

Get StatefulSets from the cluster.

Parameters:

namespace – The namespace to get the StatefulSets from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of StatefulSets to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of StatefulSets to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the StatefulSet name and the value is the StatefulSet itself.

get_serviceaccounts(namespace: str = None, fields: Dict[str, str] = None, labels: Dict[str, str] = None) → Dict[str, kubetest.objects.serviceaccount.ServiceAccount][source]

Get ServiceAccounts from the cluster.

Parameters:

namespace – The namespace to get the ServiceAccount from. If not specified, it will use the auto-generated test case namespace by default.
fields – A dictionary of fields used to restrict the returned collection of ServiceAccount to only those which match these field selectors. By default, no restricting is done.
labels – A dictionary of labels used to restrict the returned collection of ServiceAccount to only those which match these label selectors. By default, no restricting is done.

Returns:

A dictionary where the key is the ServiceAccount name and the value is the ServiceAccount itself.

static wait_for_conditions(*args, timeout: int = None, interval: Union[float, int] = 1, policy: kubetest.condition.Policy = <Policy.ONCE: 1>, fail_on_api_error: bool = True) → None[source]

Wait for all of the provided Conditions to be met.

All Conditions must be met for this to unblock. If no Conditions are provided, this method will do nothing.

Parameters:

*args – Conditions to check.
timeout – The maximum time to wait, in seconds, for the provided Conditions to be met. If all of the Conditions are not met within the given timeout, this will raise a TimeoutError. By default, there is no timeout so this will wait indefinitely.
interval – The time, in seconds, to sleep before re-evaluating the conditions. Default: 1s
policy – The condition checking policy that defines the checking behavior. Default: ONCE
fail_on_api_error – Fail the condition checks if a Kubernetes API error is incurred. An API error can be raised for a number of reasons, including a Pod being restarted and temporarily unavailable. Disabling this will cause those errors to be ignored, allowing the check to continue until timeout or resolution. (default: True).

Raises:

TimeoutError – The Conditions were not met within the specified timeout period.
ValueError – Not all arguments are a Condition.

wait_for_ready_nodes(count: int, timeout: int = None, interval: Union[int, float] = 1) → None[source]

Wait until there are at least count number of nodes available in the cluster.

Notes

This should only be used for clusters that auto-scale the nodes. This will not create/delete nodes on its own.

Parameters:	count – The number of nodes to wait for. timeout – The maximum time to wait, in seconds. interval – The time, in seconds, to sleep before re-checking the number of nodes.

wait_for_registered(timeout: int = None, interval: Union[int, float] = 1) → None[source]

Wait for all of the pre-registered objects to be ready on the cluster.

An object is pre-registered with the test client if it is specified to the test via the applymanifests pytest marker. The marker will load the manifest and add the object to the cluster, and register it with the test client. This method waits until all such loaded manifest objects are in the ready state simultaneously.

Parameters:	timeout – The maximum time to wait, in seconds. interval – The time, in seconds, to sleep before re-checking the ready state for pre-registered objects.

static wait_until_created(obj: kubetest.objects.api_object.ApiObject, timeout: int = None, interval: Union[int, float] = 1) → None[source]

Wait until the specified object has been created.

Here, creation is judged on whether or not refreshing the object (e.g. getting it) returns an object (created) or an error (not yet created).

Parameters:	obj – The ApiObject to wait on. timeout – The maximum time to wait, in seconds. interval – The time, in seconds, to sleep before re-checking the created state of the object.

Objects ¶

Kubetest wrappers around Kubernetes API Objects.

ApiObject ¶

New in version 0.0.1.

class kubetest.objects.ApiObject(api_object)[source]¶

ApiObject is the base class for many of the kubetest objects which wrap Kubernetes API objects.

This base class provides common functionality and common object properties for all API wrappers. It also defines the following abstract methods which all subclasses must implement:

create: create the resource on the cluster

delete: remove the resource from the cluster

refresh: refresh the underlying object model

is_ready: check if the object is in the ready state

Parameters:	api_object – The underlying Kubernetes API object.
Variables:	obj – The underlying Kubernetes API object.

obj_type = None¶: The default Kubernetes API object type for the class. Each subclass should define its own obj_type.

api_clients = None¶: A mapping of all the supported api clients for the API object type. Various resources can have multiple versions, e.g. “apps/v1”, “apps/v1beta1”, etc. The preferred version for each resource type should be defined under the “preferred” key. The preferred API client will be used when the apiVersion is not specified for the resource.

version¶: The API version of the Kubernetes object (obj.apiVersion`).

name¶: The name of the Kubernetes object (obj.metadata.name).

namespace¶: The namespace of the Kubernetes object (obj.metadata.namespace).

api_client¶

The API client for the Kubernetes object. This is determined by the apiVersion of the object configuration.

Raises:	`ValueError` – The API version is not supported.

classmethod preferred_client()[source]¶

The preferred API client for the Kubernetes object. This is defined in the api_clients class member dict for each object.

Raises:	`ValueError` – No preferred client is defined for the object.

wait_until_ready(timeout: int = None, interval: Union[int, float] = 1, fail_on_api_error: bool = False) → None[source]¶

Wait until the resource is in the ready state.

Parameters:

timeout – The maximum time to wait, in seconds, for the resource to reach the ready state. If unspecified, this will wait indefinitely. If specified and the timeout is met or exceeded, a TimeoutError will be raised.
interval – The time, in seconds, to wait before re-checking if the object is ready.
fail_on_api_error – Fail if an API error is raised. An API error can be raised for a number of reasons, such as ‘resource not found’, which could be the case when a resource is just being started or restarted. When waiting for readiness we generally do not want to fail on these conditions.

Raises:

TimeoutError – The specified timeout was exceeded.

wait_until_deleted(timeout: int = None, interval: Union[int, float] = 1) → None[source]¶

Wait until the resource is deleted from the cluster.

Parameters:	timeout – The maximum time to wait, in seconds, for the resource to be deleted from the cluster. If unspecified, this will wait indefinitely. If specified and the timeout is met or exceeded, a TimeoutError will be raised. interval – The time, in seconds, to wait before re-checking if the object has been deleted.
Raises:	`TimeoutError` – The specified timeout was exceeded.

classmethod load(path: str, name: Optional[str] = None) → kubetest.objects.api_object.ApiObject[source]¶

Load the Kubernetes resource from file.

Generally, this is used to load the Kubernetes manifest files and parse them into their appropriate API Object type.

Parameters:	path – The path to the YAML config file (manifest) containing the configuration for the resource. name – The name of the resource to load. If the manifest file contains a single object definition for the type being loaded, it is not necessary to specify the name. If the manifest has multiple definitions containing the same type, a name is required to differentiate between them. If no name is specified in such case, an error is raised.
Returns:	The API object wrapper corresponding to the configuration loaded from manifest YAML file.
Raises:	`ValueError` – Multiple objects of the desired type were found in the manifest file and no name was specified to differentiate between `them.`

create(namespace: str = None) → None[source]¶

Create the underlying Kubernetes resource in the cluster under the given namespace.

Parameters:	namespace – The namespace to create the resource under. If no namespace is provided, it will use the instance’s namespace member, which is set when the object is created via the kubetest client.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the underlying Kubernetes resource from the cluster.

This method expects the resource to have been loaded or otherwise assigned a namespace already. If it has not, the namespace will need to be set manually.

Parameters:	options – Options for resource deletion.

refresh() → None[source]¶: Refresh the local state (obj) of the underlying Kubernetes resource.

is_ready() → bool[source]¶

Check if the resource is in the ready state.

It is up to the wrapper subclass to define what “ready” means for that particular resource.

Returns:	True if in the ready state; False otherwise.

ClusterRoleBinding ¶

New in version 0.0.1.

class kubetest.objects.ClusterRoleBinding(api_object)[source]¶

Kubetest wrapper around a Kubernetes ClusterRoleBinding API Object.

The actual kubernetes.client.V1ClusterRoleBinding instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the ClusterRoleBinding.

obj_type¶: alias of kubernetes.client.models.v1_cluster_role_binding.V1ClusterRoleBinding

create(namespace: str = None) → None[source]¶

Create the ClusterRoleBinding under the given namespace.

Parameters:	namespace – This argument is ignored for ClusterRoleBindings.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the ClusterRoleBinding.

This method expects the ClusterRoleBinding to have been loaded or otherwise assigned a namespace already. If it has not, the namespace will need to be set manually.

Parameters:	options – Options for ClusterRoleBinding deletion.
Returns:	The status of the delete operation.

refresh() → None[source]¶: Refresh the underlying Kubernetes ClusterRoleBinding resource.

is_ready() → bool[source]¶

Check if the ClusterRoleBinding is in the ready state.

ClusterRoleBindings do not have a “status” field to check, so we will measure their readiness status by whether or not they exist on the cluster.

Returns:	True if in the ready state; False otherwise.

ConfigMap ¶

New in version 0.0.1.

class kubetest.objects.ConfigMap(api_object)[source]¶

Kubetest wrapper around a Kubernetes ConfigMap API Object.

The actual kubernetes.client.V1ConfigMap instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the ConfigMap.

obj_type¶: alias of kubernetes.client.models.v1_config_map.V1ConfigMap

create(namespace: str = None) → None[source]¶

Create the ConfigMap under the given namespace.

Parameters:	namespace – The namespace to create the ConfigMap under. If the ConfigMap was loaded via the kubetest client, the namespace will already be set, so it is not needed here. Otherwise, the namespace will need to be provided.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the ConfigMap.

This method expects the ConfigMap to have been loaded or otherwise assigned a namespace already. If it has not, the namespace will need to be set manually.

Parameters:	options – Options for ConfigMap deletion.
Returns:	The status of the delete operation.

refresh() → None[source]¶: Refresh the underlying Kubernetes ConfigMap resource.

is_ready() → bool[source]¶

Check if the ConfigMap is in the ready state.

ConfigMaps do not have a “status” field to check, so we will measure their readiness status by whether or not they exist on the cluster.

Returns:	True if in the ready state; False otherwise.

Container ¶

New in version 0.0.1.

class kubetest.objects.Container(api_object, pod)[source]¶

Kubetest wrapper around a Kubernetes Container API Object.

The actual kubernetes.client.V1Container instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the Container.

This wrapper does NOT subclass the objects.ApiObject like other object wrappers because it is not intended to be created or managed from manifest file. It is merely meant to wrap the Container spec for a Pod to make Container-targeted actions easier.

get_restart_count() → int[source]¶

Get the number of times the Container has been restarted.

Returns:	The number of times the Container has been restarted.

get_logs() → str[source]¶

Get all the logs for the Container.

Returns:	The Container logs.

search_logs(*keyword) → bool[source]¶

Search for keywords/phrases in the Container’s logs.

Parameters:	*keyword – Keywords to search for within the logs.
Returns:	True if found; False otherwise.

Deployment ¶

New in version 0.0.1.

class kubetest.objects.Deployment(*args, **kwargs)[source]¶

Kubetest wrapper around a Kubernetes Deployment API Object.

The actual kubernetes.client.V1Deployment instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the Deployment.

obj_type¶: alias of kubernetes.client.models.v1_deployment.V1Deployment

create(namespace: str = None) → None[source]¶

Create the Deployment under the given namespace.

Parameters:	namespace – The namespace to create the Deployment under. If the Deployment was loaded via the kubetest client, the namespace will already be set, so it is not needed here. Otherwise, the namespace will need to be provided.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the Deployment.

This method expects the Deployment to have been loaded or otherwise assigned a namespace already. If it has not, the namespace will need to be set manually.

Parameters:	options – Options for Deployment deletion.
Returns:	The status of the delete operation.

refresh() → None[source]¶: Refresh the underlying Kubernetes Deployment resource.

is_ready() → bool[source]¶

Check if the Deployment is in the ready state.

Returns:	True if in the ready state; False otherwise.

status() → kubernetes.client.models.v1_deployment_status.V1DeploymentStatus[source]¶

Get the status of the Deployment.

Returns:	The status of the Deployment.

get_pods() → List[kubetest.objects.pod.Pod][source]¶

Get the pods for the Deployment.

Returns:	A list of pods that belong to the deployment.

Namespace ¶

New in version 0.0.1.

class kubetest.objects.Namespace(api_object)[source]¶

Kubetest wrapper around a Kubernetes Namespace API Object.

The actual kubernetes.client.V1Namespace instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the Namespace.

obj_type¶: alias of kubernetes.client.models.v1_namespace.V1Namespace

classmethod new(name: str) → kubetest.objects.namespace.Namespace[source]¶

Create a new Namespace with object backing.

Parameters:	name – The name of the new Namespace.
Returns:	A new Namespace instance.

create(name: str = None) → None[source]¶

Create the Namespace under the given name.

Parameters:	name – The name to create the Namespace under. If the name is not provided, it will be assumed to already be in the underlying object spec. If it is not, namespace operations will fail.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the Namespace.

Parameters:	options – Options for Namespace deletion.
Returns:	The status of the delete operation.

refresh() → None[source]¶: Refresh the underlying Kubernetes Namespace resource.

is_ready() → bool[source]¶

Check if the Namespace is in the ready state.

Returns:	True if in the ready state; False otherwise.

Node ¶

New in version 0.0.1.

class kubetest.objects.Node(api_object)[source]¶

Kubetest wrapper around a Kubernetes Node API Object.

The actual kubernetes.client.V1Node instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the Node.

This wrapper does NOT subclass the objects.ApiObject like other object wrappers because it is not intended to be created or managed from manifest file. It is merely meant to wrap the Node spec to make Node-based interactions easier

refresh() → None[source]¶: Refresh the underlying Kubernetes Node resource.

status() → kubernetes.client.models.v1_node_status.V1NodeStatus[source]¶

Get the status of the Node.

Returns:	The status of the Node.

is_ready() → bool[source]¶

Check whether the Node is in the ready state.

Returns:	True if in the ready state; False otherwise.

Pod ¶

New in version 0.0.1.

class kubetest.objects.Pod(api_object)[source]¶

Kubetest wrapper around a Kubernetes Pod API Object.

The actual kubernetes.client.V1Pod instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the Pod.

obj_type¶: alias of kubernetes.client.models.v1_pod.V1Pod

create(namespace: str = None) → None[source]¶

Create the Pod under the given namespace.

Parameters:	namespace – The namespace to create the Pod under. If the Pod was loaded via the kubetest client, the namespace will already be set, so it is not needed here. Otherwise, the namespace will need to be provided.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the Pod.

This method expects the Pod to have been loaded or otherwise assigned a namespace already. If it has not, the namespace will need to be set manually.

Parameters:	options – Options for Pod deletion.
Returns:	The status of the delete operation.

refresh() → None[source]¶: Refresh the underlying Kubernetes Pod resource.

is_ready() → bool[source]¶

Check if the Pod is in the ready state.

Returns:	True if in the ready state; False otherwise.

status() → kubernetes.client.models.v1_pod_status.V1PodStatus[source]¶

Get the status of the Pod.

Returns:	The status of the Pod.

get_containers() → List[kubetest.objects.container.Container][source]¶

Get the Pod’s containers.

Returns:	A list of containers that belong to the Pod.

get_container(name: str) → Optional[kubetest.objects.container.Container][source]¶

Get a container in the Pod by name.

Parameters:	name (str) – The name of the Container.
Returns:	The Pod’s Container with the matching name. If no container with the given name is found, `None` is returned.
Return type:	Container

get_restart_count() → int[source]¶

Get the total number of Container restarts for the Pod.

Returns:	The total number of Container restarts.

http_proxy_get(path: str, query_params: Dict[str, str] = None) → kubetest.response.Response[source]¶

Issue a GET request to a proxy for the Pod.

Notes

This function does not use the kubernetes connect_get_namespaced_pod_proxy_with_path function because there appears to be lack of support for custom query parameters (as of the kubernetes==9.0.0 package version). To bypass this, parts of the core functionality from the aforementioned function are used here with the modification of allowing user-defined query parameters to be passed along.

Parameters:	path – The URI path for the request. query_params – Any query parameters for the request.
Returns:	The response data.

http_proxy_post(path: str, query_params: Dict[str, str] = None, data=None) → kubetest.response.Response[source]¶

Issue a POST request to a proxy for the Pod.

Notes

This function does not use the kubernetes connect_post_namespaced_pod_proxy_with_path function because there appears to be lack of support for custom query parameters (as of the kubernetes==9.0.0 package version). To bypass this, parts of the core functionality from the aforementioned function are used here with the modification of allowing user-defined query parameters to be passed along.

Parameters:	path – The URI path for the request. query_params – Any query parameters for the request. data – The data to POST.
Returns:	The response data.

containers_started() → bool[source]¶

Check if the Pod’s Containers have all started.

Returns:	True if all Containers have started; False otherwise.

wait_until_containers_start(timeout: int = None) → None[source]¶

Wait until all containers in the Pod have started.

This will wait for the images to be pulled and for the containers to be created and started. This will unblock once all Pod containers have been started.

This is different than waiting until ready, since a container may not be ready immediately after it has been started.

Parameters:	timeout – The maximum time to wait, in seconds, for the Pod’s containers to be started. If unspecified, this will wait indefinitely. If specified and the timeout is met or exceeded, a TimeoutError will be raised.
Raises:	`TimeoutError` – The specified timeout was exceeded.

RoleBinding ¶

New in version 0.0.1.

class kubetest.objects.RoleBinding(api_object)[source]¶

Kubetest wrapper around a Kubernetes RoleBinding API Object.

The actual kubernetes.client.V1RoleBinding instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the RoleBinding.

obj_type¶: alias of kubernetes.client.models.v1_role_binding.V1RoleBinding

create(namespace: str = None) → None[source]¶

Create the RoleBinding under the given namespace.

Parameters:	namespace – The namespace to create the RoleBinding under. If the RoleBinding was loaded via the kubetest client, the namespace will already be set, so it is not needed here. Otherwise, the namespace will need to be provided.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the RoleBinding.

This method expects the RoleBinding to have been loaded or otherwise assigned a namespace already. If it has not, the namespace will need to be set manually.

Parameters:	options – Options for RoleBinding deletion.
Returns:	The status of the delete operation.

refresh() → None[source]¶: Refresh the underlying Kubernetes RoleBinding resource.

is_ready() → bool[source]¶

Check if the RoleBinding is in the ready state.

RoleBindings do not have a “status” field to check, so we will measure their readiness status by whether or not they exist on the cluster.

Returns:	True if in the ready state; False otherwise.

Secret ¶

New in version 0.0.1.

class kubetest.objects.Secret(api_object)[source]¶

Kubetest wrapper around a Kubernetes Secret API Object.

The actual kubernetes.client.V1Secret instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the Secret.

obj_type¶: alias of kubernetes.client.models.v1_secret.V1Secret

create(namespace: str = None) → None[source]¶

Create the Secret under the given namespace.

Parameters:	namespace – The namespace to create the Secret under. If the Secret was loaded via the kubetest client, the namespace will already be set, so it is not needed here. Otherwise, the namespace will need to be provided.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the Secret.

This method expects the Secret to have been loaded or otherwise assigned a namespace already. If it has not, the namespace will need to be set manually.

Parameters:	options – Options for Secret deletion.
Returns:	The status of the delete operation.

refresh() → None[source]¶: Refresh the underlying Kubernetes Secret resource.

is_ready() → bool[source]¶

Check if the Secret is in the ready state.

Secrets do not have a “status” field to check, so we will measure their readiness status by whether or not they exist on the cluster.

Returns:	True if in the ready state; False otherwise.

Service ¶

New in version 0.0.1.

class kubetest.objects.Service(api_object)[source]¶

Kubetest wrapper around a Kubernetes Service API Object.

The actual kubernetes.client.V1Service instance that this wraps can be accessed via the obj instance member.

This wrapper provides some convenient functionality around the API Object and provides some state management for the Service.

obj_type¶: alias of kubernetes.client.models.v1_service.V1Service

create(namespace: str = None) → None[source]¶

Create the Service under the given namespace.

Parameters:	namespace – The namespace to create the Service under. If the Service was loaded via the kubetest client, the namespace will already be set, so it is not needed here. Otherwise, the namespace will need to be provided.

delete(options: kubernetes.client.models.v1_delete_options.V1DeleteOptions = None) → kubernetes.client.models.v1_status.V1Status[source]¶

Delete the Service.

This method expects the Service to have been loaded or otherwise assigned a namespace already. If it has not, the namespace will need to be set manually.

Parameters:	options – Options for Service deletion.
Returns:	The status of the delete operation.

refresh() → None[source]¶: Refresh the underlying Kubernetes Service resource.

is_ready() → bool[source]¶

Check if the Service is in the ready state.

The readiness state is not clearly available from the Service status, so to see whether or not the Service is ready this will check whether the endpoints of the Service are ready.

This comes with the caveat that in order for a Service to have endpoints, there needs to be some backend hooked up to it. If there is no backend, the Service will never have endpoints, so this will never resolve to True.

Returns:	True if in the ready state; False otherwise.

status() → kubernetes.client.models.v1_service_status.V1ServiceStatus[source]¶

Get the status of the Service.

Returns:	The status of the Service.

get_endpoints() → List[kubernetes.client.models.v1_endpoints.V1Endpoints][source]¶

Get the endpoints for the Service.

This can be useful for checking internal IP addresses used in containers, e.g. for container auto-discovery.

Returns:	A list of endpoints associated with the Service.

proxy_http_get(path: str, **kwargs) → tuple[source]¶

Issue a GET request to proxy of a Service.

Parameters:	path – The URI path for the request. kwargs – Keyword arguments for the proxy_http_get function.
Returns:	The response data

proxy_http_post(path: str, **kwargs) → tuple[source]¶

Issue a POST request to proxy of a Service.

Parameters:	path – The URI path for the request. kwargs – Keyword arguments for the proxy_http_post function.
Returns:	The response data

Conditions ¶

Define test conditions for kubetest.

Policy ¶

New in version 0.0.1.

class kubetest.condition.Policy[source]¶

Condition checking policies.

A Policy defines the behavior of how Conditions are checked.

ONCE: A condition only needs to be met once and the check will consider it met regardless of the state of any other conditions that may be checked alongside it. This is the default behavior.

SIMULTANEOUS: A condition needs to be met simultaneous to all other conditions that are being checked alongside it for the check to be successful.

Condition ¶

New in version 0.0.1.

class kubetest.condition.Condition(name: str, fn: Callable, *args, **kwargs)[source]¶

A Condition is a convenience wrapper around a function and its arguments which allows the function to be called at a later time.

The function is called in the check method, which resolves the result to a boolean value, thus the condition function should return a boolean or something that ultimately resolves to a Truthy or Falsey value.

Parameters:	name – The name of the condition to make it easier to identify. fn – The condition function that will be checked. args – Any arguments for the condition function. *kwargs – Any keyword arguments for the condition function.
Variables:	name (str) – The name of the Condition. fn (callable) – The condition function that will be checked. args (tuple) – Arguments for the checking function. kwargs (dict) – Keyword arguments for the checking function. last_check (bool) – Holds the state of the last condition check.
Raises:	`ValueError` – The given `fn` is not callable.

check() → bool[source]¶

Check that the condition was met.

Returns:	True if the condition was met; False otherwise.

Helpers ¶

kubetest.condition.check_all(*args) → bool[source]¶

Check all the given Conditions.

Parameters:	*args – The Conditions to check.
Returns:	True if all checks pass; False otherwise.

kubetest.condition.check_and_sort(*args) → Tuple[List[kubetest.condition.Condition], List[kubetest.condition.Condition]][source]¶

Check all the given Conditions and sort them into ‘met’ and ‘unmet’ buckets.

Parameters:	*args – The Conditions to check.
Returns:	The met and unmet condition buckets (in that order).