kubetest package¶

Submodules¶

kubetest.client module¶

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.

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

Bases: object

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.

kubetest.manager module¶

Kubetest manager for test client instances and namespace management.

class kubetest.manager.ObjectManager[source]¶

Bases: object

ObjectManager is a convenience class used to manage Kubernetes API objects that are registered with a test case.

The core usage of the ObjectManager is to sort each of the registered objects into different buckets by type. An “apply order” is also defined here so we can get the bucketed objects in the order that they should be applied onto the cluster.

This manager will only be used for API objects loaded from manifests that are specified by the pytest.mark.applymanifests marker on a test case by default.

ordered_buckets = ['namespace', 'rolebinding', 'clusterrolebinding', 'secret', 'networkpolicy', 'service', 'configmap', 'persistentvolumeclaim', 'ingress', 'daemonset', 'statefulset', 'replicaset', 'deployment', 'pod']¶

add(*args) → None[source]¶

Add API objects to the object manager.

This method will take in any number of ApiObjects and sort them into the correct buckets. It will not check for duplicates. If a non-ApiObject is passed in, an error will be raised.

Parameters:	args – Any subclass of the kubetest ApiObject wrapping a Kubernetes API object.
Raises:	`ValueError` – One or more arguments passed to the function are not ApiObject subclasses.

get_objects_in_apply_order() → Generator[kubetest.objects.api_object.ApiObject, None, None][source]¶

Get all of the managed objects in the order that they should be applied onto the cluster.

Within the buckets themselves, API objects are not sorted. This function only yields the buckets in the correct order.

Each of the buckets corresponds to an ApiObject wrapper that is supported by kubetest. As more ApiObject wrappers are added, the buckets here should be updated to reflect that.

The bucket order in which objects are yielded are:

Namespace
RoleBinding
ClusterRoleBinding
Secret
NetworkPolicy
Service
ConfigMap
PersistentVolumeClaim
Ingress
DaemonSet
StatefulSet
ReplicaSet
Deployment
Pod

Yields:	The kubetest ApiObject wrapper to be created on the cluster.

class kubetest.manager.TestMeta(name: str, node_id: str, namespace_create: bool = True, namespace_name: str = None)[source]¶

Bases: object

TestMeta holds information associated with a single test node.

Parameters:	name – The name of the test. node_id – The id of the test node. namespace_create – Option to create a new namespace. This can be toggled off in the event that a user is using an existing namespace on their test cluster. namespace_name – The name of the namespace to use. If the namespace is to be autogenerated, this can be left as None.

client¶: Get the TestClient for the test case.

namespace¶: Get the Namespace API Object associated with the test case.

setup() → None[source]¶

Setup the cluster state for the test case.

This performs all actions needed in order for the test client to be ready to use by a test case.

teardown() → None[source]¶

Clean up the cluster state for the given test case.

This performs all actions needed in order to clean up the state that was previously set up for the test client in setup.

yield_container_logs(tail_lines: int = None) → Generator[str, None, None][source]¶

Yield the container logs for the test case.

These logs will be printed out if the test was in error to provide more context and make it easier to debug the issue.

Parameters:	tail_lines – The number of container log lines to print.
Yields:	str – Logs for the running containers on the cluster.

register_rolebindings(*rolebindings) → None[source]¶

Register a RoleBinding requirement with the test case.

Parameters:	rolebindings – The RoleBindings that are needed for the test case.

register_clusterrolebindings(*clusterrolebindings) → None[source]¶

Register a ClusterRoleBinding requirement with the test case.

Parameters:	clusterrolebindings – The ClusterRoleBindings that are needed for the test case.

register_objects(api_objects: List[kubetest.objects.api_object.ApiObject])[source]¶

Register the provided objects with the test case.

These objects will be registered to the test client and applied to the namespace on test setup.

Parameters:	api_objects – The wrapped Kubernetes API objects to create on the cluster.

class kubetest.manager.KubetestManager[source]¶

Bases: object

The manager for kubetest state.

The KubetestManager is in charge of providing test clients for the tests that request them and mediating the namespace management corresponding to those test clients.

new_test(node_id: str, test_name: str, namespace_create: bool = True, namespace_name: str = None) → kubetest.manager.TestMeta[source]¶

Create a new TestMeta for a test case.

This will be called by the test setup hook in order to create a new TestMeta for the manager to keep track of.

Parameters:	node_id – The id of the test node. test_name – The name of the test. namespace_create – Option to create a new namespace. This can be toggled off in the event that a user is using an existing namespace on their test cluster. namespace_name – The name of the namespace to use. If the namespace is to be autogenerated, this can be left as None.
Returns:	The newly created TestMeta for the test case.

get_test(node_id: str) → Optional[kubetest.manager.TestMeta][source]¶

Get the test metadata for the specified test node.

Parameters:	node_id – The id of the test node.
Returns:	The test metadata for the given node. If no test metadata is found for the given node, None is returned.

teardown(node_id: str) → None[source]¶

Tear down the test case.

This is effectively a wrapper around the teardown method of the test client. It will also remove the test client from the manager.

Test client teardown will delete the test client’s namespace from the cluster. Deleting a namespace will delete all the things in the namespace (e.g. API objects bound to the namespace).

Parameters:	node_id – The id of the test node.

kubetest.manifest module¶

Utility functions for loading Kubernetes manifest files and constructing the corresponding Kubernetes API models.

kubetest.manifest.render(template: Union[str, TextIO], context: Dict[str, Any]) → Union[str, TextIO][source]¶

Render a manifest template into a YAML document using the module render callable.

Rendering is performed by the module global __render__ callable. The default implementation returns the input template unmodified. To implement a different default rendering behavior, set the __render__ attribute to a RenderCallable object.

Parameters:	template – Then template to render. context – A dictionary of variables available to the template.
Returns:	The rendered content of the manifest template.

class kubetest.manifest.ContextRenderer(renderer: Callable[[Union[str, TextIO], Dict[str, Any]], Union[str, TextIO]] = <function render>, context: Dict[str, Any] = {})[source]¶

Bases: object

ContextRenderer objects manage context state and render manifest templates.

ContextRenderer objects maintain a persistent context state across an arbitrary number of rendering operations. They are useful in accumulating context state during test setup when manifest templates need access to state that was established earlier.

Parameters:	renderer – A callable that renders a templated manifest. Defaults to the module local render function which renders using the __render__ callable. context – A dictionary of runtime values available to the template during rendering. Defaults to an empty dictionary.

context¶: The context variables set on the renderer.

__call__(template: Union[str, TextIO], context: Dict[str, Any]) → Union[str, TextIO][source]¶

Render a manifest template file to a YAML docoment.

Parameters:	template – The template to render. context – A dictionary of template variables available for use during rendering.
Returns:	The rendered content of the manifest template.

kubetest.manifest.load_file(path: str, *, renderer: Callable[[Union[str, TextIO], Dict[str, Any]], Union[str, TextIO]] = <function render>) → List[object][source]¶

Load an individual Kubernetes manifest YAML file.

This file may contain multiple YAML documents. It will attempt to auto-detect the type of each object to load.

Parameters:	path – The fully qualified path to the file. renderer – The callable responsible for rendering the contents of the manifest file to YAML.
Returns:	A list of the Kubernetes API objects for this manifest file.

kubetest.manifest.load_path(path: str, *, renderer: Callable[[Union[str, TextIO], Dict[str, Any]], Union[str, TextIO]] = <function render>) → List[object][source]¶

Load all of the Kubernetes YAML manifest files found in the specified directory path.

Parameters:	path – The path to the directory of manifest files. renderer – The callable responsible for rendering the contents of the manifest files to YAML.
Returns:	A list of all the Kubernetes objects loaded from manifest file.
Raises:	`ValueError` – The provided path is not a directory.

kubetest.manifest.get_type(manifest: Dict[str, Any]) → Optional[object][source]¶

Get the Kubernetes object type from the manifest kind and version.

There is no easy way for determining the internal model that a manifest should use. What this tries to do is use the version info and the kind info to create a potential internal object name for the pair and look that up in the kubernetes package locals.

Parameters:	manifest – The manifest file, loaded into a dictionary.
Returns:	The Kubernetes API object for the manifest. If no Kubernetes API object type can be determined, None is returned.
Raises:	`ValueError` – The manifest dictionary does not have a version or kind specified.

kubetest.manifest.load_type(obj_type, path: str, *, renderer: Callable[[Union[str, TextIO], Dict[str, Any]], Union[str, TextIO]] = <function render>)[source]¶

Load a Kubernetes YAML manifest file for the specified type.

While Kubernetes manifests can contain multiple object definitions in a single file (delimited with the YAML separator ‘—’), this does not currently support those files. This function expects a single object definition in the specified manifest file.

Parameters:	path – The path the manifest YAML to load. obj_type – The Kubernetes API object type that the YAML contents should be loaded into. renderer – The callable responsible for rendering the contents of the manifest file to YAML.
Returns:	A Kubernetes API object populated with the YAML contents.
Raises:	`FileNotFoundError` – The specified file was not found.

kubetest.manifest.new_object(root_type, config)[source]¶

Create a new Kubernetes API object and recursively populate it with the provided manifest configuration.

The recursive population utilizes the swagger_types/openapi_types and attribute_map members of the Kubernetes API object class to determine which config fields correspond to which input parameter, and to cast them to their expected type.

This is all based on the premise that the Python Kubernetes client will continue to be based off of an auto-generated Swagger/Openapi spec and that these fields will be available for all API objects.

Parameters:	root_type – The Kubernetes API object type that will be populated with the manifest configuration. This is expected to be known ahead of time by the caller. config – The manifest configuration for the API object.
Returns:	A Kubernetes API object recursively populated with the YAML contents.

kubetest.manifest.cast_value(value: Any, t: str) → Any[source]¶

Cast the given value to the specified type.

There are two general cases for possible casts:

A cast to a builtin type (int, str, etc.)
A cast to a Kubernetes object (V1ObjectMeta, etc)

In either case, check to see if the specified type exists in the correct type pool. If so, cast to that type, otherwise fail.

Parameters:	value – The value to cast. t – The type to cast the value to. This can be a builtin type or a Kubernetes API object type.
Returns:	The value, casted to the appropriate type.
Raises:	`ValueError` – Unable to cast the value to the specified type. `TypeError` – Unable to cast the given value to the specified type. `AttributeError` – The value is an invalid Kubernetes type.

kubetest.objects module¶

Kubetest wrappers around Kubernetes API Objects.

kubetest.plugin module¶

A pytest plugin which helps with integration testing for Kubernetes deployments.

This plugin helps with the managing of the Kubernetes cluster and provides useful test fixtures and functionality in order to interact with and test the state of the cluster.

kubetest.plugin.pytest_addoption(parser)[source]¶: Add options to pytest to configure kubetest.

See also

https://docs.pytest.org/en/latest/reference.html#_pytest.hookspec.pytest_addoption

kubetest.plugin.pytest_report_header(config)[source]¶: Augment the pytest report header with kubetest info.

See also

https://docs.pytest.org/en/latest/reference.html#_pytest.hookspec.pytest_report_header

kubetest.plugin.pytest_configure(config)[source]¶

Configure pytest with kubetest additions.

This registers the kubetest markers with pytest.

kubetest.utils module¶

Utility functions for kubetest.

kubetest.utils.new_namespace(test_name: str) → str[source]¶

Create a new namespace for the given test name.

Kubernetes namespace names follow a DNS-1123 label that consists of lower case alphanumeric characters or ‘-‘ and must start with an alphanumeric.

The test name and current timestamp are formatted to comply to this spec and appended to the ‘kubetest’ prefix.

Parameters:	test_name – The name of the test case for the namespace.
Returns:	The namespace name.

kubetest.utils.selector_string(selectors: Mapping[str, str]) → str[source]¶

Create a selector string from the given dictionary of selectors.

Parameters:	selectors – The selectors to stringify.
Returns:	The selector string for the given dictionary.

kubetest.utils.selector_kwargs(fields: Mapping[str, str] = None, labels: Mapping[str, str] = None) → Dict[str, str][source]¶

Create a dictionary of kwargs for Kubernetes object selectors.

Parameters:	fields – A mapping of fields used to restrict the returned collection of Objects to only those which match these field selectors. By default, no restricting is done. labels – A mapping of labels used to restrict the returned collection of Objects to only those which match these label selectors. By default, no restricting is done.
Returns:	A dictionary that can be used as kwargs for many Kubernetes API calls for label and field selectors.

kubetest.utils.wait_for_condition(condition: kubetest.condition.Condition, timeout: int = None, interval: Union[int, float] = 1, fail_on_api_error: bool = True) → None[source]¶

Wait for a condition to be met.

Parameters:

condition – The Condition to wait for.
timeout – The maximum time to wait, in seconds, for the condition to be met. If unspecified, this function 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 the condition.
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 specified timeout was exceeded.

Module contents¶

kubetest – a Kubernetes integration test framework in Python.