MockKubernetesApi#

class safir.testing.kubernetes.MockKubernetesApi#

Bases: object

Mock Kubernetes API for testing.

This object simulates (with almost everything left out) the CoreV1Api, CustomObjectApi, and NetworkingV1Api client objects while keeping simple internal state. It is intended to be used as a mock inside tests.

Methods ending with _for_test are outside of the API and are intended for use by the test suite.

This mock does not enforce namespace creation before creating objects in a namespace. Creating an object in a namespace will implicitly create that namespace if it doesn’t exist. However, it will not store a V1Namespace object, so to verify that a namespace was properly created (although not the order of creation), retrieve all the objects in the namespace with get_namespace_objects_for_test and one of them will be the V1Namespace object.

Objects stored with create_* or replace_* methods are NOT copied. The object provided will be stored, so changing that object will change the object returned by subsequent API calls. Likewise, the object returned by read_* calls will be the same object stored in the mock, and changing it will change the mock’s data. (Sometimes this is the desired behavior, sometimes it isn’t; we had to pick one and this is the approach we picked.)

Most APIs do not support watches. The only current exception is list_namespaced_event.

initial_pod_phase#: String value to set the status of pods to when created. If this is set to Running (the default), a pod start event will also be generated when the pod is created.

error_callback#: If set, called with the method name and any arguments whenever any Kubernetes API method is called and before it takes any acttion. This can be used for fault injection for testing purposes.

Notes

This class is normally not instantiated directly. Instead, call the patch_kubernetes function from a fixture to set up the mock. This is also why it is configurable by setting attributes rather than constructor arguments; the individual test usually doesn’t have control of the constructor.

Methods Summary

`create_namespace`(body)	Create a namespace.
`create_namespaced_config_map`(namespace, body)	Create a `ConfigMap` object.
`create_namespaced_custom_object`(group, ...)	Create a new custom namespaced object.
`create_namespaced_event`(namespace, body)	Store a new namespaced event.
`create_namespaced_network_policy`(namespace, body)	Create a network policy object.
`create_namespaced_pod`(namespace, body)	Create a pod object.
`create_namespaced_resource_quota`(namespace, body)	Create a resource quota object.
`create_namespaced_secret`(namespace, body)	Create a secret object.
`create_namespaced_service`(namespace, body)	Create a service object.
`delete_namespace`(name)	Delete a namespace.
`delete_namespaced_config_map`(name, namespace)	Delete a `ConfigMap` object.
`delete_namespaced_pod`(name, namespace)	Delete a pod object.
`get_all_objects_for_test`(kind)	Return all objects of a given kind sorted by namespace and name.
`get_namespace_objects_for_test`(namespace)	Returns all objects in the given namespace.
`get_namespaced_custom_object`(group, version, ...)	Retrieve a namespaced custom object.
`list_cluster_custom_object`(group, version, ...)	List all custom objects in the cluster.
`list_namespace`()	List known namespaces.
`list_namespaced_event`(namespace, *[, ...])	List namespaced events.
`list_namespaced_pod`(namespace, *[, ...])	List pod objects in a namespace.
`list_node`()	List node information.
`patch_namespaced_custom_object_status`(group, ...)	Patch the status of a namespaced custom object.
`patch_namespaced_secret`(name, namespace, body)	Patch a secret object.
`read_namespace`(name)	Return the namespace object for a namespace.
`read_namespaced_pod`(name, namespace)	Read a pod object.
`read_namespaced_pod_status`(name, namespace)	Read the status of a pod.
`read_namespaced_secret`(name, namespace)	Read a secret.
`replace_namespaced_custom_object`(group, ...)	Replace a custom namespaced object.
`replace_namespaced_secret`(name, namespace, body)	Replace a secret.
`set_nodes_for_test`(nodes)	Set the node structures that will be returned by list_node.

Methods Documentation

async create_namespace(body)#

Create a namespace.

The mock doesn’t truly track namespaces since it autocreates them when an object is created in that namespace (maybe that behavior should be optional). However, this method detects conflicts and stores the V1Namespace object so that it can be verified.

Parameters:: body (V1Namespace) – Namespace to create.
Raises:: kubernetes_asyncio.client.ApiException – Raised with 409 status if the namespace already exists.
Return type:: None

async create_namespaced_config_map(namespace, body)#

Create a ConfigMap object.

Parameters:

namespace (str) – Namespace in which to store the object.
body (V1ConfigMap) – Object to store.

Raises:

kubernetes_asyncio.client.ApiException – Raised with 409 status if the object already exists.

Return type:

None

async create_namespaced_custom_object(group, version, namespace, plural, body)#

Create a new custom namespaced object.

Parameters:

group (str) – API group for this custom object.
version (str) – API version for this custom object.
namespace (str) – Namespace in which to create the object.
plural (str) – API plural for this custom object.
body (dict[str, Any]) – Custom object to create.

Raises:

kubernetes_asyncio.client.ApiException – Raised with 409 status if the object already exists.

Return type:

None

async create_namespaced_event(namespace, body)#

Store a new namespaced event.

This uses the old core event API, not the new Events API.

Parameters:

namespace (str) – Namespace of the event.
body (CoreV1Event) – New event to store.

Return type:

None

async create_namespaced_network_policy(namespace, body)#

Create a network policy object.

Parameters:

namespace (str) – Namespace in which to create the object.
body (V1NetworkPolicy) – Object to create.

Raises:

kubernetes_asyncio.client.ApiException – Raised with 409 status if the object already exists.

Return type:

None

async create_namespaced_pod(namespace, body)#

Create a pod object.

If initial_pod_phase on the mock Kubernetes object is set to Running, set the state to Running and generate a startup event. Otherwise, the status is set to whatever initial_pod_phase is set to, and no event is generated.

Parameters:

namespace (str) – Namespace in which to create the pod.
body (V1Pod) – Pod specification.

Raises:

kubernetes_asyncio.client.ApiException – Raised with 409 status if the pod already exists.

Return type:

None

async create_namespaced_resource_quota(namespace, body)#

Create a resource quota object.

Parameters:

namespace (str) – Namespace in which to create the object.
body (V1ResourceQuota) – Object to create.

Raises:

kubernetes_asyncio.client.ApiException – Raised with 409 status if the object already exists.

Return type:

None

async create_namespaced_secret(namespace, body)#

Create a secret object.

Parameters:

namespace (str) – Namespace in which to create the object.
body (V1Secret) – Object to create.

Raises:

kubernetes_asyncio.client.ApiException – Raised with 409 status if the object already exists.

Return type:

None

async create_namespaced_service(namespace, body)#

Create a service object.

Parameters:

namespace (str) – Namespace in which to create the object.
body (V1Service) – Object to create.

Raises:

kubernetes_asyncio.client.ApiException – Raised with 409 status if the object already exists.

Return type:

None

async delete_namespace(name)#

Delete a namespace.

This also immediately removes all objects in the namespace.

Parameters:: name (str) – Namespace to delete.
Raises:: kubernetes_asyncio.client.ApiException – Raised with 404 status if the namespace does not exist.
Return type:: None

async delete_namespaced_config_map(name, namespace)#

Delete a ConfigMap object.

Parameters:

name (str) – Name of object.
namespace (str) – Namespace of object.

Returns:

Success status if object was deleted.

Return type:

kubernetes_asyncio.client.V1Status

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the object does not exist.

async delete_namespaced_pod(name, namespace)#

Delete a pod object.

Parameters:

name (str) – Name of pod to delete.
namespace (str) – Namespace of pod to delete.

Returns:

Success status.

Return type:

kubernetes_asyncio.client.V1Status

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the pod was not found.

get_all_objects_for_test(kind)#

Return all objects of a given kind sorted by namespace and name.

Parameters:: kind (str) – The Kubernetes kind, such as Secret or Pod. This is case-sensitive.
Returns:: All objects of that kind found in the mock, sorted by namespace and then name.
Return type:: list of Any

get_namespace_objects_for_test(namespace)#

Returns all objects in the given namespace.

Parameters:: namespace (str) – Name of the namespace.
Returns:: All objects found in that namespace, sorted by kind and then name. Due to how objects are stored in the mock, we can’t distinguish between a missing namespace and a namespace with no objects. In both cases, the empty list is returned.
Return type:: list of Any

async get_namespaced_custom_object(group, version, namespace, plural, name)#

Retrieve a namespaced custom object.

Parameters:

group (str) – API group for this custom object.
version (str) – API version for this custom object.
namespace (str) – Namespace in which to create the object.
plural (str) – API plural for this custom object.
name (str) – Name of the object to retrieve.

Returns:

Body of the custom object.

Return type:

dict of Any

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the object does not exist.

async list_cluster_custom_object(group, version, plural)#

List all custom objects in the cluster.

Parameters:

group (str) – API group for this custom object.
version (str) – API version for this custom object.
plural (str) – API plural for this custom object.

Returns:

Dictionary with one key, items, whose value is a list of all the specified custom objects stored in the cluster.

Return type:

dict

async list_namespace()#

List known namespaces.

Returns:: All namespaces, whether implicitly created or not. These will be the actual V1Namespace objects if one was stored, otherwise synthesized namespace objects.
Return type:: kubernetes_asyncio.client.V1NamespaceList

async list_namespaced_event(namespace, *, field_selector=None, resource_version='0', timeout_seconds=None, watch=False, _preload_content=True, _request_timeout=None)#

List namespaced events.

This uses the old core event API, not the new Events API. It does support watches.

Parameters:

namespace (str) – Namespace to watch for events.
field_selector (Optional[str], default: None) – Which events to retrieve when performing a watch. Currently, this is ignored.
resource_version (str, default: '0') – Where to start in the event stream when performing a watch.
timeout_seconds (Optional[int], default: None) – How long to return events for before exiting when performing a watch.
watch (bool, default: False) – Whether to act as a watch.
_preload_content (bool, default: True) – Verified to be False when performing a watch.
_request_timeout (Optional[int], default: None) – Ignored, accepted for compatibility with the watch API.

Returns:

List of events, when not called as a watch. If called as a watch, returns a mock aiohttp.Response with a readline method that yields the events.

Return type:

kubernetes_asyncio.client.CoreV1EventList or unittest.mock.Mock

async list_namespaced_pod(namespace, *, field_selector=None)#

List pod objects in a namespace.

Parameters:

namespace (str) – Namespace of pods to list.
field_selector (Optional[str], default: None) – Only metadata.name=... is supported. It is parsed to find the pod name and only pods matching that name will be returned.

Returns:

List of pods in that namespace.

Return type:

kubernetes_asyncio.client.V1PodList

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the namespace does not exist.
AssertionError – Some other field_selector was provided.

async list_node()#

List node information.

Returns:: The node information previouslyl stored with set_nodes_for_test, if any.
Return type:: kubernetes_asyncio.client.V1NodeList

async patch_namespaced_custom_object_status(group, version, namespace, plural, name, body)#

Patch the status of a namespaced custom object.

Parameters:

group (str) – API group for this custom object.
version (str) – API version for this custom object.
namespace (str) – Namespace in which to create the object.
plural (str) – API plural for this custom object.
name (str) – Name of the object to retrieve.
body (list[dict[str, Any]]) – Body of the patch. The only patch supported is one with op of replace and path of /status.

Returns:

Modified body of the custom object.

Return type:

dict of Any

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the object does not exist.
AssertionError – Raised if any other type of patch is provided.

async patch_namespaced_secret(name, namespace, body)#

Patch a secret object.

Parameters:

name (str) – Name of secret object.
namespace (str) – Namespace of secret object.
body (list[dict[str, Any]]) – Patches to apply. Only patches with op of replace are supported, and only with path of either /metadata/annotations or /metadata/labels.

Returns:

Patched secret object.

Return type:

kubernetes_asyncio.client.V1Secret

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the secret does not exist.
AssertionError – Raised if any other type of patch was provided.

async read_namespace(name)#

Return the namespace object for a namespace.

Parameters:: name (str) – Name of namespace to retrieve.
Returns:: Corresponding namespace object. If create_namespace has been called, will return the stored object. Otherwise, returns a synthesized V1Namespace object if the namespace has been implicitly created.
Return type:: kubernetes_asyncio.client.V1Namespace
Raises:: kubernetes_asyncio.client.ApiException – Raised with 404 status if the namespace does not exist.

async read_namespaced_pod(name, namespace)#

Read a pod object.

Parameters:

name (str) – Name of the pod.
namespace (str) – Namespace of the pod.

Returns:

Pod object.

Return type:

kubernetes_asyncio.client.V1Pod

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the pod was not found.

async read_namespaced_pod_status(name, namespace)#

Read the status of a pod.

Parameters:

name (str) – Name of the pod.
namespace (str) – Namespace of the pod.

Returns:

Pod object. The Kubernetes API returns a V1Pod rather than, as expected, a V1PodStatus. Presumably the acutal API populates only the status portion, but we return the whole pod for testing purposes since it shouldn’t matter.

Return type:

kubernetes_asyncio.client.V1Pod

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the pod was not found.

async read_namespaced_secret(name, namespace)#

Read a secret.

Parameters:

name (str) – Name of secret.
namespace (str) – Namespace of secret.

Returns:

Requested secret.

Return type:

kubernetes_asyncio.client.V1Secret

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the secret does not exist.

async replace_namespaced_custom_object(group, version, namespace, plural, name, body)#

Replace a custom namespaced object.

Parameters:

group (str) – API group for this custom object.
version (str) – API version for this custom object.
namespace (str) – Namespace in which to create the object.
plural (str) – API plural for this custom object.
body (dict[str, Any]) – New contents of custom object.
name (str) –

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 if the object does not exist.

Return type:

None

async replace_namespaced_secret(name, namespace, body)#

Replace a secret.

Parameters:

name (str) – Name of secret.
namespace (str) – Namespace of secret.
body (V1Secret) – New body of secret.

Raises:

kubernetes_asyncio.client.ApiException – Raised with 404 status if the secret does not exist.

Return type:

None

set_nodes_for_test(nodes)#

Set the node structures that will be returned by list_node.

Parameters:: nodes (list[V1Node]) – New node list to return.
Return type:: None