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 —
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 –
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 —
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 —
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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
is there any python e2e test framework like e2e(ginkgo) in golang?
LikeLike