In this blog post, I will show you how to write End to End Tests in Kubernetes to verify the various objects in a Kubernetes cluster like Pods, Services, Nodes, Namespaces etc.
I would highly recommend you to read my previous blog about End to End Testing in Kubernetes for context setting prior to looking at the below set of tests –
End to End Testing in Kubernetes
If you want to interact with the Kubernetes API, there is no need to write your own library. You can leverage a number of available client libraries for this purpose. I am using the Python API Client for Kubernetes to interact with the cluster in Cloud.
Official Python client library for Kubernetes –
https://github.com/kubernetes-client/python
Documentation for Kubernetes API Endpoints –
https://github.com/kubernetes-client/python/tree/master/kubernetes#documentation-for-api-endpoints
The below tests will verify the pod count and service count in the ‘default’ namespace —
| def test_to_verify_pod_count_in_default_namespace(k8s_client): | |
| podcount = len(k8s_client.list_namespaced_pod(namespace='default').items) | |
| print(podcount) | |
| assert(podcount > 0) | |
| def test_to_verify_service_count_in_default_namespace(k8s_client): | |
| svccount = len(k8s_client.list_namespaced_service(namespace='default').items) | |
| print(svccount) | |
| assert(svccount > 0) | |
The below test will iterate through all the namespaces in the K8s cluster and list it down –
| def test_to_list_all_namespaces(k8s_client): | |
| ret = k8s_client.list_namespace() | |
| nscount = len(ret.items) | |
| print(nscount) | |
| assert(nscount > 0) | |
| # Iterate through all the namespaces in the K8s cluster and list it down | |
| for item in ret.items: | |
| print(item.metadata.name) | |
The below set of tests will iterate through all the endpoints, services, nodes and pods in the K8s cluster and list them down —
| def test_to_list_endpoints_for_all_namespaces(k8s_client): | |
| ret = k8s_client.list_endpoints_for_all_namespaces() | |
| epcount = len(ret.items) | |
| print(epcount) | |
| assert(epcount > 0) | |
| # Iterate through all the endpoints in the K8s cluster and list it down | |
| for item in ret.items: | |
| print("%s\t%s\t" % (item.metadata.name, item.metadata.namespace)) | |
| def test_to_list_services_for_all_namespaces(k8s_client): | |
| ret = k8s_client.list_service_for_all_namespaces() | |
| print("\n") | |
| svccount = len(ret.items) | |
| print(svccount) | |
| assert(svccount > 0) | |
| # Iterate through all the services in the K8s cluster and list down all the services and respective namespace | |
| for item in ret.items: | |
| print("%s\t%s\t" % (item.metadata.name, item.metadata.namespace)) | |
| def test_to_list_nodes_for_all_namespaces(k8s_client): | |
| ret = k8s_client.list_node() | |
| nodecount = len(ret.items) | |
| print(nodecount) | |
| assert(nodecount > 0) | |
| # Iterate through all the nodes in the K8s cluster and list it down | |
| for item in ret.items: | |
| print("%s\t" % item.metadata.name) | |
| def test_to_list_pods_for_all_namespaces(k8s_client): | |
| ret = k8s_client.list_pod_for_all_namespaces() | |
| podcount = len(ret.items) | |
| print(podcount) | |
| assert(podcount > 0) | |
| # Iterate through all the pods in the K8s cluster and list down the pod name and respective namespace | |
| for item in ret.items: | |
| print("%s\t%s\t" % (item.metadata.name, item.metadata.namespace)) | |
The below tests lists pods and services for a specific namespace —
| def test_to_list_pods_for_specific_namespace(k8s_client): | |
| ret = k8s_client.list_namespaced_pod("default") | |
| podcount = len(ret.items) | |
| print(podcount) | |
| assert(podcount > 0) | |
| # Iterate through all the pods in the default namespace and list down the pod name and respective namespace | |
| for item in ret.items: | |
| print("%s\t%s\t" % (item.metadata.name, item.metadata.namespace)) | |
| def test_to_list_services_for_specific_namespace(k8s_client): | |
| ret = k8s_client.list_namespaced_service("default") | |
| svccount = len(ret.items) | |
| print(svccount) | |
| assert(svccount > 0) | |
| # Iterate through all the services in the default namespace and list down the service name and respective namespace | |
| for item in ret.items: | |
| print("%s\t%s\t" % (item.metadata.name, item.metadata.namespace)) | |
Using the Python API Client we saw how to interact with the K8s components in the cluster hosted in AWS or Azure. By running a set of End to End Tests we can easily determine the health of the cluster. We have just scratched the surface with these E2E Tests — you can improvise on top of these tests to interact with any K8s object in the cluster and verify functionalities.
Categories: Kubernetes
CNCF Cloud Native Landscape and Trail Map 
Microservices on Kubernetes 
Kubernetes Architecture 
Istio 1.0 – The Production Ready Service Mesh 
is there any python e2e test framework like e2e(ginkgo) in golang?
LikeLike