MockKubernetesApi

class safir.testing.kubernetes.MockKubernetesApi

Bases: object

Mock Kubernetes API for testing.

This object simulates (with almost everything left out) the CoreV1Api and CustomObjectApi 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.

If error_callback is set to a callable, it will be called with the method name and any arguments whenever any Kubernetes API method is called and before it takes any action. This can be used to inject exceptions for test purposes.

Notes

This class is normally not instantiated directly. Instead, call the patch_kubernetes function from a fixture to set up the mock.

Methods Summary

create_namespaced_config_map(namespace, ...)
create_namespaced_custom_object(group, ...)
create_namespaced_pod(namespace, pod)
create_namespaced_secret(namespace, secret)
delete_namespaced_config_map(name, namespace)
delete_namespaced_pod(name, namespace)
get_all_objects_for_test(kind)	Return all objects of a given kind sorted by namespace and name.
get_namespaced_custom_object(group, version, ...)
list_cluster_custom_object(group, version, ...)
list_namespaced_pod(namespace, *, field_selector)
patch_namespaced_custom_object_status(group, ...)
patch_namespaced_secret(name, namespace, body)
read_namespaced_pod(name, namespace)
read_namespaced_secret(name, namespace)
replace_namespaced_custom_object(group, ...)
replace_namespaced_secret(name, namespace, ...)

Methods Documentation

async create_namespaced_config_map(namespace: str, config_map: kubernetes_asyncio.client.models.v1_config_map.V1ConfigMap) → None

async create_namespaced_custom_object(group: str, version: str, namespace: str, plural: str, body: Dict[str, Any]) → None

async create_namespaced_pod(namespace: str, pod: kubernetes_asyncio.client.models.v1_pod.V1Pod) → None

async create_namespaced_secret(namespace: str, secret: kubernetes_asyncio.client.models.v1_secret.V1Secret) → None

async delete_namespaced_config_map(name: str, namespace: str) → kubernetes_asyncio.client.models.v1_status.V1Status

async delete_namespaced_pod(name: str, namespace: str) → kubernetes_asyncio.client.models.v1_status.V1Status

get_all_objects_for_test(kind: str) → List[Any]

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

objects – All objects of that kind found in the mock, sorted by namespace and then name.

Return type

List[typing.Any]

async get_namespaced_custom_object(group: str, version: str, namespace: str, plural: str, name: str) → Dict[str, Any]

async list_cluster_custom_object(group: str, version: str, plural: str) → Dict[str, List[Dict[str, Any]]]

async list_namespaced_pod(namespace: str, *, field_selector: str) → kubernetes_asyncio.client.models.v1_pod_list.V1PodList

async patch_namespaced_custom_object_status(group: str, version: str, namespace: str, plural: str, name: str, body: List[Dict[str, Any]]) → Dict[str, Any]

async patch_namespaced_secret(name: str, namespace: str, body: List[Dict[str, Any]]) → kubernetes_asyncio.client.models.v1_secret.V1Secret

async read_namespaced_pod(name: str, namespace: str) → kubernetes_asyncio.client.models.v1_pod.V1Pod

async read_namespaced_secret(name: str, namespace: str) → kubernetes_asyncio.client.models.v1_secret.V1Secret

async replace_namespaced_custom_object(group: str, version: str, namespace: str, plural: str, name: str, body: Dict[str, Any]) → None

async replace_namespaced_secret(name: str, namespace: str, secret: kubernetes_asyncio.client.models.v1_secret.V1Secret) → None